spring-boot-route（十四）整合Kafka

「爱情、让人受尽委屈。」 2021-09-18 06:02 280阅读 0赞

在上一章中SpringBoot整合RabbitMQ，已经详细介绍了消息队列的作用，这一种我们直接来学习SpringBoot如何整合kafka发送消息。

## kafka简介 ##

kafka是用Scala和Java语言开发的，高吞吐量的分布式消息中间件。高吞吐量使它在大数据领域具有天然的优势，被广泛用来记录日志。

### kafka架构分析 ###

![25edb0b89df6e5f373f67e7d2c5de485.png][]

注1：图中的红色箭头表示消息的流动过程，蓝色表示分区备份，绿色表示kafka集群注册到zookeeper。

注2：在kafka0.9版本之前，消费者消费消息的位置记录在zookeeper中，在0.9版本之后，消费消息的位置记录在kafka的一个topic上。

**kafka名词简介**：

1.  Producer：消息生产者
2.  Consumer：消息消费者
3.  Consumer Group（CG）：消费者组，一个topic可以有多个CG，每个Partition只会把消息发送给GG中的一个Consumer
4.  Broker：一台kafka服务器就是一个broker，一个broker有多个topic
5.  Topic：消息主题，消息分类，可看作队列
6.  Partition：分区，为了实现扩展，一个大的topic可能分布到多个broker上，一个topic可以分为多个partition，partition中的每条消息都会被分配一个有序的id（offset），每个partiton中的消息是有序的。
7.  Offset：kafka的存储文件都是按照offset.kafka来命名的，方便查找，第一个offset为0000000000.kafka。
8.  Leader：分区具有被备份，主分区
9.  Follower：从分区

#### 1. 生产者分区策略 ####

1.  指定分区。
2.  没有指定分区但有key值，将key的hash值与当前topic的分区个数进行取余得到分区。
3.  如果既没有指定分区又没有指定key，第一次调用时随机生成一个整数（以后调用每次在这个整数上自增），将这个随机数与该topic的分区数取余得到分区。

#### 2. 消息可靠性问题 ####

采用ack确认机制来保证消息的可靠性。

kafka在发送消息后会同步到其他分区副本，等所有副本都接收到消息后，kafka才会发送ack进行确认。采用这种模式的劣势就是当其中一个副本宕机后，则消息生产者就不会收到kafka的ack。

kafka采用**ISR**来解决这个问题。

**ISR**：Leader维护的一个和leader保持同步的follower集合。

当ISR中的folower完成数据同步之后，leader就会向follower发送ack，如果follower长时间未向leader同步数据，则该follower就会被踢出ISR，该时间阀值的设置参数为`replica.lag.time.max.ms`，默认时间为10s，leader发生故障后，就会从ISR中选举新的leader。

> 注：本文所讲的kafka版本为0.11，在0.9版本以前成为ISR还有一个条件，就是同步消息的条数。

**ack参数配置**

0：生产者不等待broker的ack。

1：leader分区接收到消息向生产者发送ack。

\-1（all）：ISR中的leader和follower同步成功后，向生产者发送ack。

#### 3. 消息一致性问题 ####

假如leader中有10条消息，向两个follower同步数据，follower A同步了8条，follower B同步了9条。这时候leader宕机了，follower A和follower B中的消息是不一致的，剩下两个follower就会重新选举出一个leader。

*  LEO（log end offset）：每个副本的最后一个offset
 *  HW（high watermark）：所有副本中最小的offset

为了保证数据的一致性，所有的follower会将各自的log文件**高出HW的部分**截掉，然后再从新的leader中同步数据。

#### 4. 消息重复性问题 ####

在kafka0.11版本中引入了一个新特性：幂等性。启用幂等性后，ack默认为-1。将生产者中的`enable.idompotence`设置为true，即启用了幂等性。

开启幂等性的Producer在初始化的时候会被分配一个PID，发往同一Partition的消息会附带Sequence Number。Broker端会对<PID,Partition,SeqNumber>做缓存，当具有相同主键的消息提交时，Broker只会缓存一条。但是每次重启PID就会发生变化，因此只能保证一次会话同一分区的消息不重复。

#### 5. 消费者组分区分配策略 ####

kafka有两种分配策略，一种是RoundRobin，另一种是Range

**RoundRobin**是按照消费者组以轮询的方式去给消费者分配分区的方式，前提条件是消费者组中的消费者需要订阅同一个topic。

**Range**是kafka默认的分配策略，它是通过当前的topic按照一定范围来分配的，假如有3个分区，消费者组有两个消费者，则消费者A去消费1和2分区，消费者B去消费3分区。

#### 6. 消费者offset维护 ####

Kafka 0.9 版本之前，consumer默认将offset保存在zookeeper中，0.9 版本开始，offset保存在kafka的一个内置topic中，该topic为`_consumer_offsets`。

#### 7. 生产者事务 ####

为了实现跨分区会话的事务，需要引入一个全局唯一的Tracscation ID，并将Producer 获得的PID与之绑定。这样当Producer重启后就可以通过正在进行的Transaction ID获得原来的PID。

为了管理Transcation ID，kafka引入了一个新的组件Transcation Coordinator。Producer就是通过和Transcation Coordinator交互获得Transction ID对应的任务状态。

## Spring Boot 整合kafka ##

### 1. 引入kafka依赖 ###

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

### 2. 配置kafka服务信息 ###

