增加 kafka 集成 springboot 的示例

lab-03/pom.xml

@@ -3,15 +3,45 @@
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
-        <artifactId>labs-parent</artifactId>
-        <groupId>cn.iocoder.springboot.labs</groupId>
-        <version>1.0-SNAPSHOT</version>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>1.5.9.RELEASE</version>
+        <relativePath/> <!-- lookup parent from repository -->
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>lab-03</artifactId>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>8</source>
+                    <target>8</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
 
     <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>
@@ -22,6 +52,36 @@
             <artifactId>fastjson</artifactId>
             <version>1.2.55</version>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-kafka</artifactId>
+            <version>3.1.0.RELEASE</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+            <version>1.1.1.RELEASE</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+            <version>2.8.2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
     </dependencies>
 
 </project>

lab-03/src/main/java/cn/iocoder/springboot/labs/lab03/KafkaApplication.java

@@ -0,0 +1,34 @@
+package cn.iocoder.springboot.labs.lab03;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.ConfigurableApplicationContext;
+
+@SpringBootApplication
+public class KafkaApplication {
+
+    public static void main(String[] args) {
+
+        ConfigurableApplicationContext context = SpringApplication.run(KafkaApplication.class, args);
+
+//        KafkaSender sender = context.getBean(KafkaSender.class);
+//
+//        for (int i = 0; i < 3; i++) {
+//            //调用消息发送类中的消息发送方法
+//            sender.send();
+//
+//            try {
+//                Thread.sleep(3000);
+//            } catch (InterruptedException e) {
+//                e.printStackTrace();
+//            }
+//        }
+
+        try {
+            Thread.sleep(300000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+}

lab-03/src/main/java/cn/iocoder/springboot/labs/lab03/KafkaReceiver.java

@@ -0,0 +1,61 @@
+package cn.iocoder.springboot.labs.lab03;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.util.Optional;
+
+@Component
+public class KafkaReceiver {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @KafkaListener(topics = {"test"})
+    public void listen(ConsumerRecord<?, ?> record) {
+
+        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
+
+        if (kafkaMessage.isPresent()) {
+
+            Object message = kafkaMessage.get();
+
+            logger.info("[test{}]----------------- record =" + record, Thread.currentThread().getId());
+            logger.info("[test{}]------------------ message =" + message, Thread.currentThread().getId());
+        }
+
+    }
+
+    @KafkaListener(topics = {"yunai"})
+    public void listen02(ConsumerRecord<?, ?> record) {
+
+        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
+
+        if (kafkaMessage.isPresent()) {
+
+            Object message = kafkaMessage.get();
+
+            logger.info("[yunai{}]----------------- record =" + record, Thread.currentThread().getId());
+            logger.info("[yunai{}]------------------ message =" + message, Thread.currentThread().getId());
+        }
+
+    }
+
+    @KafkaListener(topics = {"afei"})
+    public void listen03(ConsumerRecord<?, ?> record) {
+
+        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
+
+        if (kafkaMessage.isPresent()) {
+
+            Object message = kafkaMessage.get();
+
+            logger.info("[afei{}]----------------- record =" + record, Thread.currentThread().getId());
+            logger.info("[afei{}]------------------ message =" + message, Thread.currentThread().getId());
+        }
+
+    }
+
+}

lab-03/src/main/java/cn/iocoder/springboot/labs/lab03/pojo/Message.java

@@ -0,0 +1,16 @@
+package cn.iocoder.springboot.labs.lab03.pojo;
+
+import lombok.Data;
+
+import java.util.Date;
+
+@Data
+public class Message {
+
+    private Long id;    //id
+
+    private String msg; //消息
+
+    private Date sendTime;  //时间戳
+
+}

lab-03/src/main/resources/application.properties

@@ -0,0 +1,29 @@
+#============== kafka ===================
+# 指定kafka 代理地址，可以多个
+spring.kafka.bootstrap-servers=127.0.0.1:9092
+
+#=============== provider  =======================
+
+spring.kafka.producer.retries=0
+# 每次批量发送消息的数量
+spring.kafka.producer.batch-size=16384
+spring.kafka.producer.buffer-memory=33554432
+
+# 指定消息key和消息体的编解码方式
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
+
+#=============== consumer  =======================
+# 指定默认消费者group id
+spring.kafka.consumer.group-id=test-consumer-group6
+
+spring.kafka.listener.concurrency=10
+
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.enable-auto-commit=true
+spring.kafka.consumer.auto-commit-interval=100
+enable.auto.commit=false
+
+# 指定消息key和消息体的编解码方式
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer