Spring Boot 整合之前的内容
关于版本
项目地址
因为涉及的代码较多，所以并不会贴出所有代码。本篇文章涉及的源码下载地址：springboot-samples
kafka消息请求的原理
这里简单的介绍下kafka发送消息的流程。（更详细的原理在网络上已经有足够的文章，这里不再赘述）。另外关于如何安装kafka目前也存在相当多的文章。
kafka相关名词
brokerproducerconsumertopicpartitionproducer
Producer 作为消息的生产者，负责生产数据推入broker中。
consumer
Producer 作为消息的消费者，从broker中的某个topic中获取数据。
broker
broker 作为消息服务载体，一般又多个kafka server组成。
topic
kafka对消息进行拆分的方式，不同的数据被保存在不同的topic中，通过将消息推入某个topic或者从某个topic中获取消息来实现消息业务的进行。
partition
每个topic拆分成多个partition，每个partition又由一个一个消息组成。每个消息都被标识了一个递增序列号代表其进来的先后顺序，并按顺序存储在partition中。
consumer group
关于用户组可以简单理解为。假如有十个consumer订阅同一个topci其groupid不同，则这个topic中的任一条消息被消费10次。如果其groupid相同，此消息只能被其中一个消费。
整个消息生产流程
消息的消费流程
关于Spring-Boot整合kafka
关于依赖
主项目依赖<parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.0.8.RELEASE</version><relativePath/> </parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version></properties>
子项目依赖
<dependencies><dependency><artifactId>base-core</artifactId><groupId>daifyutils</groupId><version>1.0-SNAPSHOT</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency></dependencies>
spring-kafka配置
spring:application:name: base.kafkakafka:bootstrap-servers: kafka服务地址1:端口,kafka服务地址2:端口,kafka服务地址3:端口producer:# 写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，# 当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。retries: 0#procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：#acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。#acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。#acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。#可以设置的值为：all, -1, 0, 1acks: 1consumer:group-id: testGroup# smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallestauto-offset-reset: earliest# 设置自动提交offsetenable-auto-commit: true# 如果'mit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。auto-commit-interval: 100max-poll-records: 5server:port: 8060
对topic的操作
基于topic的操作需要我们在项目中实例化下面的bean。SpringBoot对topic的操作主要是通过KafkaAdmin进行操作
/*** 主要是初始化对kafka进行操作的admin对象* @author daify*/@Configuration@ConditionalOnClass(KafkaAdmin.class)@EnableConfigurationProperties(KafkaProperties.class)public class KafkaBaseConfiguration {private final KafkaProperties properties;public KafkaBaseConfiguration(KafkaProperties properties) {this.properties = properties;}/*** 初始化对kafka执行操作的对象* @return*/@Beanpublic KafkaAdmin kafkaAdmin() {KafkaAdmin admin = new KafkaAdmin(this.properties.buildProducerProperties());return admin;}/*** 初始化操作连接* @return*/@Beanpublic AdminClient adminClient() {return AdminClient.create(kafkaAdmin().getConfig());}}
新增topic
假如希望操作kafka中的topic可以使用下面的代码
新增
/*** 创建topic* @param topicName* @return*/public String createTopic(String topicName) {NewTopic topic = new NewTopic(topicName, 2, (short) 1);adminClient.createTopics(Arrays.asList(topic));return topicName;}
查看topic
需要查询指定topic的信息可以使用下面的方式
/*** 查询topic* @param topicName* @return*/public String queryTopic(String topicName) {DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList(topicName));StringBuffer sb = new StringBuffer("topic信息:");try {result.all().get().forEach((k,v)->sb.append("key").append(k).append(";v:").append(v));} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}return sb.toString();}
比如上面我们创建的topic用上面代码查询可以得到下面内容
name=createTopic3, internal=false, partitions=(partition=0, leader=localhost:9093 (id: 1 rack: null),replicas=localhost:9093 (id: 1 rack: null), localhost:9094 (id: 2 rack: null), isr=localhost:9093 (id: 1 rack: null), localhost:9094 (id: 2 rack: null)),(partition=1, leader=localhost:9094 (id: 2 rack: null), replicas=localhost:9094 (id: 2 rack: null), localhost:9095 (id: 3 rack: null), isr=localhost:9094 (id: 2 rack: null), localhost:9095 (id: 3 rack: null))
删除topic
需要删除指定topic的信息可以使用下面的方式
/*** 删除topic* @param topicName* @return*/public String deleteTopic(String topicName) {adminClient.deleteTopics(Arrays.asList(topicName));return topicName;}
消息的生产和消费
消息的生产
需要向kafka发送消息可以使用下面的语句。
/*** kafka消息发送者*/@Componentpublic class KafkaSender {@Autowiredprivate KafkaTemplate<String,Object> kafkaTemplate;/*** 发送文字消息* @param message* @return*/public String sendStr(String message){kafkaTemplate.send(KafkaConfig.TOPIC1,message);return message;}/*** 发送对象消息* @param obj* @return*/public String sendObj(Object obj){String message = JSON.toJSONString(obj);kafkaTemplate.send(KafkaConfig.TOPIC2,message);return message;}}
消息的消费
Spring-Boot封装了大量kafka的方法。所以仅仅是需要获得某些topic的数据可以使用这种方法创建消息监听器
@Component@Log4j2public class KafkaConsumerListener {@KafkaListener(topics = "kafka-topic1")public void onMessage1(String message){System.out.println(message);log.info("kafka-topic1接收结果:{}",message);}@KafkaListener(topics = "kafka-topic2")public void onMessage2(String message){System.out.println(message);log.info("kafka-topic2接收结果:{}",message);}}
上面的代码订阅量两个不同的topic。而我们在初始化配置的时候有一项group-id: testGroup这个时候项目启动时，使用上面的方法消费者实例会被归入到testGroup用户组。
当我们向某个topic发送消息的时候控制台会输入下面内容
{"id":100,"name":"message对象","type":2}-05-02 00:09:01.400 INFO 20296 --- [ntainer#0-0-C-1] d.s.k.b.consumer.KafkaConsumerListener : kafka-topic2接收结果:{"id":100,"name":"message对象","type":2}
到目前，springboot关于kafka的简单操作就截止了。后面我会介绍一些其他的内容。
个人水平有限，上面的内容可能存在没有描述清楚或者错误的地方，假如开发同学发现了，请及时告知，我会第一时间修改相关内容。假如我的这篇内容对你有任何帮助的话，麻烦给我点一个赞。你的点赞就是我前进的动力。
如果觉得《Spring Boot 整合——Spring Boot整合kafka整合》对你有帮助，请点赞、收藏，并留下你的观点哦！
