原创 | ​Spring Boot 2.x 快速整合消息中间件 Kafka
收藏

点击上方“小哈学Java”,选择星标

回复“666”,领取100G独家整理的学习资料哟~

目录

  • 一、什么是 Kafka ?

  • 二、为什么要用 Kafka ?

  • 三、Kafka 环境安装

  • 四、Spring Boot 2.x 整合 Kafka

  • 五、总结

  • 六、GitHub 源码地址

什么是 Kafka?

Kafka 是 Apache 基金会开源的一个分布式发布 - 订阅消息中间件,流处理平台。它起源于 LinkedIn,由 Scala 和 Java两种语言编写而成。于 2011 年成为 Apache 项目,2012 成为 Apache 基金会下顶级项目。

Kafka 专为分布式高吞吐系统而设计。相比较其他消息中间件,如 RabbitMq 等,Kafka 具有更好的吞吐量,内置分区,复制和固有的容错能力,使得它非常适合应用在大数据领域。另外,Kafka 还支持离线、在线消费消息。

为什么要用 Kafka

  • 低延迟 - Kafka 支持低延迟消息传递,速度极快,能达到 200w 写/秒;

  • 高性能 - Kafka对于消息的发布、订阅都具有高吞吐量。即使存储了 TB 级的消息,依然能够保证稳定的性能;

  • 可靠性 - Kafka 是分布式,分区,复制和容错的,保证零停机和零数据丢失。

  • 可拓展性 - Kafka 支持集群水平拓展。

  • 耐用性 - Kafka 使用"分布式提交日志",消息能够快速的持久化的磁盘上。

Kafka 环境安装

接下来,小哈为大家演示一下,在 Linux 系统中,采用最简单的单机安装方式, 因为本文着重点还是介绍 Spring Boot 2.x 快速集成整合 Kafka.

下载 Kafka

访问 Kafka 官网 http://kafka.apache.org/downloads,下载 tgz 包, 这里演示版本为最新的 2.3.0 版本。

解压,进入目录

下载下来过后,放置到指定位置,执行命令解压:

  1. tar -zxvf kafka_2.11-2.3.0.tgz

解压完成后,进入 Kafka 目录下:

  1. cd kafka_2.11-2.3.0

启动 zookeeper

通过 bin 目录下的 zookeeper-server-start.sh 启动脚本,来启动 zk 单节点实例:

  1. bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

启动 Kafka

通过 bin 目录下的 kafka-server-start.sh 来启动 :

  1. bin/kafka-server-start.sh config/server.properties

注意:Kafka 默认使用 9092 端口,注意关闭防火墙,阿里云服务器的话,记得添加安全组。

Spring Boot 2.x 开始整合

新建一个 Spring Boot 2.x Web 工程。

项目结构

添加 maven 依赖

小哈这里完整的 maven 依赖如下:

  1. <?xml version="1.0" encoding="UTF-8"?>

  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

  3. xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

  4. <modelVersion>4.0.0</modelVersion>

  5. <parent>

  6. <groupId>org.springframework.boot</groupId>

  7. <artifactId>spring-boot-starter-parent</artifactId>

  8. <version>2.1.2.RELEASE</version>

  9. <relativePath/> <!-- lookup parent from repository -->

  10. </parent>

  11. <groupId>site.exception</groupId>

  12. <artifactId>spring-boot-kafka</artifactId>

  13. <version>0.0.1-SNAPSHOT</version>

  14. <name>spring-boot-kafka</name>

  15. <description>Demo project for Spring Boot</description>


  16. <properties>

  17. <java.version>1.8</java.version>

  18. </properties>


  19. <dependencies>

  20. <dependency>

  21. <groupId>org.springframework.boot</groupId>

  22. <artifactId>spring-boot-starter-web</artifactId>

  23. </dependency>


  24. <!-- Kafka -->

  25. <dependency>

  26. <groupId>org.springframework.kafka</groupId>

  27. <artifactId>spring-kafka</artifactId>

  28. </dependency>


  29. <dependency>

  30. <groupId>org.springframework.kafka</groupId>

  31. <artifactId>spring-kafka-test</artifactId>

  32. <scope>test</scope>

  33. </dependency>


  34. <!-- lombok -->

  35. <dependency>

  36. <groupId>org.projectlombok</groupId>

  37. <artifactId>lombok</artifactId>

  38. <optional>true</optional>

  39. </dependency>


  40. <!-- 阿里巴巴 fastjson -->

  41. <dependency>

  42. <groupId>com.alibaba</groupId>

  43. <artifactId>fastjson</artifactId>

  44. <version>1.2.58</version>

  45. </dependency>


  46. <dependency>

  47. <groupId>org.springframework.boot</groupId>

  48. <artifactId>spring-boot-devtools</artifactId>

  49. <scope>runtime</scope>

  50. <optional>true</optional>

  51. </dependency>


  52. <dependency>

  53. <groupId>org.springframework.boot</groupId>

  54. <artifactId>spring-boot-starter-test</artifactId>

  55. <scope>test</scope>

  56. </dependency>

  57. </dependencies>


  58. <build>

  59. <plugins>

  60. <plugin>

  61. <groupId>org.springframework.boot</groupId>

  62. <artifactId>spring-boot-maven-plugin</artifactId>

  63. </plugin>

  64. </plugins>

  65. </build>


  66. </project>

