springboot2整合kafaka

桃扇骨 2022-03-26 14:40 276阅读 0赞

Kafka 概述

Apache Kafka 是一个分布式流处理平台，用于构建实时的数据管道和流式的应用.它可以让你发布和订阅流式的记录，可以储存流式的记录，并且有较好的容错性，可以在流式记录产生时就进行处理。

Apache Kafka是分布式发布-订阅消息系统，在 kafka官网上对 Kafka 的定义：一个分布式发布-订阅消息传递系统。

Kafka 特性

1.  高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作；
2.  可扩展性：kafka集群支持热扩展；
3.  持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失；
4.  容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）；
5.  高并发：支持数千个客户端同时读写；
6.  支持实时在线处理和离线处理：可以使用Storm这种实时流处理系统对消息进行实时进行处理，同时还可以使用Hadoop这种批处理系统进行离线处理；

Kafka 使用场景

1.  日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如Hadoop、Hbase、Solr等；
2.  消息系统：解耦和生产者和消费者、缓存消息等；
3.  用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到Hadoop、数据仓库中做离线分析和挖掘；
4.  运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告；
5.  流式处理：比如spark streaming和storm；
6.  事件源；

Spring Boot2.0 + Kafka

1，安装配置Kafka ,Zookeeper

安装和配置过程很简单，就不详细说了，参考官网：[http://kafka.apache.org/quickstart][http_kafka.apache.org_quickstart]

使用命令启动Kafka： ```bin``/kafka-server-start``.sh config``/server``.properties```

下面给出我的环境：

Centos 7.5,  Kafka 2.11, Zookeeper-3.4.13,  JDK1.8+

2，创建 Spring Boot 项目

注意版本：该项目使用Spring Boot 2.0 +，低版本可能不对

1.  pom.xml引用

1.  `<dependency>`
2.  `<groupId>org.springframework.boot</groupId>`
3.  `<artifactId>spring-boot-starter</artifactId>`
4.  `</dependency>`
5.  `<dependency>`
6.  `<groupId>org.springframework.kafka</groupId>`
7.  `<artifactId>spring-kafka</artifactId>`
8.  `</dependency>`
9.  `<dependency>`
10. `<groupId>com.alibaba</groupId>`
11. `<artifactId>fastjson</artifactId>`
12. `<version>1.2.47</version>`
13. `</dependency>`

1.  定义消息生产者  
    直接使用 KafkaTemplate 发送消息 ，Spring Boot自动装配，不需要自己定义一个Kafka配置类，吐槽一下网站的文章，全都是互相抄，全都写一个 ProduceConfig Consumerconfig 类, Kafka 的参数配置 硬编码在代码中，简直无法直视。。  
    定义一个泛型类`KafkaSender<T>` T 就是你需要发送的消息 对象，序列化使用阿里的 fastjson

消息发送后，可以在回调类里面处理自己的业务，`ListenableFutureCallback` 类有两个方法，分别是 `onFailureon` 和 `onSuccess` ，实际场景可以在这两个方法，处理自己的具体业务，这里不做实现。

1.   
2.  `/**`
3.  `* 消息生产者`
4.  `*`
5.  `* @author Jarvis`
6.  `* @date 2018/8/3`
7.  `*/`
8.  `@Component`
9.  `public class KafkaSender<T> {`
10.  
11. `private Logger logger = LoggerFactory.getLogger(KafkaSender.class);`
12.  
13. `@Autowired`
14. `private KafkaTemplate<String, Object> kafkaTemplate;`
15.  
16. `/**`
17. `* kafka 发送消息`
18. `*`
19. `* @param obj 消息对象`
20. `*/`
21. `public void send(T obj) {`
22. `String jsonObj = JSON.toJSONString(obj);`
23. `logger.info("------------ message = {}", jsonObj);`
24.  
25. `//发送消息`
26. `ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send("kafka.tut", jsonObj);`
27. `future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {`
28. `@Override`
29. `public void onFailure(Throwable throwable) {`
30. `logger.info("Produce: The message failed to be sent:" + throwable.getMessage());`
31. `}`
32.  
33. `@Override`
34. `public void onSuccess(SendResult<String, Object> stringObjectSendResult) {`
35. `//TODO 业务处理`
36. `logger.info("Produce: The message was sent successfully:");`
37. `logger.info("Produce: _+_+_+_+_+_+_+ result: " + stringObjectSendResult.toString());`
38. `}`
39. `});`
40. `}`
41. `}`

1.  定义消息消费者  
    使用`@KafkaListener` 注解监听 topics 消息，此处的`topics` 必须和 send 函数中的 一致

`@Header(KafkaHeaders.RECEIVED_TOPI` 直接获取 topic

1.   
2.  `/**`
3.  `* 监听kafka.tut 的 topic`
4.  `*`
5.  `* @param record`
6.  `* @param topic topic`
7.  `*/`
8.  `@KafkaListener(id = "tut", topics = "kafka.tut")`
9.  `public void listen(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {`
10. `//判断是否NULL`
11. `Optional<?> kafkaMessage = Optional.ofNullable(record.value());`
12.  
13. `if (kafkaMessage.isPresent()) {`
14. `//获取消息`
15. `Object message = kafkaMessage.get();`
16.  
17. `logger.info("Receive： +++++++++++++++ Topic:" + topic);`
18. `logger.info("Receive： +++++++++++++++ Record:" + record);`
19. `logger.info("Receive： +++++++++++++++ Message:" + message);`
20. `}`
21. `}`

1.  配置文件 application.yml

1.   
2.  `spring:`
3.  `application:`
4.  `name: kafka-tutorial`
5.  `kafka:`
6.  `# 指定kafka 代理地址，可以多个`
7.  `bootstrap-servers: 192.168.10.100:9092`
8.  `producer:`
9.  `retries: 0`
10. `# 每次批量发送消息的数量`
11. `batch-size: 16384`
12. `# 缓存容量`
13. `buffer-memory: 33554432`
14. `# 指定消息key和消息体的编解码方式`
15. `key-serializer: org.apache.kafka.common.serialization.StringSerializer`
16. `value-serializer: org.apache.kafka.common.serialization.StringSerializer`
17. `consumer:`
18. `# 指定默认消费者group id`
19. `group-id: consumer-tutorial`
20. `auto-commit-interval: 100`
21. `auto-offset-reset: earliest`
22. `enable-auto-commit: true`
23. `# 指定消息key和消息体的编解码方式`
24. `key-deserializer: org.apache.kafka.common.serialization.StringDeserializer`
25. `value-deserializer: org.apache.kafka.common.serialization.StringDeserializer`
26. `# 指定listener 容器中的线程数，用于提高并发量`
27. `listener:`
28. `concurrency: 3`

1.  直接使用 @Autowired 对类 KafkaSender 自动装配，然后调用 send 方法发送消息即可，下面给出代码：

1.  `@Autowired`
2.  `private KafkaSender<User> kafkaSender;`
3.   
4.  `@Test`
5.  `public void kafkaSend() throws InterruptedException {`
6.  `//模拟发消息`
7.  `for (int i = 0; i < 5; i++) {`
8.   
9.  `User user = new User();`
10. `user.setId(System.currentTimeMillis());`
11. `user.setMsg(UUID.randomUUID().toString());`
12. `user.setSendTime(new Date());`
13.  
14. `kafkaSender.send(message);`
15. `Thread.sleep(3000);`
16.  
17. `}`
18. `}`

控制台可以看到执行成功：  
![278758-20180803161103590-1060801605.png][]

在服务器执行 `bin/kafka-topics.sh --list --zookeeper localhost:2181` 可以看到topic  
![278758-20180803161122421-591754640.png][]

Kafka如何保证数据的不丢失

1.生产者数据的不丢失

*  新版本的producer采用异步发送机制。KafkaProducer.send(ProducerRecord)方法仅仅是把这条消息放入一个缓存中(即RecordAccumulator，本质上使用了队列来缓存记录)，同时后台的IO线程会不断扫描该缓存区，将满足条件的消息封装到某个batch中然后发送出去。显然，这个过程中就有一个数据丢失的窗口：若IO线程发送之前client端挂掉了，累积在accumulator中的数据的确有可能会丢失。 kafka的ack机制：在kafka发送数据的时候，每次发送消息都会有一个确认反馈机制，确保消息正常的能够被收到。
 *  如果是同步模式：ack机制能够保证数据的不丢失，如果ack设置为0，风险很大，一般不建议设置为0  
    producer.type=sync  
    request.required.acks=1
 *  如果是异步模式：通过buffer来进行控制数据的发送，有两个值来进行控制，时间阈值与消息的数量阈值，如果buffer满了数据还没有发送出去，如果设置的是立即清理模式，风险很大，一定要设置为阻塞模式  
    producer.type=async  
    request.required.acks=1  
    queue.buffering.max.ms=5000  
    queue.buffering.max.messages=10000  
    queue.enqueue.timeout.ms = -1  
    batch.num.messages=200
 *  结论：producer有丢数据的可能，但是可以通过配置保证消息的不丢失  
    2.消费者数据的不丢失
 *  如果在消息处理完成前就提交了offset，那么就有可能造成数据的丢失。由于Kafka consumer默认是自动提交位移的，所以在后台提交位移前一定要保证消息被正常处理了，因此不建议采用很重的处理逻辑，如果处理耗时很长，则建议把逻辑放到另一个线程中去做。为了避免数据丢失，现给出两点建议：  
    enable.auto.commit=false 关闭自动提交位移  
    在消息被完整处理之后再手动提交位移
 *  如果使用了storm，要开启storm的ackfail机制；
 *  如果没有使用storm，确认数据被完成处理之后，再更新offset值。低级API中需要手动控制offset值。通过offset commit 来保证数据的不丢失，kafka自己记录了每次消费的offset数值，下次继续消费的时候，接着上次的offset进行消费即可。

源码 github：[https://github.com/jarvisqi/java-tutorial/tree/master/kafka-tutorial][https_github.com_jarvisqi_java-tutorial_tree_master_kafka-tutorial]

参考：

1.  [http://kafka.apache.org/quickstart][http_kafka.apache.org_quickstart]
2.  [https://docs.spring.io/spring-kafka/reference/htmlsingle/\#kafka][https_docs.spring.io_spring-kafka_reference_htmlsingle_kafka]
3.  [https://blog.csdn.net/tzs\_1041218129/article/details/78988439][https_blog.csdn.net_tzs_1041218129_article_details_78988439]

[http_kafka.apache.org_quickstart]: http://kafka.apache.org/quickstart
[278758-20180803161103590-1060801605.png]: /images/20220326/950e26fc900c423fae5a415f18fe38a7.png
[278758-20180803161122421-591754640.png]: /images/20220326/61a32a91904844d8b7f7559ade7dd933.png
[https_github.com_jarvisqi_java-tutorial_tree_master_kafka-tutorial]: https://github.com/jarvisqi/java-tutorial/tree/master/kafka-tutorial
[https_docs.spring.io_spring-kafka_reference_htmlsingle_kafka]: https://docs.spring.io/spring-kafka/reference/htmlsingle/#kafka
[https_blog.csdn.net_tzs_1041218129_article_details_78988439]: https://blog.csdn.net/tzs_1041218129/article/details/78988439