spring:
      kafka:
        # kafka服务地址
        bootstrap-servers: 47.104.155.182:9092
        producer:
          # 生产者消息key序列化方式
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          # 生产者消息value序列化方式
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer:
          # 消费者组
          group-id: test-consumer-group
          # 消费者消息value反序列化方式
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          # 消费者消息value反序列化方式
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

### 3. 消费者 ###

@Component
    @Slf4j
    @KafkaListener(topics = { "first-topic"},groupId = "test-consumer-group")
    public class Consumer { 
    
        @KafkaHandler
        public void receive(String message){ 
            
            log.info("我是消费者，我接收到的消息是："+message);
        }
    }

### 4. 生产者 ###

@RestController
    public class Producer { 
    
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        @GetMapping("send")
        public void send(){ 
    
            String message = "你好，我是Java旅途";
            // 第一个参数 topic
            // 第二个参数 消息
            kafkaTemplate.send("first-topic",message);
        }
    }

--------------------

#### 本文示例代码已上传至[github][]，点个`star`支持一下！ ####

## Spring Boot系列教程目录 ##

[spring-boot-route（一）Controller接收参数的几种方式][spring-boot-route_Controller]

[spring-boot-route（二）读取配置文件的几种方式][spring-boot-route]

[spring-boot-route（三）实现多文件上传][spring-boot-route 1]

[spring-boot-route（四）全局异常处理][spring-boot-route 2]

[spring-boot-route（五）整合swagger生成接口文档][spring-boot-route_swagger]

[spring-boot-route（六）整合JApiDocs生成接口文档][spring-boot-route_JApiDocs]

[spring-boot-route（七）整合jdbcTemplate操作数据库][spring-boot-route_jdbcTemplate]

[spring-boot-route（八）整合mybatis操作数据库][spring-boot-route_mybatis]

[spring-boot-route（九）整合JPA操作数据库][spring-boot-route_JPA]

[spring-boot-route（十）多数据源切换][spring-boot-route 3]

[spring-boot-route（十一）数据库配置信息加密][spring-boot-route 4]

[spring-boot-route（十二）整合redis做为缓存][spring-boot-route_redis]

[spring-boot-route（十三）整合RabbitMQ][spring-boot-route_RabbitMQ]

[spring-boot-route（十四）整合Kafka][spring-boot-route_Kafka]

[spring-boot-route（十五）整合RocketMQ][spring-boot-route_RocketMQ]

[spring-boot-route（十六）使用logback生产日志文件][spring-boot-route_logback]

[spring-boot-route（十七）使用aop记录操作日志][spring-boot-route_aop]

[spring-boot-route（十八）spring-boot-adtuator监控应用][spring-boot-route_spring-boot-adtuator]

[spring-boot-route（十九）spring-boot-admin监控服务][spring-boot-route_spring-boot-admin]

[spring-boot-route（二十）Spring Task实现简单定时任务][spring-boot-route_Spring Task]

[spring-boot-route（二十一）quartz实现动态定时任务][spring-boot-route_quartz]

[spring-boot-route（二十二）实现邮件发送功能][spring-boot-route 5]

[spring-boot-route（二十三）开发微信公众号][spring-boot-route 6]

[spring-boot-route（二十四）分布式session的一致性处理][spring-boot-route_session]

[spring-boot-route（二十五）两行代码实现国际化][spring-boot-route 7]

[spring-boot-route（二十六）整合webSocket][spring-boot-route_webSocket]

[25edb0b89df6e5f373f67e7d2c5de485.png]: /images/20210918/0a0edf2bceb44676b94d345853730b11.png
[github]: https://github.com/binzh303/spring-boot-route
[spring-boot-route_Controller]: https://javatip.cn/archives/66
[spring-boot-route]: https://javatip.cn/archives/72
[spring-boot-route 1]: https://javatip.cn/archives/74
[spring-boot-route 2]: https://javatip.cn/archives/79
[spring-boot-route_swagger]: https://javatip.cn/archives/90
[spring-boot-route_JApiDocs]: https://javatip.cn/archives/91
[spring-boot-route_jdbcTemplate]: https://javatip.cn/archives/92
[spring-boot-route_mybatis]: https://javatip.cn/archives/93
[spring-boot-route_JPA]: https://javatip.cn/archives/94
[spring-boot-route 3]: https://javatip.cn/archives/95
[spring-boot-route 4]: https://javatip.cn/archives/113
[spring-boot-route_redis]: https://javatip.cn/archives/114
[spring-boot-route_RabbitMQ]: https://javatip.cn/archives/115
[spring-boot-route_Kafka]: https://javatip.cn/archives/116
[spring-boot-route_RocketMQ]: https://javatip.cn/archives/117
[spring-boot-route_logback]: https://javatip.cn/archives/118
[spring-boot-route_aop]: https://javatip.cn/archives/119
[spring-boot-route_spring-boot-adtuator]: https://javatip.cn/archives/120
[spring-boot-route_spring-boot-admin]: https://javatip.cn/archives/121
[spring-boot-route_Spring Task]: https://javatip.cn/archives/122
[spring-boot-route_quartz]: https://javatip.cn/archives/123
[spring-boot-route 5]: https://javatip.cn/archives/124
[spring-boot-route 6]: https://javatip.cn/archives/125
[spring-boot-route_session]: https://javatip.cn/archives/284
[spring-boot-route 7]: https://javatip.cn/archives/256
[spring-boot-route_webSocket]: https://javatip.cn/archives/255