4. Producer生产者

在微服务链路中，Producer（生产者）是消息的源头，其配置与行为直接决定了消息投递的可靠性、吞吐量和延迟。本章将深入剖析Kafka生产者的核心配置项、三种消息发送方式、acks确认机制以及重试策略，帮助你在实际微服务场景中做出最佳选择。

4.1 生产者客户端核心配置

一个典型的生产者客户端需要配置以下关键参数，它们共同决定了生产者的行为特征：

4.2 消息发送方式：同步 vs 异步

Kafka生产者提供三种消息发送方式，每种方式在微服务场景中有不同的适用场景：

• 发送并忘记（Fire-and-Forget）：调用 send() 后不关心返回结果。适用于日志采集、监控指标等允许少量丢失的非关键数据。
• 同步发送（Sync）：调用 send() 后调用 get() 阻塞等待结果。适用于订单创建、支付通知等需要确认投递结果的业务。
• 异步发送（Async）：调用 send() 时传入 Callback 回调函数。适用于大部分微服务间消息通信，兼顾可靠性与性能。

以下是三种发送方式的代码示例：

// 1. 发送并忘记（Fire-and-Forget）
producer.send(new ProducerRecord<>("order-topic", "orderId-123", orderJson));

// 2. 同步发送（Sync）
try {
    RecordMetadata metadata = producer.send(
        new ProducerRecord<>("order-topic", "orderId-456", orderJson)
    ).get(); // 阻塞等待
    System.out.println("消息发送成功，offset: " + metadata.offset());
} catch (Exception e) {
    System.err.println("消息发送失败: " + e.getMessage());
}

// 3. 异步发送（Async）—— 微服务推荐方式
producer.send(
    new ProducerRecord<>("order-topic", "orderId-789", orderJson),
    (metadata, exception) -> {
        if (exception == null) {
            // 发送成功，可记录日志或更新状态
            log.info("消息发送成功，partition: {}, offset: {}", 
                     metadata.partition(), metadata.offset());
        } else {
            // 发送失败，执行补偿逻辑
            log.error("消息发送失败: ", exception);
            // 可在此处将消息写入本地失败队列，后续重试
        }
    }
);

4.3 消息确认机制（acks）

acks 参数控制生产者要求leader副本在确认消息写入前必须收到的确认数量，直接影响消息的持久性和可用性：

4.4 重试机制

Kafka生产者的重试机制是保证消息最终投递成功的关键防线。重试主要针对可恢复的异常，如网络抖动、leader选举等。

核心重试参数：

• retries：重试次数，默认 2147483647（几乎无限重试）。建议保留默认值，避免因重试次数耗尽导致消息丢失。
• retry.backoff.ms：重试间隔，默认 100（100ms）。建议设置为 300-1000ms，避免频繁重试加重集群压力。
• delivery.timeout.ms：发送超时时间，默认 120000（2分钟）。超过此时间后，即使未达到重试次数上限，也会放弃发送。
• enable.idempotence：启用幂等性，默认 false。开启后，生产者会为每条消息分配唯一序列号，避免重试导致的消息重复。

以下是一个经过微服务场景优化的生产者配置示例：

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");
props.put("enable.idempotence", "true");  // 开启幂等性，自动设置 acks=all
props.put("retries", Integer.MAX_VALUE);
props.put("retry.backoff.ms", "500");     // 重试间隔500ms
props.put("delivery.timeout.ms", "30000"); // 30秒超时

// 吞吐量优化
props.put("batch.size", "32768");         // 32KB
props.put("linger.ms", "10");             // 等待10ms
props.put("buffer.memory", "67108864");   // 64MB缓冲区

// 压缩（可选，减少网络带宽）
props.put("compression.type", "snappy");  // 使用Snappy压缩

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

4.5 重试与幂等性的协同工作

当开启幂等性后，Kafka生产者的重试行为会发生重要变化，理解这些变化对微服务架构设计至关重要：

4.6 微服务链路中的重试策略最佳实践

在微服务架构中，生产者的重试策略需要与业务链路协同设计，以下是经过实践验证的推荐策略：

1. 区分可重试与不可重试异常：
   • 可重试：LeaderNotAvailableException、NotEnoughReplicasException、网络超时等
   • 不可重试：SerializationException、InvalidTopicException、消息体过大等
2. 结合本地重试与链路重试：
   • 生产者本地重试：处理瞬时故障，建议3-5次，间隔递增（如100ms、200ms、500ms）
   • 微服务链路重试：如果本地重试耗尽，将消息写入本地失败队列，由定时任务或补偿流程处理
3. 监控与告警：
   • 监控 record-error-rate 和 retries-per-second 指标
   • 当重试率超过阈值（如5%）时触发告警，排查集群或网络问题
4. 避免重试风暴：
   • 设置合理的 retry.backoff.ms（建议500ms-1s），防止重试请求压垮broker
   • 使用 max.in.flight.requests.per.connection=1（开启幂等性后自动设置），保证重试顺序