Kafka开发实录

前言

最近我总是在做大胆的事情，莫不是少年也需要冲冲冲，明明我不是这样的人啊，难不成这就是命运，来自命运石之门的选择！废话不多说，本次是Kafka的实战篇，为什么这篇这么快呢？看了开头部署就知道啦，这是我之前的笔记的PLUS修订版，有点老了，部署这块凑合看吧，最新的也差不多。

本文的主旨依旧是和前文一样，带来一套有体系的打法，就是咱们RPG推BOSS的攻略一样，要详细。首先配置部分，简单明了，说实话现在中间件很多安装部分都是极简化，看过我之前部署Redis的就知道，太方便了，Kafka稍微麻烦点，但不多。然后是Kafka监控平台常见的配置参数精解和监控指标，说实话，一个中间件要想快速上手，除了参数配置就是监控指标比较重要，这俩代表了中间件的核心在哪，其他都是学院派需要关注的理论和架构知识，对于自由派来说实战则重点关注性能优化和核心即可。接下来介绍了Kafka原生的打法以及在Java老大哥Spring框架下的固定输出手法。最后简单记录了我之前的一些小实验，贴了一些Kafka真实实战的案例。

我对本文的定位是，一本相对详细的速通攻略，新手村圣剑，也许无法陪伴你打到关底BOSS，但是力压前中期毫无问题。朋友们，石中剑就在这里，你想做那个拔出神剑的人吗？

Kafka参数精解

Kafka监控指标

topic
topic名称
Partitions
分区数
Brokers
该topic 队列分布的broker数量。
Brokers Spread %
该topic中队列在Broker中的使用率，例如集群中有5个broker，但topic只在４个broker上创建了队列，那使用率为80％。
Brokers Skew %
topic的队列倾斜率。如果集群中存在5个broker节点，topic的总分区数量为4,副本因子为2，但这些队列只分布在其中的４台broker中。那topic的broker使用率(Broker Spread)为80%。
众所周知，引入多节点的目的就是负载均衡，队列在broker中的分配自然是希望越均衡越好，期望每台broker上存储2个队列(副本因子为2，总共８个队列)，表示没有发生倾斜，如果一台broker中的存在３个队列，而另外一个broker上１个队列，那说明发生了倾斜，计算公式为超过平均队列数的broker节点个数除以总所在Broker数量，其Brokers Skew等于(1/3)=33%。
Brokers Leader Skew %
topic分区中Leader分区的倾斜率。在Kafka中，只有分区的Leader节点具有读写权限，真正影响性能读写性能的是Leader分区是否均衡，试想一下，如果一个topic有6个分区，但所有的Leader分区只分布在一两个Broker节点上，这个topic的写入、读取性能将受到制约，这个值建议维持在0％。
Replicas
副本数、副本因子，即一个分区数据存储的份数，该数值包含Leader分区。
Under Replicated %
没有跟上复制进度的副本比例，在Kafka的复制模型中，主分区负责读写，该复制组内的其他副本从主节点同步数据，如果跟不上主节点的复制进度，将被提出ISR，被剔除ISR的副本不具备选举Leader的资格，这个数据如果长期或频繁高于0，说明集群一定出现了问题。
Producer Message/Sec
消息发送实时TPS，通过JMX采集
Summed Recent Offsets
该主题当前最大的消息偏移量。
常见消费者监控信息

生产者参数

acks
这个参数用老指定分区中必须由多少个副本收到消息，之后生产者才会认为这条消息写入是成功的。acks参数有三种类型的值（都是字符串类型）。

acks=1 默认值为1.生产者发送消息之后，只要分区的leader副本成功的写入消息，生产端就会收到来自服务端的成功响应，说明发送成功。如果消息无法写入leader副本，比如在leader副本崩溃、重新选举新的leader副本的过程中，生产者就会收到一个错误的响应，为了避免消息丢失，生产者就会选择重发消息；如果消息写入leader副本并成功响应给生产者，并且在其他follower副本拉取之前leader副本崩溃，此时消息还会丢失，因为新选举的leader副本中并没有这条对应的消息。acks设置为1，是消息可靠性和吞吐量之间的这种方案。
acks=0 生产者发送消息之后，不需要等待任何服务端的响应。如果在消息从发送到写入kafka的过程中出现异常，导致kafka并没有收到消息，此时生产者是不知道的，消息也就丢失了。akcs设置为0时，kafka可以达到最大的吞吐量。
acks=-1或acks=all 生产者在消息发送之后，需要等待isr中所有的副本都成功写入消息此案能够收到服务端的成功响应。acks设置为-1，可以达到相对最强的可靠性。但这不一定是最可靠的，因为isr中可能就只有leader副本，这样就退化成了acks=1 的情况。

注意，acks参数是一个字符串类型，而不是一个整数类型**（部分配置错误会报异常，部分会自动转换）。
max.request.size
生产者客户端能发送消息的最大值，默认值为1048576B(1MB)。不建议盲目修改，这个参数涉及其他的一些参数的联动，比如broker端的message.max.bytes参数，如果broker的message.max.bytes参数设置为10，而max.request.size设置为20，当发送一条大小为15B的消息时，生产者参数就会报错。(通常可以修改，producer和broker端一起修改就行)**
retries和retry.backoff.ms
生产者重试次数，默认值为0(该参数的设置已经在kafka 2.4版本中默认设置为Integer.MAX_VALUE；同时增加了delivery.timeout.ms的参数设置)。消息在从生产者从发出到成功写入broker之前可能发生一些临时性异常，比如网络抖动、leader副本选举等，这些异常往往是可以自行恢复的，生产者可以配置retries的值，通过生产端的内部重试来恢复而不是一味的将异常抛给生产者；如果重试达到设定次数，生产者才会放弃重试并抛出异常。但是！并不是所有的异常都可以通过重试来解决，比如消息过大，超过max.request.size参数配置的数值。

