跳至主要內容

Kafka

ZJX2024年5月14日大约 10 分钟

简介

Kafka是一个开源分布式事件流平台，由 Scala 和 Java 编写（运行它需要 Java 环境）。Zookeeper 是 Kafka 早期架构中的“大脑和通讯官”，负责 broker 协调、控制器选举和元数据存储等，现在它逐步被 KRaft 架构取代，最新的kafka 4.0 版本已经完全脱离 ZooKeeper，默认以 KRaft 模式运行。

目前 kafka 几乎都是配合 zk 一起安装的，步骤稍繁琐，为了快速学习，直接安装最新版 kafka 4.0。推荐使用 Docker 安装kafka集群。

基本概念

  • 主题(Topic)

    可以把主题 Topic 理解为一个分类箱，生产者（Producer）把消息发送到某个 Topic，消费者（Consumer）从 Topic 中拉取消息。

  • Broker

    Broker 就是一个 Kafka 节点（消息服务器），负责接收、存储和转发消息。

  • 分区(Partition)

    分区是 kafka 用来对数据进行分片（一个 Topic 的数据被分散到多个 Broker 中）的一种手段，每个 Topic 中可以有多个分区。多个消费者通过并发访问不同 Broker ，实现消息负载均衡。

  • 副本(replica)

    每个分区有多个副本，这些副本被分布保存到不同的 Broker 中，以提高可用性和容错性。一个分区（Partition）在任何时候，只有一个 Broker 上的副本是 Leader 副本（仅读写），其他副本被标记为 follower 副本（仅备份）。

  • 消费者组（Consumer Group）

    一个消费者组共同消费一个 Topic 中多个分区的消息，每个消费者组中的成员不会重复消费同一个分区的数据（一个消费者可以同时消费多个分区，一个分区在同一时刻，只能被同一个消费者组内的一个消费者消费）。对于两个不同的消费者组，他们之间相互独立，即它们消费相同的 Topic ，消息互不影响。

  • 控制器（Controller）

    选举领导者（Broker）的服务器，负责维护整个 Kafka 集群的状态，并且负责管理 Topic 的增加和 Broker 的新增、删除和故障转移。在 4.x 之前，Kafka 以 ZooKeeper 作为控制器，之后使用 KRaft 模式替代它。

主题

使用 Kafaka ui 创建主题时，需要指定几个重要参数：

使用 kafka ui 创建 Topic
使用 kafka ui 创建 Topic

也可以用命令行方式创建主题，以下命令行和上图等效：

kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \        # 可指定集群中任意一个Broker，Kafka会自动与集群协调
  --topic orders \
  --partitions 3 \
  --replication-factor 3 \                   # 副本数
  --config min.insync.replicas=2 \           # 最小同步副本数
  --config retention.ms=86400000 \           # 数据保留1天（86400000ms=24小时），默认7天
  --config cleanup.policy=compact,delete \   # 混合策略：压缩Key（相同Key只保留最新数据）+按时间/大小删除，默认delete
  --config retention.bytes=1073741824 \      # 单个分区最大1GB（1073741824bytes），超过优先删除旧消息；默认无限制
  --config max.message.bytes=10485760        # 单条消息最大10MB（10485760bytes），超过无法发送；默认1M
参数说明
--create指定创建主题
--topic指定主题名称
--partitions指定该主题拥有多少个分区；一个分区只能被同一消费者组内的一个实例消费，分区数量会直接影响到吞吐量、并行度等，建议分区数 = 3 × Broker 数量，后续根据业务适当调整
--replication-factor指定每个分区都有多少份副本（包含 Leader + 所有 Follower）；推荐值为 3
min.insync.replicas指定每个分区最小同步副本；本例中副本为3，最小同步副本为2，表示容忍一个节点故障
--bootstrap-server可以指定集群中任意一个 Broker，Kafka 会自动与集群协调

提示

  1. 无需过多考虑 min.insync.replicas 的值该怎么去定义，生产环境中，副本数=3 + 最小同步副本=2 + acks=all 是生产环境黄金组合，其中 acks=all 是消息生产者的一项配置，表示需要所有副本确认，才算生产成功，详见下文整合 SpringBoot 配置文件中的 acks 配置项；
  2. 注意控制消费者实例数量，通常消费者数量 ≤ 主题分区数

整合 SpringBoot

本节源码：cnb.cool

配置

