Kafka-多线程消费及分区设置

一、Kafka是什么？

Kafka是一种高吞吐量、分布式、基于发布/订阅的消息系统。可满足每秒百万级的消息生产和消费；有一套完善的消息存储机制，确保数据高效安全且持久化；Kafka作为一个集群运行在一个或多个服务器上，可以跨多个机房，当某台故障时，生产者和消费者转而使用其他的Kafka。

消息系统：Publish/subscribe（发布/订阅者）模式

1.消息发布者发布消息到主题中，有多个订阅者消费该消息。
2.当发布者发布消息时，不管是否有订阅者都不会报错。
3.一定要先有消息发布者，后有消息订阅者。

相关术语

1.Broker：Kafka服务器，负责创建topic、消息存储和转发。
2.Topic：消息类别（主题），用于区分消息。
3.Partition：分区，真正的存储数据单元。每个Topic包含一个或多个分区，用于保存消息和维护偏移量。(一般为kafka节点数CPU的总核心数量)
4.offset：分区消息此时被消费的位置。分区中消息的唯一id。
5.Producer：消息生产者。
6.Consumer：消息消费者。
7.Consumer Group：消费者组。由消费不同的分区的多个消费者实例组成，共用同一个Group-id。
8.Message：消息，由offset（分区上的消息id）、MessageSize（消息内容data大小）、data（消息具体内容）组成。

二、初步使用

1.yml文件配置

spring:
	kafka:
	    bootstrap-servers: http://127.0.0.1:9002
	    properties:
	      security:
	        protocol: SASL_PLAINTEXT
	      sasl:
	        mechanism: PLAIN
	        jaas:
	          config: org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="123456";
	    producer:
	      # 发生错误后，消息重发的次数。
	      retries: 0
	      #当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
	      batch-size: 16384
	      # 设置生产者内存缓冲区的大小。
	      buffer-memory: 33554432
	      # 键的序列化方式
	      key-serializer: org.apache.kafka.common.serialization.StringSerializer
	      # 值的序列化方式
	      value-serializer: org.apache.kafka.common.serialization.StringSerializer
	      # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
	      # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
	      # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
	      acks: 1
	    consumer:
	      # 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5D
	      auto-commit-interval: 1S
	      # 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
	      # latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
	      # earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
	      auto-offset-reset: earliest
	      # 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量
	      enable-auto-commit: false
	      # 键的反序列化方式
	      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
	      # 值的反序列化方式
	      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
	      # 消费者超时时间 6秒
	      properties:
	        max:
	          poll:
	            interval:
	              ms: 6000
	    listener:
	      # 在侦听器容器中运行的线程数。消费者组中的实例数量。 【本次重点】
	      concurrency: 5
	      #listner负责ack，每调用一次，就立即commit
	      ack-mode: manual_immediate
	      missing-topics-fatal: false

2.生产者类

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Component
@Slf4j
public class KafkaProducer {

    // 消费者组
    public static final String TOPIC_GROUP2 = "topic.group2";


    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    public void send(String topic,Object obj) {
        String obj2String = JSONObject.toJSONString(obj);
        log.info("准备发送消息为：{}", obj2String);
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, obj);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                //发送失败的处理
                log.info(topic + " - 生产者 发送消息失败：" + throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                //成功的处理
                log.info(topic + " - 生产者 发送消息成功：" + stringObjectSendResult.toString());
            }
        });
    }
}

3.消费者类

使用注解的方式来创建主题和分区。

