springboot整合kafka

安装了kafak2.3.0的前提下，并且已经打开服务。安装教程：https://mp.csdn.net/postedit/102837603

pom引入依赖：

<!--引入kafak和spring整合的jar-->         <dependency>             <groupId>org.springframework.kafka</groupId>             <artifactId>spring-kafka</artifactId>             <version>2.3.0.RELEASE</version>         </dependency>

pom文件如下：

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" 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"> <modelVersion>4.0.0</modelVersion> <groupId>com.freshbin</groupId> <artifactId>spring-boot-01</artifactId> <version>1.0-SNAPSHOT</version> <!-- Inherit defaults from Spring Boot --> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.2.0.RELEASE</version> </parent> <!-- Add typical dependencies for a web application --> <dependencies> <!--引入kafak和spring整合的jar--> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>2.3.0.RELEASE</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.49</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>commons-beanutils</groupId> <artifactId>commons-beanutils</artifactId> <version>1.8.0</version> </dependency> <dependency> <groupId>commons-collections</groupId> <artifactId>commons-collections</artifactId> <version>3.2.1</version> </dependency> <dependency> <groupId>commons-lang</groupId> <artifactId>commons-lang</artifactId> <version>2.4</version> </dependency> <dependency> <groupId>commons-logging</groupId> <artifactId>commons-logging</artifactId> <version>1.1.1</version> </dependency> <dependency> <groupId>net.sf.ezmorph</groupId> <artifactId>ezmorph</artifactId> <version>1.0.6</version> </dependency> <dependency> <groupId>net.sf.json-lib</groupId> <artifactId>json-lib</artifactId> <classifier>jdk15</classifier> <version>2.2.3</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.poi/poi --> <dependency> <groupId>org.apache.poi</groupId> <artifactId>poi</artifactId> <version>4.0.1</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.poi/poi-ooxml --> <dependency> <groupId>org.apache.poi</groupId> <artifactId>poi-ooxml</artifactId> <version>4.0.1</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.poi/poi-scratchpad --> <dependency> <groupId>org.apache.poi</groupId> <artifactId>poi-scratchpad</artifactId> <version>4.0.1</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.poi/poi-excelant --> <dependency> <groupId>org.apache.poi</groupId> <artifactId>poi-excelant</artifactId> <version>4.0.1</version> </dependency> </dependencies> <!-- Package as an executable jar --> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build> <!-- Add Spring repositories --> <!-- (you don't need this if you are using a .RELEASE version) --> <repositories> <repository> <id>spring-snapshots</id> <url>https://repo.spring.io/snapshot</url> <snapshots><enabled>true</enabled></snapshots> </repository> <repository> <id>spring-milestones</id> <url>https://repo.spring.io/milestone</url> </repository> </repositories> <pluginRepositories> <pluginRepository> <id>spring-snapshots</id> <url>https://repo.spring.io/snapshot</url> </pluginRepository> <pluginRepository> <id>spring-milestones</id> <url>https://repo.spring.io/milestone</url> </pluginRepository> </pluginRepositories> </project>

不知道为什么，pom文件会有个红×，但是不影响项目运行。

发送消息的数据类UserLog：

package com.freshbin.example01.kafka.bean; import lombok.Data; import lombok.experimental.Accessors; /** * @Author freshbin * @Description 定义用户发送的日志数据 * @Date 2019-10-31 15:42:14 */ @Data @Accessors(chain = true) public class UserLog { private String username; private String userid; private String state; }

生产者类UserLogProducer：

package com.freshbin.example01.kafka.producer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Component; import com.alibaba.fastjson.JSON; import com.freshbin.example01.kafka.bean.UserLog; /** * @Author freshbin * @Description * @Date 2019-10-31 15:46:16 */ @Component public class UserLogProducer { @Autowired private KafkaTemplate kafkaTemplate; /** * 发送数据 * @param userid */ public void sendLog(String userid){ UserLog userLog = new UserLog(); userLog.setUsername("freshbin").setUserid(userid).setState("0"); System.out.println("发送用户日志数据:" + userLog); kafkaTemplate.send("user-log", JSON.toJSONString(userLog)); } }

消费者类UserLogConsumer：

package com.freshbin.example01.kafka.comsumer; import java.util.Optional; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; import lombok.extern.slf4j.Slf4j; /** * @Author freshbin * @Description * @Date 2019-10-31 15:49:23 */ @Component @Slf4j public class UserLogConsumer { @KafkaListener(topics = { "user-log" }) public void consumer(ConsumerRecord<?, ?> consumerRecord) { // 判断是否为null Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value()); log.info(">>>>>>>>>> record =" + kafkaMessage); if (kafkaMessage.isPresent()) { // 得到Optional实例中的值 Object message = kafkaMessage.get(); System.out.println("消费消息:" + message); } } }

启动主类KafkaApplication，直接在init方法中发送消息：

package com.freshbin.example01.kafka; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; import com.freshbin.example01.kafka.producer.UserLogProducer; import javax.annotation.PostConstruct; /** * @Author freshbin * @Description * @Date 2019-10-31 15:46:07 */ @SpringBootApplication(exclude = {DataSourceAutoConfiguration.class}) public class KafkaApplication { @Autowired private UserLogProducer kafkaSender; @PostConstruct public void init(){ for (int i = 0; i < 10; i++) { //调用消息发送类中的消息发送方法 kafkaSender.sendLog(String.valueOf(i)); System.out.println("已经发送了一个消息：" + i); } } public static void main(String[] args) { SpringApplication.run(KafkaApplication.class, args); } }

application.properties文件配置如下：

spring.application.name=kafka-user server.port=8080 #============== kafka =================== # 指定kafka 代理地址，可以多个 spring.kafka.bootstrap-servers=localhost: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=user-log-group spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.enable-auto-commit=true spring.kafka.consumer.auto-commit-interval=100 # 指定消息key和消息体的编解码方式 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

启动之后控制台就会打印相应的信息：

 

同时zk和kafka的dos窗口也会打印一些东西，不过没有仔细看打印了什么，反正也看不懂的。

虽然springboot整合了kafka，感觉demo还挺简单的，但是还不知道如何应用到真实的项目中。