Springboot与Kafka整合-yellowcong

- 日理万妓 2022-03-08 06:40 247阅读 0赞

> 在springboot上，实现kafka得消费，非常得简单，比在原始得springmvc框架上做，简单方便不少，实现kafka消费主要有以下三个步骤：1.导入依赖 2. 配置生产者，并写生产逻辑，3.配置消费者，加上一个`@KafkaListener(topics= {"${kafka.topic.name}"})`注解，即可完成消息消费得监听了。

## 项目地址 ##

https://gitee.com/yellowcong/springboot-demo/tree/master/springboot-kafka

## 1环境搭建 ##

### 1.1 导入依赖 ###

导入kafka得pom依赖文件，spring-kafka 这个是关于kafka和springboot整合得依赖。

<dependency>
 	<groupId>org.springframework.kafka</groupId>
 	<artifactId>spring-kafka</artifactId>
 </dependency>

完整得pom.xml配置如下

<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>yellowcong</groupId>
 	<artifactId>springboot-kafka</artifactId>
 	<version>0.0.1-SNAPSHOT</version>
 	<packaging>jar</packaging>
 	<name>springboot-kafka</name>
 
 	<url>http://maven.apache.org</url>
 	
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
 		<version>2.0.2.RELEASE</version>
 	</parent>
 
 	<properties>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<log4j.version>1.2.17</log4j.version>
 	</properties>
 
 	<dependencies>
 		
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-web</artifactId>
 		</dependency>
 		
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
 			
 		
 		 <dependency>
 			<groupId>org.springframework.kafka</groupId>
 			<artifactId>spring-kafka</artifactId>
 		</dependency>
 		
 		<dependency>
 			<groupId>com.alibaba</groupId>
 			<artifactId>fastjson</artifactId>
 			<version>1.2.47</version>
 		</dependency>
 
 		
 		<dependency>
 			<groupId>log4j</groupId>
 			<artifactId>log4j</artifactId>
 			<version>${log4j.version}</version>
 		</dependency>
 		
 	</dependencies>
 
 	<build>
 		<plugins>
 			
 			<plugin>
 				<groupId>org.springframework.boot</groupId>
 				<artifactId>spring-boot-maven-plugin</artifactId>
 				<configuration>
 					<mainClass>com.yellowcong.App</mainClass>
 				</configuration>
 				<executions>
 					<execution>
 						<goals>
 							<goal>repackage</goal>
 						</goals>
 					</execution>
 				</executions>
 				
 				<dependencies>
 					<dependency>
 						<groupId>org.springframework</groupId>
 						<artifactId>springloaded</artifactId>
 						<version>1.2.4.RELEASE</version>
 					</dependency>
 				</dependencies>
 			</plugin>
 			
 			<plugin>
 				<groupId>org.apache.maven.plugins</groupId>
 				<artifactId>maven-compiler-plugin</artifactId>
 				<configuration>
 					<source>1.7</source>
 					<target>1.7</target>
 				</configuration>
 			</plugin>
 			
 		</plugins>
 	</build>
 </project>

### 1.2 配置application.properties ###

配置kafka得生产者和消费得信息

#spring.profiles.active=dev
    #配置日志信息
    logging.config=classpath:log4j2.xml
    #============== kafka ===================
    # 指定kafka 代理地址，可以多个
    #spring.kafka.bootstrap-servers=192.168.100.10:9092
    
    spring.kafka.bootstrap-servers=192.168.10.18:6667,192.168.10.19:6667,192.168.10.20:6667,192.168.10.3:6667,192.168.10.4:6667
    #=============== 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-group
    #实时生产，实时消费，不会从头开始消费
    #earliest
    #当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 
    #latest (生产使用)
    #当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 
    #none 
    # topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
    spring.kafka.consumer.auto-offset-reset=earliest 
    #是否自动提交
    spring.kafka.consumer.enable-auto-commit=true
    
    #连接超时
    kafka.consumer.session.timeout=20000
    
    spring.kafka.consumer.auto-commit-interval=100
    
    #消费线程数
    kafka.consumer.concurrency=10
    
    # 指定消息key和消息体的编解码方式
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    
    
    # 写入得topic 
    kafka.topic.name=ba_spam_content