重试还和参数retry.backoff.ms有关，默认值为100,用来设定两次重试之间的时间间隔，避免无效的频繁重试。在配置retries和retry.backoff.ms之前，最好先估算一下可能的异常恢复时间，这样可以设定总的重试时间要大于异常恢复时间，避免生产者过早的放弃重试。(这个配置看情况调，一般别动，更建议重要的数据捕获异常降级处理，而不是无脑重试)
connections.max.idele.ms
这个参数用来制动多久之后关闭限制的连接，默认值540000（ms），9分钟。(没必要动）
batch.size
因为理论上来说，提升batch的大小，可以允许更多的数据缓冲在里面，那么一次Request发送出去的数据量就更多了，这样吞吐量可能会有所提升。

但是这个东西也不能无限的大，过于大了之后，要是数据老是缓冲在Batch里迟迟不发送出去，那么岂不是你发送消息的延迟就会很高。

比如说，一条消息进入了Batch，但是要等待5秒钟Batch才凑满了64KB，才能发送出去。那这条消息的延迟就是5秒钟。所以需要在这里按照生产环境的发消息的速率，调节不同的Batch大小自己测试一下最终出去的吞吐量以及消息的延迟，设置一个最合理的参数。(这个配置看情况调，追求低时延就默认，高吞吐量按压测情况调大，需要配合linger.ms才能达到预期效果）
linger.ms
这个参数用来指定生产者发送ProducerBatch之前等待更多的消息（ProducerRecord）假如ProducerBatch的时间，默认值为0。ProducerBatch在被填满或者时间超过linger.ms值时发送出去。增大这个参数的值回增加消息的延迟（消费端接收延迟），但能够提升一定的吞吐量。(这个配置看情况调，追求低时延就默认，高吞吐量按压测情况调大，需要配合batch.size才能达到预期效果）
receive.buffer.bytes
这个参数用来设置socket接收消息缓冲区的大小，默认值为32768（B），即32KB。如果设置为-1，则使用操作系统的默认值。（如果Producer和Kafka处于不同的机房，则可以适当的调大这个参数值）
send.buffer.bytes
这个参数用来设置socket发送消息缓冲区的大小，默认值为131072（B），即128KB。与receive.buffer.bytes参数一样，如果设置为-1，则使用操作系统的默认值。(没必要动）
request.timeout.ms
这个参数用来配置Producer等待请求响应的最长时间，默认值为3000（ms）。请求超时之后可以选择进行重试。这个参数需要比broker端参数replica.lag.time.max.ms值要大，这样可以介绍因客户端重试引起的消息重复的概率。(这个配置看情况调，网络不好或者带宽不大量大的情况下调大）
enable.idempotence
幂等性开启，默认为false。(没必要动）
bootstrap.servers
broker集群地址，可以设置一到多个，建议至少设置为2个，若在应用程序启动的时候，一个broker节点宕机，还可以对另一个已提供节点进行连接。(基础配置，非性能配置）
buffer.memory
Kafka的客户端发送数据到服务器，不是来一条就发一条，而是经过缓冲的，也就是说，通过KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里，然后把很多消息收集成一个一个的Batch，再发送到Broker上去的，这样性能才可能高。

buffer.memory的本质就是用来约束Kafka Producer能够使用的内存缓冲的大小的，默认值32MB。如果buffer.memory设置的太小，可能导致的问题是：消息快速的写入内存缓冲里，但Sender线程来不及把Request发送到Kafka服务器，会造成内存缓冲很快就被写满。而一旦被写满，就会阻塞用户线程，不让继续往Kafka写消息了。需要结合实际业务情况压测，测算在生产环境中用户线程会以每秒多少消息的频率来写入内存缓冲。经过压测，调试出来一个合理值。(量大才调）
compression.type
压缩即空间换时间，通过空间的压缩带来速度的提升，即通过少量的cpu消耗来减少磁盘和网络传输的io。如果cpu负载比较高，不适合启用压缩；如果带宽不足，而cpu负载不高，最适合启用压缩，节约大量的带宽；尽量避免消息格式不一致带来的解压缩消耗。Broker 端也有一个参数叫 compression.type，默认值是 producer，这表示 Broker 端会“尊重”Producer 端使用的压缩算法。（一般用LZ4，如下图优势明显）

max.in.flight.requests.per.connection
此配置设置客户端在单个连接上能够发送的未确认请求的最大数量，默认为5，超过此数量会造成阻塞。设置大的值可以提高吞吐量但会增加内存使用，但是需要注意的是，当设置值大于1而且发送失败时，如果启用了重试配置，有可能会改变消息的顺序。设置为1时，即使重新发送消息，也可以保证发送的顺序和写入的顺序一致。(没必要动）

消费者参数

bootstrap.servers
broker集群地址，格式：ip1:port,ip2:port…，不需要设定全部的集群地址，设置两个或者两个以上即可。(基础配置，非性能配置）
group.id
消费者隶属的消费者组名称，如果为空会报异常，一般而言，这个参数要有一定的业务意义。(基础配置，非性能配置）
fetch.min.bytes
该参数用来配置 Consumer 在一次拉取请求（调用 poll() 方法）中能从 Kafka 中拉取的最小数据量，默认值为1（B）。Kafka 在收到 Consumer 的拉取请求时，如果返回给 Consumer 的数据量小于这个参数所配置的值，那么它就需要进行等待，直到数据量满足这个参数的配置大小。可以适当调大这个参数的值以提高一定的吞吐量，不过也会造成额外的延迟（latency），对于延迟敏感的应用可能就不可取了。(这个配置推荐调，只有追求低时延才默认，高吞吐量按压测情况调大）
fetch.max.bytes
该参数与 fetch.min.bytes 参数对应，它用来配置 Consumer 在一次拉取请求中从Kafka中拉取的最大数据量，默认值为52428800（B），也就是50MB。
如果这个参数设置的值比任何一条写入 Kafka 中的消息要小，那么会不会造成无法消费呢？该参数设定的不是绝对的最大值，如果在第一个非空分区中拉取的第一条消息大于该值，那么该消息将仍然返回，以确保消费者继续工作。Kafka 中所能接收的最大消息的大小通过服务端参数 message.max.bytes（对应于主题端参数 max.message.bytes）来设置。(这个配置看情况调，一般不用）
fetch.max.wait.ms
这个参数也和 fetch.min.bytes 参数有关，如果 Kafka 仅仅参考 fetch.min.bytes 参数的要求，那么有可能会一直阻塞等待而无法发送响应给 Consumer，显然这是不合理的。fetch.max.wait.ms 参数用于指定 Kafka 的等待时间，默认值为500（ms）。如果 Kafka 中没有足够多的消息而满足不了 fetch.min.bytes 参数的要求，那么最终会等待500ms。这个参数的设定和 Consumer 与 Kafka 之间的延迟也有关系，如果业务应用对延迟敏感，那么可以适当调小这个参数。(这个配置推荐调，只有追求低时延才默认，高吞吐量按压测情况调大）
max.partition.fetch.bytes
这个参数用来配置从每个分区里返回给 Consumer 的最大数据量，默认值为1048576（B），即1MB。这个参数与 fetch.max.bytes 参数相似，只不过前者用来限制一次拉取中每个分区的消息大小，而后者用来限制一次拉取中整体消息的大小。同样，如果这个参数设定的值比消息的大小要小，那么也不会造成无法消费，Kafka 为了保持消费逻辑的正常运转不会对此做强硬的限制。(这个配置看情况调，一般不用）
max.poll.records
这个参数用来配置 Consumer 在一次拉取请求中拉取的最大消息数，默认值为500（条）。如果消息的大小都比较小，则可以适当调大这个参数值来提升一定的消费速度。(这个配置推荐调，fetch.min.bytes、fetch.max.wait.ms和max.poll.records应同步调整）
connections.max.idle.ms
这个参数用来指定在多久之后关闭闲置的连接，默认值是540000（ms），即9分钟。(这个配置看情况调，一般不用）
exclude.internal.topics
Kafka 中有两个内部的主题： __consumer_offsets 和 __transaction_state。exclude.internal.topics 用来指定 Kafka 中的内部主题是否可以向消费者公开，默认值为 true。如果设置为 true，那么只能使用 subscribe(Collection)的方式而不能使用 subscribe(Pattern)的方式来订阅内部主题，设置为 false 则没有这个限制。(这个配置看情况调，一般不用）
receive.buffer.bytes
这个参数用来设置 Socket 接收消息缓冲区（SO_RECBUF）的大小，默认值为65536（B），即64KB。如果设置为-1，则使用操作系统的默认值。如果 Consumer 与 Kafka 处于不同的机房，则可以适当调大这个参数值。(这个配置看情况调，一般不用）
send.buffer.bytes
这个参数用来设置Socket发送消息缓冲区（SO_SNDBUF）的大小，默认值为131072（B），即128KB。与receive.buffer.bytes参数一样，如果设置为-1，则使用操作系统的默认值。(这个配置看情况调，一般不用）
request.timeout.ms
这个参数用来配置 Consumer 等待请求响应的最长时间，默认值为30000（ms）。(这个配置推荐调，他算的时间不是你拿到消息的时间而是提交偏移量的时间，如果大批量运算，很容易超过30s，最好调整）
metadata.max.age.ms
这个参数用来配置元数据的过期时间，默认值为300000（ms），即5分钟。如果元数据在此参数所限定的时间范围内没有进行更新，则会被强制更新，即使没有任何分区变化或有新的 broker 加入**(没必要动）**
reconnect.backoff.ms
这个参数用来配置尝试重新连接指定主机之前的等待时间（也称为退避时间），避免频繁地连接主机，默认值为50（ms）。这种机制适用于消费者向 broker 发送的所有请求。(没必要动）
auto.offset.reset
参数值为字符串类型，有效值为“earliest”“latest”“none”，配置为其余值会报出异常**(基础配置，非性能配置）**
enable.auto.commit
boolean 类型，配置是否开启自动提交消费位移的功能，默认开启**(基础配置，非性能配置）**
auto.commit.interval.ms
当enbale.auto.commit参数设置为 true 时才生效，表示开启自动提交消费位移功能时自动提交消费位移的时间间隔**(基础配置，非性能配置）**
partition.assignment.strategy(基础配置，非性能配置）
消费者的分区分配策略
interceptor.class(基础配置，非性能配置）
用来配置消费者客户端的拦截器
max.poll.interval.ms
这个参数定义了两次poll()之间的最大间隔，默认值为300000(5分钟)。如果超过这个间隔同样会触发rebalance。在多数情况下这个参数是导致rebalance消息重复的关键，即业务处理消息耗时太长。（request.timeout.ms调大了，这个也得调大，不然再平衡会导致Kafka暂停消费）

Kafka在非Spring和Spring环境如何开放

运行环境

kafka版本2.5.1

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

样例代码

原生API-客户端(常见于各种Kafka监控平台)

import com.message.messagenew.exception.MessageException;
import com.message.messagenew.pojo.dto.KafkaProp;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @Classname KafkaService
 * @Date 2022/3/7 17:17
 * @Author WangZY
 * @Description kafka操作
 */
@EnableConfigurationProperties(KafkaProp.class)
@Service
@Slf4j
public class KafkaAdminService {
    @Autowired
    private KafkaProp kafkaProp;
    @Autowired
    private KafkaAdminService adminService;

    private static final Map<String, AdminClient> adminClientMap = new ConcurrentHashMap<>(8);

    /**
     * 根据集群ID创建，后续扩展用，现在统一使用default
     *
     * @param clusterId 集群ID 可配置在数据库
     * @return Kafka AdminClient
     */
    public AdminClient initAdminClient(String clusterId) {
        AdminClient existClient = adminClientMap.get(clusterId);
        if (existClient == null) {
            String servers = kafkaProp.getServers();
            Properties prop = new Properties();
            prop.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
            AdminClient adminClient = AdminClient.create(prop);
            adminClientMap.put("default", adminClient);
            return adminClient;
        } else {
            return existClient;
        }
    }

    @Cacheable(cacheNames = "message:topicList")
    public List<String> listTopic() throws Exception {
        ListTopicsOptions options = new ListTopicsOptions();
        //默认false，不展示Kafka内部topic
        options.listInternal(false);
        //设置超时时间，默认无，这里设置为3s
        options.timeoutMs(3000);
        ListTopicsResult result = initAdminClient("default").listTopics(options);
        Collection<TopicListing> topicListings = result.listings().get();
        return topicListings.stream().map(TopicListing::name).collect(Collectors.toList());
    }

    /**
     * 获取topic详细配置
     *
     * @param topicNameList topic名字列表
     */
    public void getTopicConfig(List<String> topicNameList) throws Exception {
        DescribeTopicsResult aDefault = initAdminClient("default").describeTopics(topicNameList);
        Map<String, TopicDescription> topicDescriptionMap = aDefault.all().get();
        System.out.println(topicDescriptionMap);
    }

    /**
     * 创建topic
     *
     * @param topicName         名字
     * @param numPartitions     分区
     * @param replicationFactor 副本
     * @param topicConfig       精确创建 topic 通过 Map<String, String> configs
     *                          过期或达到日志上限的清理策略（delete-删除；compact-压缩）默认值为delete
     *                          topicConfig.put(TopicConfig.CLEANUP_POLICY_CONFIG,TopicConfig.CLEANUP_POLICY_DELETE);
     *                          指定给该topic最终的压缩类型（uncompressed；snappy；lz4；gzip；producer）默认值为producer
     *                          topicConfig.put(TopicConfig.COMPRESSION_TYPE_CONFIG,"snappy");
     *                          压缩日志保留的最长时间，也是消费端消息的最长时间。单位为毫秒（默认值：86400000）
     *                          topicConfig.put(TopicConfig.DELETE_RETENTION_MS_CONFIG,"86400000");
     *                          文件在文件系统上被删除前的保留时间，默认为60秒
     *                          topicConfig.put(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG,"60000");
     *                          在消息刷到磁盘前，日志分区收集的消息数（默认值：MAX_VALUE）
     *                          topicConfig.put(TopicConfig.FLUSH_MESSAGES_INTERVAL_CONFIG,"9223372036854775807");
     *                          消息刷到磁盘前，保存在内存中的最长时间，单位ms（默认值：MAX_VALUE）
     *                          topicConfig.put(TopicConfig.FLUSH_MS_CONFIG,"9223372036854775807");
     *                          当执行fetch操作后，需要一定的空间来扫描最近的offset大小。设置越大，扫描速度更快但更耗内存，一般情况下不用设置此参数。（默认值：4096）
     *                          topicConfig.put(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG,"4096");
     *                          log中能够容纳消息的最大字节数（默认值：1000012）
     *                          topicConfig.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG,"1000012");
     *                          记录标记时间与kafka本机时间允许的最大间隔，超过此值的将被拒绝
     *                          topicConfig.put(TopicConfig.MESSAGE_TIMESTAMP_DIFFERENCE_MAX_MS_CONFIG,
     *                          "9223372036854775807");
     *                          标记时间类型，是创建时间还是日志时间 CreateTime/LogAppendTime
     *                          topicConfig.put(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG,"CreateTime");
     *                          如果日志压缩设置为可用的话，设置日志压缩器清理日志的频率。默认情况下，压缩比率超过50%时会避免清理日志。
     *                          此比率限制重复日志浪费的最大空间，设置为50%，意味着最多50%的日志是重复的。更高的比率设置意味着更少、更高效 的清理，但会浪费更多的磁盘空间。
     *                          topicConfig.put(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG,"0.5");
     *                          消息在日志中保持未压缩状态的最短时间，只对已压缩的日志有效
     *                          topicConfig.put(TopicConfig.MIN_COMPACTION_LAG_MS_CONFIG,"0");
     *                          当一个producer的ack设置为all（或者-1）时，此项设置的意思是认为新记录写入成功时需要的最少副本写入成功数量。
     *                          如果此最小数量没有达到，则producer抛出一个异常（NotEnoughReplicas
     *                          或者NotEnoughReplicasAfterAppend）。 你可以同时使用min.insync.replicas
     *                          和ack来加强数据持久话的保障。一个典型的情况是把一个topic的副本数量设置为3,
     *                          min.insync.replicas的数量设置为2,producer的ack模式设置为all，这样当没有足够的副本没有写入数据时，producer会抛出一个异常。
     *                          topicConfig.put(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,"1");
     *                          如果设置为true，会在新日志段创建时预分配磁盘空间
     *                          topicConfig.put(TopicConfig.PREALLOCATE_CONFIG,"true");
     *                          当保留策略为删除（delete）时，此设置控制在删除就日志段来清理磁盘空间前，保存日志段的partition能增长到的最大尺寸。
     *                          默认情况下没有尺寸大小限制，只有时间限制。。由于此项指定的是partition层次的限制，它的数量乘以分区数才是topic层面保留的数量。
     *                          topicConfig.put(TopicConfig.RETENTION_BYTES_CONFIG,"-1");
     *                          日志文件保留的毫秒值，默认七天。数据存储的最大时间超过这个时间会根据cleanup.policy策略处理数据。
     *                          topicConfig.put(TopicConfig.RETENTION_MS_CONFIG,"604800000");
     *                          此项用于控制日志段的大小，日志的清理和持久话总是同时发生，所以大的日志段代表更少的文件数量和更小的操作粒度。
     *                          topicConfig.put(TopicConfig.SEGMENT_BYTES_CONFIG,"1073741824");
     *                          此项用于控制映射数据记录offsets到文件位置的索引的大小。我们会给索引文件预先分配空间，然后在日志滚动时收缩它。 一般情况下你不需要改动这个设置。
     *                          topicConfig.put(TopicConfig.SEGMENT_INDEX_BYTES_CONFIG,"10485760");
     *                          从预订的段滚动时间中减去最大的随机抖动，避免段滚动时的惊群（thundering herds）
     *                          topicConfig.put(TopicConfig.SEGMENT_JITTER_MS_CONFIG,"0");
     *                          此项用户控制kafka强制日志滚动时间，在此时间后，即使段文件没有满，也会强制滚动，以保证持久化操作能删除或压缩就数据。默认7天
     *                          topicConfig.put(TopicConfig.SEGMENT_MS_CONFIG,"604800000");
     *                          是否把一个不在isr中的副本被选举为leader作为最后手段，即使这样做会带来数据损失
     *                          topicConfig.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG,"false");
     */
    public void createTopic(String topicName, int numPartitions, int replicationFactor,
                            Map<String, String> topicConfig) throws Exception {
        AdminClient client = initAdminClient("default");
        List<String> listTopic = adminService.listTopic();
        if (listTopic.contains(topicName)) {
            throw new MessageException("已创建相同名称的TOPIC，请更换名字");
        } else {
            NewTopic newTopic = new NewTopic(topicName, numPartitions, (short) replicationFactor);
            newTopic.configs(topicConfig);
            CreateTopicsResult createTopicResult = client
                    .createTopics(Collections.singleton(newTopic));
            createTopicResult.all().get();
        }
    }

    /**
     * 批量删除topic
     *
     * @param topicNameList topic集合
     */
    public void deleteTopic(List<String> topicNameList) throws Exception {
        DeleteTopicsResult res = initAdminClient("default").deleteTopics(topicNameList);
        res.all().get();
        if (!res.all().isDone()) {
            throw new MessageException("删除主题" + topicNameList.toString() + "失败");
        }
    }

    /**
     * 样例，后续封装
     *
     * @param topicName 主题名
     */
    public void updateTopic(String topicName) throws Exception {
        ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
        ConfigEntry configEntry = new ConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "86400000");
        AlterConfigOp op = new AlterConfigOp(configEntry, AlterConfigOp.OpType.SET);
        Map<ConfigResource, Collection<AlterConfigOp>> configs = new HashMap<>();
        configs.put(topicResource, Collections.singleton(op));
        AlterConfigsResult aDefault = initAdminClient("default").incrementalAlterConfigs(configs);
        aDefault.all().get();
    }
}

原生API-生产者

import com.message.messagenew.pojo.dto.KafkaProp;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Service;

import java.util.Properties;

/**
 * @Classname KafkaProducerService
 * @Date 2022/3/11 14:47
 * @Author WangZY
 * @Description 生产者
 */
@EnableConfigurationProperties(KafkaProp.class)
@Service
@Slf4j
public class KafkaProducerService {
    @Autowired
    private KafkaProp kafkaProp;

    /**
     * 创建Kafka生产者
     *
     * @param type 常量KafkaKey.PRODUCER_XXX
     */
    public KafkaProducer<String, String> initKafkaProducer(String type) {
        Properties props = new Properties();
        // 指定多个kafka集群多个地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProp.getServers());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        switch (type) {
            case "performance":
                // 批量消息大小。默认值是16384字节也就是16KB   推荐100-512KB
                props.put(ProducerConfig.BATCH_SIZE_CONFIG, 163840);
                // 延迟发送时间毫秒值，默认值0，与上合用，设值为5，会在没有负载的情况下为发送的记录增加5ms的延迟,
                // 启用该功能能有效减少生产者发送消息次数，从而提高并发量  推荐10-100ms
                props.put(ProducerConfig.LINGER_MS_CONFIG, 20);
                // 生产者可以使用的总内存字节来缓冲等待发送到服务器的记录，默认33554432(32MB)
                props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);
                // 消息的最大大小限制,也就是说send的消息大小不能超过这个限制, 默认1048576(1MB)
                props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760);
                // 指定分区中必须有多少个副本收到这条消息，才算消息发送成功，默认值1，字符串类型
                props.put(ProducerConfig.ACKS_CONFIG, "0");
                // 客户端将等待请求的响应的最大时间,如果在这个时间内没有收到响应，客户端将重发请求，超过重试次数将抛异常，默认30000ms
                props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
                // 压缩消息，支持四种类型，分别为：none、lz4、gzip、snappy，默认为none。
                // 消费者默认支持解压，所以压缩设置在生产者，消费者无需设置。
                props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
                break;
            case "reliableHigh":
                props.put(ProducerConfig.RETRIES_CONFIG, "3");
                props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "1000");
                props.put(ProducerConfig.ACKS_CONFIG, "-1");
                break;
            case "time":
                props.put(ProducerConfig.RETRIES_CONFIG, "1");
                props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "1000");
                //该参数指定了生产者在收到服务器响应之前可以发送多少个消息。它的值越高，就会占用越多的内存，不过也会提升吞吐量。把它设为 1 可以保证消息是按照发送的顺序写入服务器的，即使发生了重试。默认值为5
                props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
                break;
            case "normal":
            default:
                // 消息发送失败重试次数，默认0
                props.put(ProducerConfig.RETRIES_CONFIG, "1");
                // 重试间隔毫秒值，默认值100ms
                props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "1000");
                props.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
                props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
                break;
        }
        return new KafkaProducer<>(props);
    }

    public void sendEasy(String productType, String topic, String msg) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, msg);
        send(productType, record);
    }

    public void send(String productType, ProducerRecord<String, String> record) {
        try (KafkaProducer<String, String> producer = initKafkaProducer(productType)) {
            //send(record)底层调用的就是异步发送的send(record,   callback),只不过是自动将callback置为null了，所以send(record)也是异步发送的
            producer.send(record).get();
        } catch (Exception e) {
            log.error("传递消息失败，消息生产者类型={}，消息主题={}，消息体={}", productType, record.topic(), record.value(), e);
        }
    }
}

原生API-消费者

非常不建议在Spring环境使用原生消费者，不好管理该Bean的生命周期。同时提一嘴，Spring-Kafka消费者的代码非常值得学习，建议一看

import com.message.messagenew.constant.KafkaKey;
import com.message.messagenew.pojo.dto.KafkaProp;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.Properties;
import java.util.regex.Pattern;

/**
 * @Classname KafkaProducerService
 * @Date 2022/3/11 14:47
 * @Author WangZY
 * @Description 生产者
 */
@EnableConfigurationProperties(KafkaProp.class)
@Service
@Slf4j
public class KafkaConsumerService {
    @Autowired
    private KafkaProp kafkaProp;

    /**
     * 创建Kafka生产者
     *
     * @param type 常量KafkaKey.PRODUCER_XXX
     */
    public KafkaConsumer<String, String> initKafkaConsumer(String type) {
        Properties props = new Properties();
        // 指定多个kafka集群多个地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProp.getServers());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringSerializer.class);
        switch (type) {
            case "performance":
                props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerPerformanceGroup");
                props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1048576);
                props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);
                break;
            case "reliableHigh":
                props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerReliableHigh");
                props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
                break;
            case "normal":
            default:
                props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerNormal");
                break;
        }
        return new KafkaConsumer<>(props);
    }

    public void startConsumer(String type) {
        KafkaConsumer<String, String> consumer = initKafkaConsumer(type);
        consumer.subscribe(Pattern.compile(type + "-[A-Za-z0-9]+"));
        Duration timeout = Duration.ofMillis(500);
        if (KafkaKey.TOPIC_PERFORMANCE.equalsIgnoreCase(type)) {
            timeout = Duration.ofSeconds(2);
        }
        ConsumerRecords<String, String> records = consumer.poll(timeout);
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.topic());
            System.out.println(record.partition());
            System.out.println(record.key());
            System.out.println(record.value());
            System.out.println(record.offset());
        }
    }
}

Spring-Kafka-生产者

import com.ruijie.transfer.properties.TransferProperties;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author WangZY
 * @Date 2022/3/15 14:51
 * @Description 多种生产者配置
 **/
@EnableConfigurationProperties(value = {TransferProperties.class})
@Configuration
public class KafkaProducerConfig {
    @Autowired
    private TransferProperties prop;

    /**
     * @author WangZY
     * @date 2022/7/14 11:38
     * @description 高性能
     */
    @Bean
    public KafkaTemplate<String, String> performanceKafkaTemplate() {
        Map<String, Object> props = new HashMap<>(16);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getKafkaServers());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 163840);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 20);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760);
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        //config.getInt，自动强转这里不用在意是字符串还是数字
        props.put(ProducerConfig.RETRIES_CONFIG, 1);
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1000);
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
    }

    /**
     * @author WangZY
     * @date 2022/7/14 11:40
     * @description 高可靠性
     */
    @Bean
    public KafkaTemplate<String, String> reliableHighKafkaTemplate() {
        Map<String, Object> props = new HashMap<>(16);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getKafkaServers());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.RETRIES_CONFIG, "3");
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "1000");
        props.put(ProducerConfig.ACKS_CONFIG, "-1");
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
    }

    /**
     * @author WangZY
     * @date 2022/7/14 11:40
     * @description 低时延
     */
    @Bean
    public KafkaTemplate<String, String> timeKafkaTemplate() {
        Map<String, Object> props = new HashMap<>(16);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getKafkaServers());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.RETRIES_CONFIG, "1");
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "1000");
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
    }

    /**
     * @author WangZY
     * @date 2022/7/14 11:41
     * @description 普通
     */
    @Primary
    @Bean
    public KafkaTemplate<String, String> normalKafkaTemplate() {
        Map<String, Object> props = new HashMap<>(16);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getKafkaServers());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.RETRIES_CONFIG, "1");
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "1000");
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
    }
}
import com.message.messagenew.config.KafkaProducerConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * @Classname ProducerService
 * @Date 2022/3/15 14:44
 * @Author WangZY
 * @Description spring kafka生产者
 */
@Service
@Slf4j
@ConditionalOnBean(value = KafkaProducerConfig.class)
public class ProducerService {
    @Autowired
    @Qualifier("performanceKafkaTemplate")
    private KafkaTemplate<String, String> performanceProducer;
    @Autowired
    @Qualifier("reliableHighKafkaTemplate")
    private KafkaTemplate<String, String> reliableHighProducer;
    @Autowired
    @Qualifier("timeKafkaTemplate")
    private KafkaTemplate<String, String> timeProducer;
    @Autowired
    @Qualifier("normalKafkaTemplate")
    private KafkaTemplate<String, String> normalProducer;

    public void send(String type, String topic, String msg) {
        ListenableFuture<SendResult<String, String>> sendListener = null;
        switch (type) {
            case "performance":
                sendListener = performanceProducer.send(topic, msg);
                break;
            case "reliableHigh":
                sendListener = reliableHighProducer.send(topic, msg);
                break;
            case "time":
                sendListener = timeProducer.send(topic, topic, msg);
                break;
            case "normal":
            default:
                sendListener = normalProducer.send(topic, msg);
                break;
        }
        sendListener.addCallback(success -> {
            log.info("消息发送成功");
        }, err -> {
            log.error("消息发送失败", err);
        });
    }
}

Spring-Kafka-消费者

import com.ruijie.transfer.properties.TransferProperties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author WangZY
 * @Date 2022/3/15 14:51
 * @Description 多种消费者配置
 **/
@EnableConfigurationProperties(value = {TransferProperties.class})
@Configuration
public class KafkaConsumerConfig {
    @Autowired
    private TransferProperties prop;

    /**
     * @author WangZY
     * @date 2022/7/14 11:35
     * @description 高性能
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> performanceFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> container =
                new ConcurrentKafkaListenerContainerFactory<>();
        Map<String, Object> props = new HashMap<>(16);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getKafkaServers());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, StringUtils.isEmpty(prop.getConsumerGroupId()) ?
                "performanceConsumerGroup" : prop.getConsumerGroupId());
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1048576);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);
        container.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
        container.setConcurrency(3);
        container.setBatchListener(true);
        return container;
    }

    /**
     * @author WangZY
     * @date 2022/7/14 11:36
     * @description 日志用手动
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> manualFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> container =
                new ConcurrentKafkaListenerContainerFactory<>();
        Map<String, Object> props = new HashMap<>(16);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getKafkaServers());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, StringUtils.isEmpty(prop.getConsumerGroupId()) ?
                "manualConsumerGroup" : prop.getConsumerGroupId());
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1048576);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000);
        //关闭默认自动
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        container.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
        container.setConcurrency(3);
        container.setBatchListener(true);
        /*
          AckMode针对ENABLE_AUTO_COMMIT_CONFIG=false时生效，有以下几种：
          RECORD 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
          BATCH(默认) 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
          TIME 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
          COUNT 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
          COUNT_TIME TIME或COUNT满足其中一个时提交
          MANUAL poll()拉取一批消息，处理完业务后，手动调用Acknowledgment.acknowledge()
            先将offset存放到map本地缓存，在下一次poll之前从缓存拿出来批量提交
          MANUAL_IMMEDIATE 每处理完业务手动调用Acknowledgment.acknowledge()后立即提交
         */
        container.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return container;
    }

    /**
     * @author WangZY
     * @date 2022/7/14 11:36
     * @description 高可靠
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> reliableHighFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> container =
                new ConcurrentKafkaListenerContainerFactory<>();
        Map<String, Object> props = new HashMap<>(16);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getKafkaServers());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, StringUtils.isEmpty(prop.getConsumerGroupId()) ?
                "reliableHighConsumerGroup" : prop.getConsumerGroupId());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        container.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
        container.setConcurrency(3);
        container.setBatchListener(true);
        container.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return container;
    }

    /**
     * @author WangZY
     * @date 2022/7/14 11:38
     * @description 普通
     */
    @Primary
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> normalFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> container =
                new ConcurrentKafkaListenerContainerFactory<>();
        Map<String, Object> props = new HashMap<>(16);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getKafkaServers());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, StringUtils.isEmpty(prop.getConsumerGroupId()) ?
                "normalConsumerGroup" : prop.getConsumerGroupId());
        container.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
        container.setConcurrency(3);
        container.setBatchListener(true);
        return container;
    }
}
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.DependsOn;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @Classname ConsumerService
 * @Date 2022/3/15 17:03
 * @Author WangZY
 * @Description 消费者
 */
