kafka

02. Kafka 生产者 API 与发送语义

Kafka 生产者: ACK 策略、重试机制、幂等性与事务消息保证

Leeting Yan 5 分钟阅读 2254 字

Kafka 生产者是整个数据流的入口，其发送语义直接影响到数据的一致性和系统的可靠性。Kafka 提供了从"最多一次"到"恰好一次"的完整消息保证等级，理解这些机制需要深入到 ACK 策略、幂等性实现和事务模型的底层原理。

1. 生产者核心配置

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// ── 可靠性配置 ──
props.put("acks", "all");                        // 等待 ISR 全部确认
props.put("retries", Integer.MAX_VALUE);          // 无限重试（推荐）
props.put("enable.idempotence", true);            // 开启幂等性（Kafka 0.11+）
props.put("max.in.flight.requests.per.connection", 5); // 并发请求数

// ── 性能配置 ──
props.put("linger.ms", 10);                       // 批量等待时间
props.put("batch.size", 32768);                   // 批量大小 32KB
props.put("compression.type", "zstd");            // 压缩算法
props.put("buffer.memory", 67108864);             // 缓冲区 64MB

Producer<String, String> producer = new KafkaProducer<>(props);

2. ACK 策略：数据可靠性的基石

acks 参数控制生产者在收到 Broker 确认前需要等待的程度，是可靠性配置中最核心的参数：

acks 值行为吞吐可靠性适用场景
0发送后立即返回，不等待任何确认最高可能丢失日志采集、允许丢失的指标
1等待 Leader 确认中高Leader 宕机时可能丢失一般业务场景
all等待 ISR 中所有副本确认较低不丢失金融交易、订单等关键业务

当 acks=all 时的完整流程：

Producer ──→ Leader(Broker 1) ──→ Follower(Broker 2)
                │                      │
                │←── Ack (Replica OK) ─│
                │                        │
                │←── Ack (Replica OK) ───│
                │                        │
Producer ←──── Ack (ISR 全部确认) ────────

配合 min.insync.replicas=2 使用：只有当 ISR 中至少有 2 个副本确认时，才认为写入成功。如果 ISR 副本数不足，生产者会收到 NotEnoughReplicasException，此时可选择阻塞等待或报错。

3. 幂等性：Exactly-Once 的基础

Kafka 0.11 引入了生产者幂等性，解决了重试导致的消息重复问题：

props.put("enable.idempotence", true);  // 开启幂等性
props.put("max.in.flight.requests.per.connection", 5); // 允许并发（Kafka 1.1+）

幂等性的实现原理：

字段说明
PID（Producer ID）每个幂等性生产者被分配的唯一标识
Sequence Number每条消息在每个 Partition 内的单调递增序号
Broker 端去重Broker 根据 (PID, Partition, SeqNum) 三元组去重
Producer 发送消息:
  PID=1001, Partition=0, Seq=0  → Broker 接收并存储
  PID=1001, Partition=0, Seq=1  → Broker 接收并存储
  PID=1001, Partition=0, Seq=1  → Broker 检测到重复，返回上次的响应（幂等去重）
  PID=1001, Partition=0, Seq=3  → Broker 检测到 Seq 不连续，返回 OutOfOrderSequenceException

重要限制：

  • 幂等性只在单个生产者会话和单个分区内生效
  • 生产者重启后 PID 会变化
  • 跨区域场景需要配合事务使用

4. 事务生产者：跨分区 Exactly-Once

事务生产者将多条消息的写入和 Offset 提交原子化，实现跨分区的"恰好一次"语义：

Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("transactional.id", "order-service-producer-1");  // 事务 ID（必须唯一）
props.put("enable.idempotence", true);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", "order-001", jsonOrder));
    producer.send(new ProducerRecord<>("payments", "order-001", jsonPayment));
    producer.commitTransaction();  // 原子提交
} catch (Exception e) {
    producer.abortTransaction();    // 回滚
    throw e;
}

事务的隔离级别：

隔离级别行为
read_uncommitted（默认）消费者可读取到未提交的消息
read_committed消费者只能读取到已提交的消息（事务完成后的）
// 消费者侧配置
props.put("isolation.level", "read_committed");

5. 发送模式详解

异步发送（生产环境推荐）

// 带回调的异步发送
producer.send(new ProducerRecord<>("orders", orderId, json),
    (metadata, exception) -> {
        if (exception != null) {
            // 异常处理：记录日志、告警、死信队列
            log.error("Send failed for partition {} offset {}",
                metadata.partition(), metadata.offset(), exception);
        }
    });