## 2. 生产者 ##

### 2.1 消息pojo ###

对发送得消息进行简单得包装

package com.yellowcong.entity;
 
 import java.util.Date;
 
 /** * 代码创建: yellowcong * 创建日期: 2019年3月14日 * 功能描述: */
 public class Message { 
 	private Long id; // id
 
 	private String msg; // 消息
 
 	private Date sendTime; // 时间戳
 
 	public Long getId() { 
 		return id;
 	}
 
 	public void setId(Long id) { 
 		this.id = id;
 	}
 
 	public String getMsg() { 
 		return msg;
 	}
 
 	public void setMsg(String msg) { 
 		this.msg = msg;
 	}
 
 	public Date getSendTime() { 
 		return sendTime;
 	}
 
 	public void setSendTime(Date sendTime) { 
 		this.sendTime = sendTime;
 	}
 
 }

### 2.2 消息生产者 ###

这个地方得生产者是异步得发送，不影响正常得业务处理。

package com.yellowcong.kafka;
 
 import java.util.Date;
 
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Component;
 
 import com.alibaba.fastjson.JSON;
 import com.yellowcong.entity.Message;
 
 /** * 代码创建: yellowcong * 创建日期: 2019年3月14日 * 功能描述: */
 @Component
 public class ProducerService { 
 
 	@Autowired
 	private KafkaTemplate<String, String> kafkaTemplate;
 	
 	@Value("${kafka.topic.name}")
 	private String topicNm;
 	
 	/** * 代码创建: yellowcong * 创建日期: 2019年3月14日 * 功能描述: 发送消息 * @param msgStr */
 	public void send(String msgStr) { 
 		
 		Message msg = new Message();
 		msg.setId(System.currentTimeMillis());
 		msg.setMsg(msgStr);
 		msg.setSendTime(new Date());
 		
 		//发送消费者到目标topic
 		this.kafkaTemplate.send(topicNm, JSON.toJSONString(msg));
 		System.out.println(msgStr);
 	}
 }

## 3 消费者 ##

### 3.1 添加ServletContextListener ###

这个监听器，用于监听kafka得消费情况,增加注解@KafkaListener，设定topic信息

package com.yellowcong.kafka;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
 import com.yellowcong.utils.LogUtils;
 
 import java.util.Optional;
 /** * 代码创建: yellowcong * 创建日期: 2019年3月14日 * 功能描述: */
 @Component
 public class ConsumerService { 
 
 	
 	@KafkaListener(topics= { "${kafka.topic.name}"})
 	public void listeener(ConsumerRecord<?, ?> record) { 
 		
 		Optional<?> msg = Optional.ofNullable(record.value());
 		//判断你消息是否存在
 		if(msg.isPresent()) { 
 			Object obj = msg.get();
 			
 			LogUtils.INFO.info(Thread.currentThread().getName()+"---"+obj.toString());
 			LogUtils.INFO.info(record.toString());
 		}
 	}
 }

## 常见问题 ##

### 1. Connection to node 0 could not be established. Broker may not be available. ###

导致这个问题得原因并不是 kafka没有开端口得问题，是由于配置得PLAINTEXT 有问题，不是设定得是 本机ip，而是`0.0.0.0`导致的。

listeners=PLAINTEXT://192.168.100.10:9092

## 参考文章 ##

[https://www.cnblogs.com/niechen/p/8672623.html][https_www.cnblogs.com_niechen_p_8672623.html]  
[https://www.cnblogs.com/likehua/p/3999538.html][https_www.cnblogs.com_likehua_p_3999538.html]  
[https://blog.csdn.net/tzs\_1041218129/article/details/78988439][https_blog.csdn.net_tzs_1041218129_article_details_78988439]

[https_www.cnblogs.com_niechen_p_8672623.html]: https://www.cnblogs.com/niechen/p/8672623.html
[https_www.cnblogs.com_likehua_p_3999538.html]: https://www.cnblogs.com/likehua/p/3999538.html
[https_blog.csdn.net_tzs_1041218129_article_details_78988439]: https://blog.csdn.net/tzs_1041218129/article/details/78988439