添加 kafka 配置

修改 application.yml 文件,添加 kafka 相关配置:

  1. spring:

  2. kafka:

  3. # 指定 kafka 地址,我这里在本地,直接就 localhost, 若外网地址,注意修改【PS: 可以指定多个】

  4. bootstrap-servers: localhost:9092

  5. consumer:

  6. # 指定 group_id

  7. group-id: group_id

  8. auto-offset-reset: earliest

  9. # 指定消息key和消息体的编解码方式

  10. key-deserializer: org.apache.kafka.common.serialization.StringDeserializer

  11. value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

  12. producer:

  13. # 指定消息key和消息体的编解码方式

  14. key-deserializer: org.apache.kafka.common.serialization.StringDeserializer

  15. value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

关于 auto-offset-reset

auto.offset.reset 配置有3个值可以设置,分别如下:

  • earliest:当各分区下有已提交的 offset 时,从提交的 offset 开始消费;无提交的 offset时,从头开始消费;

  • latest:当各分区下有已提交的 offset 时,从提交的 offset 开始消费;无提交的 offset 时,消费新产生的该分区下的数据;

  • nonetopic各分区都存在已提交的 offset 时,从 offset 后开始消费;只要有一个分区不存在已提交的 offset,则抛出异常;

默认建议用 earliest, 设置该参数后 kafka出错后重启,找到未消费的offset可以继续消费。

而 latest 这个设置容易丢失消息,假如 kafka 出现问题,还有数据往topic中写,这个时候重启kafka,这个设置会从最新的offset开始消费, 中间出问题的那些就不管了。

none 这个设置没有用过,兼容性太差,经常出问题。

新增一个订单类

模拟业务系统中,用户每下一笔订单,就发送一个消息,供其他服务消费:

  1. /**

  2. * @author 犬小哈(公众号:小哈学Java)

  3. * @date 2019/4/12

  4. * @time 下午3:05

  5. * @discription 订单实体类

  6. **/

  7. @Data

  8. @Builder

  9. @AllArgsConstructor

  10. @NoArgsConstructor

  11. public class Order {

  12. /**

  13. * 订单id

  14. */

  15. private long orderId;

  16. /**

  17. * 订单号

  18. */

  19. private String orderNum;

  20. /**

  21. * 订单创建时间

  22. */

  23. private LocalDateTime createTime;

  24. }

添加一个消息发布者

新建一个 KafkaProvider 消息提供者类,源码如下:

  1. /**

  2. * @author 犬小哈(公众号:小哈学Java)

  3. * @date 2019/4/12

  4. * @time 下午3:05

  5. * @discription 消息提供者

  6. **/

  7. @Component

  8. @Slf4j

  9. public class KafkaProvider {


  10. /**

  11. * 消息 TOPIC

  12. */

  13. private static final String TOPIC = "xiaoha";


  14. @Autowired

  15. private KafkaTemplate<String, String> kafkaTemplate;


  16. public void sendMessage(long orderId, String orderNum, LocalDateTime createTime) {

  17. // 构建一个订单类

  18. Order order = Order.builder()

  19. .orderId(orderId)

  20. .orderNum(orderNum)

  21. .createTime(createTime)

  22. .build();


  23. // 发送消息,订单类的 json 作为消息体

  24. ListenableFuture<SendResult<String, String>> future =

  25. kafkaTemplate.send(TOPIC, JSONObject.toJSONString(order));


  26. // 监听回调

  27. future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

  28. @Override

  29. public void onFailure(Throwable throwable) {

  30. log.info("## Send message fail ...");

  31. }


  32. @Override

  33. public void onSuccess(SendResult<String, String> result) {

  34. log.info("## Send message success ...");

  35. }

  36. });

  37. }

  38. }

