kafka

whatsmars-mq/whatsmars-mq-kafka-server/README.md

-本模块为用代码启动 kafka broker
+本模块为用代码启动 kafka broker (先启动zookeeper)
 
 研究kafka的源码有必要了解下scala语言<br>
 https://github.com/javahongxi/scala-examples
\ No newline at end of file

whatsmars-mq/whatsmars-mq-kafka/README.md

@@ -2,7 +2,7 @@
 
 1. 启动zk org.hongxi.whatsmars.zk.ZKStartup
 1. 启动kafka org.hongxi.whatsmars.kafka.KafkaStartup
-1. 启动client org.hongxi.whatsmars.kafka.KafkaApplication
+1. 启动client org.hongxi.whatsmars.kafka.boot.KafkaApplication
 
 linux:<br>
 启动zk ./zookeeper-server-start.sh ../config/zookeeper.properties<br>

whatsmars-mq/whatsmars-mq-kafka/src/main/java/org/hongxi/whatsmars/kafka/Consumer.java → whatsmars-mq/whatsmars-mq-kafka/src/main/java/org/hongxi/whatsmars/kafka/boot/Consumer.java

-package org.hongxi.whatsmars.kafka;
+package org.hongxi.whatsmars.kafka.boot;
 
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;

whatsmars-mq/whatsmars-mq-kafka/src/main/java/org/hongxi/whatsmars/kafka/KafkaApplication.java → whatsmars-mq/whatsmars-mq-kafka/src/main/java/org/hongxi/whatsmars/kafka/boot/KafkaApplication.java

-package org.hongxi.whatsmars.kafka;
+package org.hongxi.whatsmars.kafka.boot;
 
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;

whatsmars-mq/whatsmars-mq-kafka-server/src/main/java/org/hongxi/whatsmars/kafka/examples/Consumer.java → whatsmars-mq/whatsmars-mq-kafka/src/main/java/org/hongxi/whatsmars/kafka/quickstart/Consumer.java

-package org.hongxi.whatsmars.kafka.examples;
+package org.hongxi.whatsmars.kafka.quickstart;
 
-import kafka.utils.ShutdownableThread;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -10,12 +9,11 @@ import java.time.Duration;
 import java.util.Collections;
 import java.util.Properties;
 
-public class Consumer extends ShutdownableThread {
+public class Consumer extends Thread {
     private final KafkaConsumer<Integer, String> consumer;
     private final String topic;
 
     public Consumer(String topic) {
-        super("KafkaConsumerExample", false);
         Properties props = new Properties();
         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT);
         props.put(ConsumerConfig.GROUP_ID_CONFIG, "DemoConsumer");
@@ -30,21 +28,13 @@ public class Consumer extends ShutdownableThread {
     }
 
     @Override
-    public void doWork() {
+    public void run() {
         consumer.subscribe(Collections.singletonList(this.topic));
-        ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
-        for (ConsumerRecord<Integer, String> record : records) {
-            System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
+        while (true) {
+            ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
+            for (ConsumerRecord<Integer, String> record : records) {
+                System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
+            }
         }
     }
-
-    @Override
-    public String name() {
-        return null;
-    }
-
-    @Override
-    public boolean isInterruptible() {
-        return false;
-    }
 }

whatsmars-mq/whatsmars-mq-kafka-server/src/main/java/org/hongxi/whatsmars/kafka/examples/DemoCallBack.java → whatsmars-mq/whatsmars-mq-kafka/src/main/java/org/hongxi/whatsmars/kafka/quickstart/DemoCallBack.java

-package org.hongxi.whatsmars.kafka.examples;
+package org.hongxi.whatsmars.kafka.quickstart;
 
 import org.apache.kafka.clients.producer.Callback;
 import org.apache.kafka.clients.producer.RecordMetadata;

whatsmars-mq/whatsmars-mq-kafka-server/src/main/java/org/hongxi/whatsmars/kafka/examples/KafkaConsumerProducerDemo.java → whatsmars-mq/whatsmars-mq-kafka/src/main/java/org/hongxi/whatsmars/kafka/quickstart/KafkaConsumerProducerDemo.java

-package org.hongxi.whatsmars.kafka.examples;
+package org.hongxi.whatsmars.kafka.quickstart;
 
 public class KafkaConsumerProducerDemo {
+
     public static void main(String[] args) {
         boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync");
         Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync);

whatsmars-mq/whatsmars-mq-kafka-server/src/main/java/org/hongxi/whatsmars/kafka/examples/KafkaProperties.java → whatsmars-mq/whatsmars-mq-kafka/src/main/java/org/hongxi/whatsmars/kafka/quickstart/KafkaProperties.java

-package org.hongxi.whatsmars.kafka.examples;
+package org.hongxi.whatsmars.kafka.quickstart;
 
 public class KafkaProperties {
     public static final String TOPIC = "topic1";

whatsmars-mq/whatsmars-mq-kafka-server/src/main/java/org/hongxi/whatsmars/kafka/examples/Producer.java → whatsmars-mq/whatsmars-mq-kafka/src/main/java/org/hongxi/whatsmars/kafka/quickstart/Producer.java

-package org.hongxi.whatsmars.kafka.examples;
+package org.hongxi.whatsmars.kafka.quickstart;
 
 import org.apache.kafka.clients.producer.*;
 import org.apache.kafka.common.serialization.IntegerSerializer;