pom.xml
<!-- Spring Boot Kafka 支持，以下依赖版本都被 SpringBoot 管理，无需指定 -->
<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>
<!-- 可选依赖：kafka-streams 主要用于处理消息运算，和 Java Stream 操作集合类似，只是它操作的是消息 -->
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-streams</artifactId>
</dependency>
<!-- Jackson 序列化 -->
<dependency>
  <groupId>com.fasterxml.jackson.core</groupId>
  <artifactId>jackson-databind</artifactId>
</dependency>
<!-- Jackson Java 8 时间模块支持 -->
<dependency>
  <groupId>com.fasterxml.jackson.datatype</groupId>
  <artifactId>jackson-datatype-jsr310</artifactId>
</dependency>

生产者

生产者
@Slf4j
@Service
@RequiredArgsConstructor
public class ProducerService {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 单条消息通用消息发送方法（支持事务 + 异步回调）
     *
     * @param topicName 主题名称
     * @param msg       消息实体
     * @param <T>       泛型业务类型，必须可序列化
     */
    public <T extends Serializable> void send(String topicName, KafkaMessage<T> msg) {
        // 开启事务（executeInTransaction 是同步的）
        kafkaTemplate.executeInTransaction(operations -> {
            this.sendOperation(operations, topicName, msg);
            return true; // 给事务随便返回一个值，它就是 executeInTransaction 方法的结果，本例中没啥用
        });
        log.debug("生产者已提交事务，topic={} bizKey={}", topicName, msg.getBizKey());
    }

    /**
     * 批量消息通用消息发送方法（支持事务 + 异步回调）
     *
     * @param topicName 主题名称
     * @param msgs      消息实体集合
     * @param <T>       泛型业务类型，必须可序列化
     */
    public <T extends Serializable> void sendBatch(String topicName, List<KafkaMessage<T>> msgs) {
        kafkaTemplate.executeInTransaction(operations -> {
            for (KafkaMessage<T> msg : msgs) {
                this.sendOperation(operations, topicName, msg);
            }
            return true; // 返回值无关事务
        });
        log.debug("生产者已提交批量事务，topic={} 条数={}", topicName, msgs.size());
    }

    private void sendOperation(KafkaOperations<String, Object> operations, String topicName, KafkaMessage<?> msg) {
        // 异步发送消息（ send 方法是异步的，返回 CompletableFuture ）
        CompletableFuture<SendResult<String, Object>> future = operations.send(topicName, msg.getBizKey(), msg);
        // 添加回调处理
        future.whenComplete((result, ex) -> {
            if (ex == null) this.sendSuccess(result, msg);
            else this.sendError(msg, ex);
        });
    }

    private <T extends Serializable> void sendSuccess(SendResult<String, Object> result, KafkaMessage<T> msg) {
        RecordMetadata metadata = result.getRecordMetadata();
        log.debug("消息 bizKey={} 发送成功，topic={} partition={} offset={}",
                msg.getBizKey(),
                metadata.topic(),
                metadata.partition(),
                metadata.offset());
    }

    private <T extends Serializable> void sendError(KafkaMessage<T> msg, Throwable ex) {
        log.error("批量消息 bizKey={} 发送失败，原因：{}", msg.getBizKey(), ex.getMessage(), ex);
        // kafkaTemplate.send(topicName + ".DLQ", orderMsg.getBizKey(), orderMsg); // 可以尝试重发或加入死信队列
    }
}

消费者

// 消费者；不同组消费同一主题，他们完全隔离互不影响
@Slf4j
@Service
public class ConsumerService {

    @KafkaListener(
            topics = "orders",
            groupId = "order-consumer-group-a",
            concurrency = "3" // 启动 3 个消费者线程，相当于创建了 3 个消费者实例;
    )
    public void consumeOrder(ConsumerRecord<String, KafkaMessage<Order>> consumerRecord, Acknowledgment ack) {
        long threadId = Thread.currentThread().getId();
        KafkaMessage<Order> orderMsg = consumerRecord.value();
        log.info("消费者线程{}收到订单: {},开始处理业务...", threadId, orderMsg.getPayload());

        ack.acknowledge(); // ！！！手动提交 offset，保证消息真正被消费
        log.info("{}业务处理完成，手动通知 kafka 已完成消费...", threadId);

    }

    @KafkaListener(
            topics = "orders",
            groupId = "other-group",
            concurrency = "1"
    )
    public void consumeOther(ConsumerRecord<String, KafkaMessage<Order>> consumerRecord, Acknowledgment ack) {
        KafkaMessage<Order> orderMsg = consumerRecord.value();
        log.info(" other-group 收到订单: {},开始处理业务...", orderMsg.getPayload());
        ack.acknowledge(); // ！！！手动提交 offset，保证消息真正被消费
    }
}
最近更新：