添加一个消息消费者

消息发送出去了,当然就需要一个消费者,消费者拿到消息后,再做相关的业务处理,这里,小哈仅仅是打印消息体。

添加 KafkaConsumer 消费者类:

  1. /**

  2. * @author 犬小哈(公众号:小哈学Java)

  3. * @date 2019/4/12

  4. * @time 下午3:05

  5. * @discription 消息消费者

  6. **/

  7. @Component

  8. @Slf4j

  9. public class KafkaConsumer {


  10. @KafkaListener(topics = "xiaoha", groupId = "group_id")

  11. public void consume(String message) {

  12. log.info("## consume message: {}", message);

  13. }

  14. }

通过 @KafkaListener注解,我们可以指定需要监听的 topic 以及 groupId, 注意,这里的 topics 是个数组,意味着我们可以指定多个 topic,如:@KafkaListener(topics={"xiaoha","xiaoha2"},groupId="group_id")

注意:消息发布者的 TOPIC 需要保持与消费者监听的 TOPIC 一致,否者消费不到消息。

单元测试

新建单元测试,功能测试消息发布,以及消费。

  1. /**

  2. * @author 犬小哈(公众号:小哈学Java)

  3. * @date 2019/4/12

  4. * @time 下午3:05

  5. * @discription

  6. **/

  7. @RunWith(SpringRunner.class)

  8. @SpringBootTest

  9. public class SpringBootKafkaApplicationTests {


  10. @Autowired

  11. private KafkaProvider kafkaProvider;


  12. @Test

  13. public void sendMessage() throws InterruptedException {

  14. // 发送 1000 个消息

  15. for (int i = 0; i < 1000; i++) {

  16. long orderId = i+1;

  17. String orderNum = UUID.randomUUID().toString();

  18. kafkaProvider.sendMessage(orderId, orderNum, LocalDateTime.now());

  19. }


  20. TimeUnit.MINUTES.sleep(1);

  21. }

  22. }

发送 1000 个消息,看消息是否能够被正常发布与消费,控制台日志如下:

可以发现,1000 个消息被成功发送,且被正常消费。

我们再验证下 Kafka 的 topic 列表,看 xiaoha 这个 topic 是否正常被创建, 执行 bin 目录下查看 topic 列表的 kafka-topics.sh 脚本:

  1. bin/kafka-topics.sh --list --zookeeper localhost:2181

好了,大功告成!

总结

小哈今天主要和大家分享了,如何安装单机版的 kafka 环境、如何在 Spring Boot 2.x 中快速集成消息中间件 Kafka,以及演示了相关示例代码来发布消息、消费消息,希望大家看完过后有所收获,下期见!

GitHub 源码地址

https://github.com/weiwosuoai/spring-boot-tutorial/tree/master/spring-boot-kafka

Ref

https://zh.wikipedia.org/wiki/Kafka

https://www.w3cschool.cn/apache_kafka/

https://juejin.im/post/5d406a925188255d352ab24e

https://www.jianshu.com/p/e1df7d18bb8f

欢迎大家添加个人微信号  jiangbingdreamer,加入群聊,一起深入交流~

免费无套路分享 | 面试&学习福利资源

最近在网上发现一个不错的 PDF 资源《Java 核心知识&面试.pdf》内容覆盖很广,Java 核心基础、Java 多线程、高并发、Spring、微服务、Netty 与 RPC、Zookeeper、Kafka、RabbitMQ、Habase、设计模式、负载均衡、分布式缓存、Hadoop、Spark、Storm、云计算等

获取方式: 关注本公众号: 小哈学Java, 后台回复资源,既可免费无套路获取资源链接


更多推荐内容

↓↓↓

Linux 最常用命令:简单易学,但能解决 95% 以上的问题

给你一份Spring Boot核心知识清单 | 建议收藏

32个 IDEA 推荐配置,团队开发必备!

收到面试通知后,如何准备可以大大提升面试成功率?

盘点16个有用的Java工具类,显著提升工作效率!



如果你喜欢本文

请长按二维码,关注小哈学Java

转发朋友圈,是对我最大的支持哟


文章有帮助的话,在看,转发吧。

谢谢支持哟 (*^__^*) 

官方公众号