@Service
@Slf4j
@DependsOn({"kafkaConsumerConfig"})
public class ConsumerService {

    //@KafkaListener-topicPattern动态扫描topic时间约为5分钟，
    //刚生成的topic马上发送消息，因为消费者组没有订阅该topic，并且消费默认是lastest，会导致消息丢失
    @KafkaListener(topicPattern = "performance-[A-Za-z0-9]+", containerFactory = "performanceFactory")
    public void performanceListen(List<ConsumerRecord<String, String>> records) {
        System.out.println("高性能");
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.value());
        }
    }

    @KafkaListener(topicPattern = "reliableHigh-[A-Za-z0-9]+", containerFactory = "reliableHighFactory")
    public void reliableHighListen(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        System.out.println("高可用");
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.value());
        }
        ack.acknowledge();
    }

    @KafkaListener(topicPattern = "time-[A-Za-z0-9]+", containerFactory = "normalFactory")
    public void timeListen(List<ConsumerRecord<String, String>> records) {
        System.out.println("低时延");
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.value());
        }
    }

    @KafkaListener(topicPattern = "normal-[A-Za-z0-9]+", containerFactory = "normalFactory")
    public void normalListen(List<ConsumerRecord<String, String>> records) {
        System.out.println("普通");
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.value());
        }
    }
}

