生产者

这一课你将能做到

1. 用 Java KafkaProducer 向 coderoadmap.order-events 发送带 orderId key 的 JSON，并在回调里打印 partition/offset
2. 对照表格说清 acks=0/1/all 各自「可能丢什么、换来什么延迟」，并能据此为订单事件选型
3. 说明「先 MySQL 事务提交再 send」的原因，并指出无 key 导致乱序的错法

为什么要学

Consumer、Spring Kafka、投递语义课都建立在「消息已经可靠进了分区日志」这个前提上。Producer 侧三件套：序列化、分区键、acks/重试。配错会出现两种事故：以为发出去了其实丢了（下单成功通知永远不来）、同订单事件乱序（Shipped 比 Paid 先被消费）。

本课的代码独立成类即可运行，与你的 Spring Boot 主工程（MySQL + Redis）互不干扰；心智上请记住：Producer 只负责把事件可靠写入日志，业务成败仍由数据库事务决定。

人话：Producer = 往日记本某一页（分区）追加一行，并选择「等多久才敢说这句写稳了」。

先建立图像

业务线程
  │ 1. MySQL 提交订单（真相）
  │ 2. kafkaProducer.send(record)
  ▼
key/value 序列化 → 按 key 哈希选 partition → 进 RecordAccumulator 批量缓冲
  │
  ├─ acks=0   : 不等 Broker 确认（最快，可能丢）
  ├─ acks=1   : Leader 分区写入即确认（折中）
  └─ acks=all : ISR 内所有副本确认（最稳，稍慢）
       │
       ▼ 回调 (metadata | exception)

acks	谁确认才算成功	丢消息风险	典型场景
0	没人，发出即算	网络/Broker 一挂就丢	可丢的指标、埋点
1	Leader 写入	Leader 确认后立刻宕机且未同步给 Follower 时丢	一般业务事件
all（-1）	ISR 全部副本	配合 min.insync.replicas≥2 基本不丢	订单、支付相关事件

跟着做

步骤 1：确认 Broker 与 Topic

按上一课保证 localhost:9092 可用，Topic coderoadmap.order-events 3 分区。核对命令：

docker exec kafka-lab /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic coderoadmap.order-events

步骤 2：写最小可运行 Producer

Maven 依赖（Boot 项目加 spring-kafka 即含客户端；纯 Java 项目用 org.apache.kafka:kafka-clients:3.9.0）：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.9.0</version>
</dependency>

完整代码：

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class OrderEventProducerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 订单类事件：等 ISR 全部确认
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        // 幂等生产者：Broker 端按 (PID, 分区, 序号) 去重，网络重试不重复追加
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        // 批量与延迟微调：攒 16KB 或 10ms 再发，吞吐更高
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            String orderId = "1001"; // key：决定分区，保同订单顺序
            String json = "{\"event\":\"OrderCreated\",\"orderId\":\"1001\"}";
            ProducerRecord<String, String> record =
                    new ProducerRecord<>("coderoadmap.order-events", orderId, json);

            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    // 失败必须当一等公民处理：记日志/告警/补偿，不能假装成功
                    System.err.println("发送失败: " + exception.getMessage());
                } else {
                    System.out.printf("成功 partition=%d offset=%d%n",
                            metadata.partition(), metadata.offset());
                }
            });
            producer.flush(); // 进程退出前把缓冲刷出
        }
    }
}