update

Author: kimmking

09mq/kafka-demo/src/main/java/io/kimmking/javacourse/kafka/KafkaConsumerDemo.java

@@ -0,0 +1,17 @@
+packageio.kimmking.javacourse.kafka;
+
+importio.kimmking.javacourse.kafka.kimmking.ConsumerImpl;
+importio.kimmking.javacourse.kafka.kimmking.ProducerImpl;
+
+publicclassKafkaConsumerDemo{
+
+publicstaticvoidmain(String[] args){
+testConsumer();
+ }
+
+privatestaticvoidtestConsumer(){
+ConsumerImplconsumer = newConsumerImpl();
+consumer.consumeOrder();
+
+ }
+}

…kimmking/javacourse/kafka/KafkaDemo.java …/javacourse/kafka/KafkaProducerDemo.java09mq/kafka-demo/src/main/java/io/kimmking/javacourse/kafka/KafkaDemo.java renamed to 09mq/kafka-demo/src/main/java/io/kimmking/javacourse/kafka/KafkaProducerDemo.java 09mq/kafka-demo/src/main/java/io/kimmking/javacourse/kafka/KafkaDemo.java renamed to 09mq/kafka-demo/src/main/java/io/kimmking/javacourse/kafka/KafkaProducerDemo.java

@@ -3,17 +3,10 @@
 importio.kimmking.javacourse.kafka.kimmking.ConsumerImpl;
 importio.kimmking.javacourse.kafka.kimmking.ProducerImpl;
 
-publicclassKafkaDemo{
+publicclassKafkaProducerDemo{
 
 publicstaticvoidmain(String[] args){
-//testProducer();
-testConsumer();
- }
-
-privatestaticvoidtestConsumer(){
-ConsumerImplconsumer = newConsumerImpl();
-consumer.consumeOrder();
-
+testProducer();
  }
 
 privatestaticvoidtestProducer(){
@@ -22,7 +15,5 @@ private static void testProducer(){
 producer.send(newOrder(1000L + i,System.currentTimeMillis(),"USD2CNY", 6.5d));
 producer.send(newOrder(2000L + i,System.currentTimeMillis(),"USD2CNY", 6.51d));
  }
-
  }
-
 }

09mq/kafka-demo/src/main/java/io/kimmking/javacourse/kafka/kimmking/ConsumerImpl.java

@@ -21,7 +21,7 @@ public ConsumerImpl(){
 // properties.put("enable.auto.commit", false);
 // properties.put("isolation.level", "read_committed");
 // properties.put("auto.offset.reset", "latest");
-properties.put("group.id", "java0-kimmking");
+properties.put("group.id", "java1-kimmking");
 properties.put("bootstrap.servers", "localhost:9092");
 properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
@@ -58,7 +58,7 @@ public void consumeOrder(){
 e.printStackTrace();
  } finally{
 try{
-consumer.commitSync(currentOffsets);
+consumer.commitSync();//currentOffsets);
  } catch (Exceptione){
 consumer.close();
  }

09mq/kafka-demo/src/main/java/io/kimmking/javacourse/kafka/kimmking/ProducerImpl.java

@@ -16,37 +16,36 @@ public class ProducerImpl implements Producer{
 
 publicProducerImpl(){
 properties = newProperties();
-properties.put("queue.enqueue.timeout.ms", -1);
-properties.put("enable.idempotence", true);
 properties.put("bootstrap.servers", "localhost:9092");
-properties.put("transactional.id", "transactional_1");
-properties.put("acks", "all");
-properties.put("retries", "3");
-properties.put("max.in.flight.requests.per.connection", 1);
+// properties.put("queue.enqueue.timeout.ms", -1);
+// properties.put("enable.idempotence", true);
+// properties.put("transactional.id", "transactional_1");
+// properties.put("acks", "all");
+// properties.put("retries", "3");
+// properties.put("max.in.flight.requests.per.connection", 1);
 properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 producer = newKafkaProducer<String, String>(properties);
-producer.initTransactions();
+//producer.initTransactions();
  }
 
 @Override
 publicvoidsend(Orderorder){
-Objectjson = JSON.toJSON(order);
 try{
-producer.beginTransaction();
-ProducerRecordrecord = newProducerRecord(topic, order.getId().toString(), json.toString());
+//producer.beginTransaction();
+ProducerRecordrecord = newProducerRecord(topic, order.getId().toString(), JSON.toJSONString(order));
 producer.send(record, (metadata, exception) ->{
-if (exception != null){
-producer.abortTransaction();
-thrownewKafkaException(exception.getMessage() + " , data: " + record);
- }
+// if (exception != null){
+// producer.abortTransaction();
+// throw new KafkaException(exception.getMessage() + " , data: " + record);
+// }
  });
-producer.commitTransaction();
+//producer.commitTransaction();
 
  } catch (Throwablee){
-producer.abortTransaction();
+//producer.abortTransaction();
  }
-System.out.println("************" + json + "************");
+//System.out.println("************" + json + "************");
  }
 
 @Override