Springboot配置文件

#============== kafka地址 ===================
spring.kafka.bootstrap-servers=172.16.3.105:9092,172.16.3.106:9092,172.16.3.107:9092
#=============== producer  =======================
#procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
#acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
#acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
#acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
#可以设置的值为：all, -1, 0, 1
#spring.kafka.producer.acks=
spring.kafka.producer.retries=1
#每当多个记录被发送到同一分区时，生产者将尝试将记录一起批量处理为更少的请求，
#这有助于提升客户端和服务器上的性能，此配置控制默认批量大小（以字节为单位），默认值为16384
spring.kafka.producer.batch-size=16384
#生产者可用于缓冲等待发送到服务器的记录的内存总字节数，默认值为33554432
spring.kafka.producer.buffer-memory=33554432
#ID在发出请求时传递给服务器，用于服务器端日志记录
#spring.kafka.producer.client-id
#生产者生成的所有数据的压缩类型，此配置接受标准压缩编解码器（'gzip'，'snappy'，'lz4'），
#它还接受'uncompressed'以及'producer'，分别表示没有压缩以及保留生产者设置的原始压缩编解码器，默认值为producer
#spring.kafka.producer.compression-type=producer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
#=============== consumer  =======================
spring.kafka.consumer.group-id=group1
#当Kafka中没有初始偏移量或者服务器上不再存在当前偏移量时该怎么办，默认值为latest，表示自动将偏移重置为最新的偏移量
#可选的值为latest, earliest, none
spring.kafka.consumer.auto-offset-reset=latest
#如果为true，则消费者的偏移量将在后台定期提交，默认值为true
spring.kafka.consumer.enable-auto-commit=true
#如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
spring.kafka.consumer.auto-commit-interval=5000
#如果没有足够的数据立即满足“fetch.min.bytes”给出的要求，服务器在回答获取请求之前将阻塞的最长时间（以毫秒为单位）默认值为500
#spring.kafka.consumer.fetch-max-wait=
#服务器应以字节为单位返回获取请求的最小数据量，默认值为1，对应的kafka的参数为fetch.min.bytes。
#spring.kafka.consumer.fetch-min-size=
#一次调用poll()操作时返回的最大记录数，默认值为500
#spring.kafka.consumer.max-poll-records=
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#=============== listener  =======================
#侦听器的AckMode,参见https://docs.spring.io/spring-kafka/reference/htmlsingle/#committing-offsets
#当enable.auto.commit的值设置为false时，该值会生效；为true时不会生效
#spring.kafka.listener.ack-mode=
spring.kafka.listener.poll-timeout=1000
#默认是单线程，KafkaListener的形参会变化
spring.kafka.listener.type=batch
#在侦听器容器中运行的线程数，Kafka的消费者动态创建，数量与线程数一致
spring.kafka.listener.concurrency=3
spring.cloud.sentinel.transport.dashboard=192.168.58.61:8080

