博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
SpringBoot整合SpringKafka实现消费者史上最简代码实现
阅读量:6264 次
发布时间:2019-06-22

本文共 3550 字,大约阅读时间需要 11 分钟。

该项目是使用的技术:SpringBoot  + SpringKafka + Maven

先看pom.xml文件中引入的依赖:

4.0.0
com.xuebusi.consumer
consumer
0.0.1-SNAPSHOT
jar
springkafkaconsumer
Demo project for Spring Boot
org.springframework.boot
spring-boot-starter-parent
1.5.8.RELEASE
UTF-8
UTF-8
1.8
org.springframework.boot
spring-boot-starter-freemarker
org.springframework.kafka
spring-kafka
1.0.6.RELEASE
org.springframework.boot
spring-boot-starter-web
org.springframework.boot
spring-boot-starter-test
test
org.springframework.boot
spring-boot-maven-plugin

 

注意:这里我使用的spring-kafka(它包装了apache的kafka-client)的依赖包版本是 1.0.6.RELEASE, 是因为我Linux服务器上部署的kafka服务器的版本是kafka_2.10-0.9.0.1,使用的kafka的时候要注意,kafka客户端(kafka-client)的版本要和kafka服务器的版本一一对应,否则,消息会消费失败。

Spring官方网站上给出了SpringKafka和kafka-client版本(它的版本号要和kafka服务器的版本保持一致)的对应关系:

https://projects.spring.io/spring-kafka/

 

下面是生产者的配置文件,既然使用的是SpringBoot,配置文件就是 application.yml

server:  port: 8082spring:  kafka:    consumer:      enable-auto-commit: true      group-id: applog      auto-offset-reset: latest      bootstrap-servers: 192.168.71.11:9092,192.168.71.12:9092,192.168.71.13:9092

 

在上面的配置中,我们给消费者分配的端口号是8082,服务器有3台,分别对应3个ip地址和端口。 并配置了kafka服务器的ip地址;

enable-auto-commit: true //指定消息被消费之后自动提交偏移量(即消息的编号,表示消费到了哪个位置,消费者每消费完一条消息就会向kafka服务器汇报自己消消费到的那个消息的编号,以便于下次继续消费)。

group-id: applog //消费者组
auto-offset-reset: latest //从最近的地方开始消费

想了解关于kafka消费者相关的更多配置的话,可以自行查阅相关资料进行学习。

 

下面是kafka消费者的核心代码,实现了消息的消费逻辑:

package com.xuebusi.consumer;import org.springframework.kafka.annotation.KafkaListener;import org.springframework.stereotype.Component;/** * 消费者 * 使用@KafkaListener注解,可以指定:主题,分区,消费组 */@Componentpublic class KafkaConsumer {    @KafkaListener(topics = {"app_log"})    public void receive(String message){        System.out.println("app_log--消费消息:" + message);    }}

 

在上面的代码中,负责消费消息的关键之处就是SpringKafka提供的@KafkaListener注解,在方法上使用该注解,并指定要消费的topic(也可以指定消费组以及分区号,支持正则表达式匹配),这样,消费者一旦启动,就会监听kafka服务器上的topic,实时进行消费消息。当然,我们可以在该类中定义多个不同的方法,并都在方法上使用 @KafkaListener ,为它指定不同的topic及分区信息,这样每个方法就相当于一个消费者了。

 

下面就启动SpringBoot项目测试:

package com.xuebusi;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class SpringkafkaconsumerApplication {    public static void main(String[] args) {        SpringApplication.run(SpringkafkaconsumerApplication.class, args);    }}

 

使用鼠标右键运行main方法即可启动SpringBoot项目,在控制台你会看到消费者会成功连接到kafka服务器,并打印出消费者的参数配置信息以及消费者和topic分区的分配信息,消费者一旦发现要消费的topic中有新的消息,就会立即进行消费。

 

转载地址:http://wjdpa.baihongyu.com/

你可能感兴趣的文章
css3翻书效果
查看>>
[转]通过零拷贝实现有效数据传输
查看>>
Android基于box2d开发弹弓类游戏[二]-------------游戏界面的搭建&移动游戏场景
查看>>
spring mvc 接受数组
查看>>
syslog服务器配置
查看>>
visual studioC#调用MATLAB生成的DLL
查看>>
ArrayList,LinkedList源码解析
查看>>
java推荐书籍及下载(持续更新)
查看>>
解决iframe周围的空白问题 td自适应iframe高度
查看>>
雷达标定
查看>>
[解决]小程序要求的 TLS 版本必须大于等于 1.2
查看>>
jQuery箭头切换图片 - 学习笔记
查看>>
第七周编程总结
查看>>
济南-1031试题解题报告
查看>>
最短路径(迪杰斯特拉算法)- 数据结构和算法64
查看>>
WCF或webservice跨域 这可能是由于试图以跨域方式访问服务而又没有正确的跨域策略...
查看>>
链表例题
查看>>
网站设置中的各个功能
查看>>
微软职位内部推荐-SW Engineer II for Skype
查看>>
python中的关键字符
查看>>