同步发送

// 阻塞直到收到 Broker 确认
RecordMetadata metadata = producer.send(record).get();
System.out.println("Sent to partition " + metadata.partition() +
    " at offset " + metadata.offset());

注意：同步发送会阻塞线程，在高吞吐场景下建议使用带回调的异步发送 + semaphore 限流：

Semaphore semaphore = new Semaphore(1000); // 限制并发未确认消息数

public void sendWithBackpressure(String topic, String message) {
    semaphore.acquire();
    producer.send(new ProducerRecord<>(topic, message),
        (metadata, exception) -> {
            semaphore.release();
            if (exception != null) log.error("Send failed", exception);
        });
}

6. 分区器（Partitioner）

分区器决定消息发送到哪个 Partition，默认行为：

// 有 key：murmur2(key) % numPartitions
// 无 key：round-robin（sticky 策略，Kafka 2.4+）

自定义分区器

public class UserPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int numPartitions = cluster.partitionCountForTopic(topic);
        int vipPartition = numPartitions - 1;  // 最后一个分区给 VIP 用户

        String userId = (String) key;
        if (userId.startsWith("VIP_")) {
            return vipPartition;
        }
        return Math.abs(userId.hashCode()) % (numPartitions - 1);
    }

    @Override public void configure(Map<String, ?> configs) {}
    @Override public void close() {}
}

生产者元数据与分区策略原理

Producer 维护一份 Cluster 元数据（Topic、Partition、Leader Broker 等），默认每 5 分钟（metadata.max.age.ms）刷新一次：

发送消息流程：
1. 生产者拦截器（Interceptor）处理
2. Serializer 序列化 key/value
3. Partitioner 选择 partition（优先按 key，无 key 时用 Sticky Partitioner）
4. RecordAccumulator 将消息累加到 batch（按 topic-partition 分组）
5. Sender 线程批量发送 batch 到对应 Broker
6. 等待 Broker ACK（按 acks 配置）
7. Callback 执行 / Future 返回

Sticky Partitioner（Kafka 2.4+ 默认）：

  • 无 key 消息不再简单 round-robin，而是"粘性"地填充同一个 partition 的 batch
  • 好处：减少小 batch 数量，提升吞吐；linger.ms 内到达的消息尽量进入同一 batch
  • 切换 partition 时机：batch 满、linger 超时、或单 partition 未确认请求过多

消息顺序性保证

Kafka 仅在单个 partition 内保证消息顺序。如需全局有序：

// 方案 1：单 partition（吞吐受限）
props.put("max.in.flight.requests.per.connection", 1); // 严格顺序

// 方案 2：按业务键分区（同一用户订单始终进入同一 partition）
producer.send(new ProducerRecord<>("orders", userId, orderJson));
配置顺序保证吞吐影响
max.in.flight=1 + acks=all严格有序最低
max.in.flight=5 + 幂等性单分区有序（可能重排）高
max.in.flight=5（无幂等）无顺序保证最高

Schema Registry 集成

生产环境强烈建议使用 Schema Registry（Confluent/Apicurio）管理消息格式：

// Avro 序列化配置
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://schema-registry:8081");

// 自动注册 schema（生产环境建议关闭，改为人工审批）
props.put("auto.register.schemas", false);
props.put("use.latest.version", true);

Schema Registry 的好处：

  • 模式演进兼容检查（BACKWARD / FORWARD / FULL）
  • 减少消息体积（仅用 schema ID，而非完整 JSON Schema）
  • 多语言/多团队消费时格式一致性保障

生产者监控指标

JMX Metric说明告警阈值建议
record-error-rate发送错误率> 0.1%
record-retry-rate重试率> 5%
request-latency-avg平均请求延迟> 100ms
buffer-available-bytes缓冲区剩余空间< 10%
connection-count活跃连接数接近 max.connections
iotime-total网络 IO 耗时趋势监控

7. 生产环境最佳实践

# ── 高可靠配置（金融/订单场景） ──
acks=all
enable.idempotence=true
retries=2147483647
delivery.timeout.ms=120000
request.timeout.ms=30000
min.insync.replicas=2

# ── 高性能配置（日志/指标场景） ──
acks=1
compression.type=lz4
linger.ms=50
batch.size=262144          # 256KB
buffer.memory=134217728    # 128MB

延伸阅读

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「kafka」更多文章

  1. 集群升级与滚动重启实践
  2. Kafka 应用测试策略：Testcontainers 与集成测试
  3. 压缩算法选型：lz4、zstd、snappy 与 gzip