常见问题

集群节点有一个挂了怎么办

先查看__consumer_offsets的副本数，保证不受此影响
/wa/broker_2.5.1_0/bin/kafka-topics.sh --zookeeper localhost:2181 --topic __consumer_offsets --describe
如果副本数为1，则使用以下脚本及命令动态扩容
/wa/broker_2.5.1_0/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file /wa/__consumer_offsets.json --execute
根据数据量，副本迁移，CMAK(一个Kafka监控，可以使用know streaming等监控)上可以看到迁移过程，标红参数百分百就成功了，33正常因为50个分区三个节点不能均分

动态修改副本数坑点记录：
通过脚本修改副本数时，因为副本集中的第一个副本默认成为领导者副本，负责读写，所以如果按如下脚本修改，会导致一个问题。所有领导者副本都在0分区这台机器上，这样的话，就会导致读写次topic的负载不均衡，监控指标Preferred Replicas为33%，显示服务器只有三分之一的使用率

{
	"version": 1,
	"partitions": [{
			"topic": "test1",
			"partition": 0,
			"replicas": [0, 1, 2]
		},
		{
			"topic": "test1",
			"partition": 1,
			"replicas": [0, 1, 2]
		},
		{
			"topic": "test1",
			"partition": 2,
			"replicas": [0, 1, 2]
		}
	]
}