package com.lezhi.szxy.oa.core.kafka;

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.protobuf.ServiceException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.poi.ss.formula.functions.T;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConsumerRecordRecoverer;
import org.springframework.kafka.listener.RetryingBatchErrorHandler;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.backoff.FixedBackOff;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class KafkaConsumer {

    @Resource
    private addService addService;

    @Resource
    private RedisLockUtil redisLockUtil;
    @Resource
    RedissonClient redissonClient;

    @Resource
    RedisTemplate<String,String> redisTemplate;

    private static final String ADD_LOCK_PREFIX = "ADD_LOCK_PREFIX";
  
    ObjectMapper objectMapper = new ObjectMapper();


    /**
     * 初始化主题分区
     * @return
     */
    @Bean
    public NewTopic batchTopic() {
        log.info("初始化主题分区batchTopic : add_topic，分区：5，副本数：1 >>>>>>>>>>>>>>>>>>>>>>>>>>>>> ");
        return new NewTopic("add_topic", 5, (short) 1);
    }

    /**
     * 添加消息
     * @param ack
     */
    @KafkaListener(topics = "add_topic"C,groupId = KafkaProducer.TOPIC_GROUP2)
    public void handleAddMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.info("add_topic-队列消费端 topic:{}, 收到消息>>>>>>>>>>>>>>>>>", topic);
        Optional message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            Object msg = message.get();
            try {
                ParamImport param =  objectMapper.readValue(String.valueOf(msg) , ParamImport .class);
                String fullKey = redisLockUtil.getFullKey(ADD_LOCK_PREFIX , String.valueOf(msg));
                if(redisLockUtil.getLock(fullKey , 10000)){
					// 业务代码...
                    
                    log.info("add_topic 消费了： Topic:" + topic + ",Message:" + String.valueOf(msg));
                }else {
                    log.info("add_topic 已经被消费： Topic:" + topic + ",Message:" + String.valueOf(msg));
                }
                ack.acknowledge();
            } catch (Exception e) {
                e.printStackTrace();
                log.error("解析 <"+OaConstant.SALARY_SEND_MESSAGE_KAFKA_TOPIC+"> 数据异常");
            }
        }
    }
}

配置消费端主题分区启动后，查看kafka，add_topic主题生成五个分区实例

注意：一个消费线程，可以对应若干分区。但是为了保证数据的一致性，同一个分区同时只能备一个消费者实例消费，所以超过分区数量的消费者实例个数是多余的，会被闲置。

将消费者实例（消费线程）比为一个人，分区消息相当于一个办公位。办公位数>人数时，哪个办公位有消息待消费，人就到哪一个工位处理消息。当办公位数<人数时，后面的人数需要排队等待前面的人离开，才可以进入办公位消费。
当人再多时，只有一个办公位，人也得排队办公，属于同步消费；当办公位有多个时，才能实现多人同时操作。

单机kafka分区最好不超过5。默认使用轮询策略。

4.发送消息

public void addTopicMsg(ParamImport param) throws ServiceException {
        String json;
        try {
            json = objectMapper.writeValueAsString(param);
        } catch (JsonProcessingException e) {
            log.error("addTopicMsg-发送消息，kafka消息转换失败：{}", e);
            throw new ServiceException("发送失败");
        }
        log.info("addTopicMsg-发送消息，发送kafka请求>>>>>>>>>>>>>>>>>>>>>>>");
        kafkaTemplate.send("add_topic", json);
    }

三、减少分区数量

上文中，我们使用了new NewTopic()的方式创建分区，分区数量只能动态增加不能减少。所以我们需要根据以下步骤来重新生成分区，达成减少分区的目的。

1.停止业务服务进程

停止业务服务进程，使得不会重复生成分区。修改代码内配置的new NewTopic()配置分区数。

2.停止kafka服务进程

停止kafka服务进程，清空分区、主题等数据。

3.重新启动kafka服务

4.重新启动业务服务

此时就会根据修改后的分区设置重新生成分区。

参考文章

【SpringBoot】在Springboot中怎么设置Kafka自动创建Topic
SpringBoot+Kafka之如何优雅的创建topic
想弄明白Kafka到底是什么吗？看完这篇你就知道了！(概念、数据存储、生产者、消费者)
图解Kafka，看本篇就足够啦！