接下来，模拟节点挂了的情况(三个节点，主题三分区一个副本)
注意：最重要的一个点是，要确保__consumer_offsets这个内部主题是三副本。如果只有单副本的话，该节点挂了，其上副本对应的所有Topic都不能消费，理论详见前篇大碗宽面-Kafka一本道万事通

消费者项目日志开始打印Kafka警告(生产者无异常)，消息能正常生产，生产到没挂的节点(分区)

挂掉任一节点都能正常消费消息

当前能生产消息的原因是切换到了另一正常的分区，如果发送消息指定分区的情况下，该分区节点挂了，那么不能生产消息。当然也有解决方案，那就是增加副本，注意领导者副本分布在不同的机器上

{
    "version": 1, 
    "partitions": [
        {
            "topic": "test1", 
            "partition": 0, 
            "replicas": [1,2,0]
        },
        {
            "topic": "test1", 
            "partition": 1, 
            "replicas": [2,0,1]
        },
        {
            "topic": "test1", 
            "partition": 2, 
            "replicas": [0,2,1]
        }
    ]
}
执行命令
/wa/broker_2.5.1_0/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file /wa/changeReplica.json --execute

修改后如图

正常生产及消费数据，状况如图，节点2承担挂掉的节点0进行工作。重启节点后会有一段时间的再平衡，此时节点0仍不可用，还是节点2代劳

重启节点后，正常消费消息

Spring-Kafka设置并发模式

/**
 * 当配置为spring.kafka.listener.type=batch并发消费时，形参为List<ConsumerRecord<String, String>>
 * 默认单个消费时ConsumerRecord<String, String>
 */
@KafkaListener(topics = {"${kafka.topic}"})
public void kafkaListen(List<ConsumerRecord<String, String>> records) {

消费报错

Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
翻译：无法完成提交，因为组已重新平衡并将分区分配给另一个成员。这意味着后续调用 poll()之间的时间比配置的 max.poll.interval.ms 长，这通常意味着轮询循环花费了太多时间处理消息。您可以通过增加 max.poll.interval.ms 或使用 max.poll.records 减小 poll()中返回的批处理的最大大小来解决此问题。
说人话：该异常是因为一次性poll拉取（默认500）消息后处理时间过长，导致两次拉取时间间隔超过了max.poll.interval.ms阈值（默认五分钟）

实战场景

Filebeat+Kafka+数据处理服务+Elasticsearch+Kibana+Skywalking日志收集系统----2532阅读44赞62收藏，该组件服务于我自己设计并开发的完整日志收集系统，并提供Kafka生产者和消费者的模板配置，后面是本文介绍。
一个由我独立设计并开发的，完整的日志收集系统，到今天成功运行了一年半了，接入了团队的三四十个大小项目，成功抢了架构组的活，装了个大大的逼。文章详细描述了三次完整的迭代过程，为什么需要迭代？我做了什么优化？这一阶段我是怎么想的？以上大家最关心的问题，我都做出了解答。毫无疑问，这是我做过最疯狂的操作，难度系数拉满。后续更新的时候追加了一些扩充日志，以及部分配置的优化。对我来说，真的是一次很有挑战，也很长知识的经历，我至今难以想象我是如何用下班和周末时间，自己捣鼓出来这么一套庞大的东西，真TM离谱。

消息积压问题难？思路代码优化细节全公开–2600阅读55赞82收藏，同时组件为本文的Kafka配置提供了代码支持，后面是该文介绍。
我很奇怪，这篇纯纯的实战文真的是榨干了我，花了大量的时间来测试和佐证我的结论。有消息积压问题的详细处理思路和伪代码，还对Kafka的生产者消费者配置的优化给出了解释。我在整个过程中遇到的问题也有详细的记录和解决方案。