上手实操：把原理写成可运行的 Java

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>4.0.0</version>   <!-- 4.x；本章 API 在 4.0–4.3 通用 -->
</dependency>
<dependency>
  <groupId>org.slf4j</groupId>
  <artifactId>slf4j-simple</artifactId>   <!-- 任意 SLF4J 实现即可 -->
  <version>2.0.13</version>
</dependency>

# 官方镜像 apache/kafka 自带 KRaft 默认配置，开箱即用、无需 ZooKeeper
docker run -d --name kafka -p 9092:9092 apache/kafka:latest

# 验证：列出 topic（应为空列表，不报错即连通）
docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 --list

# 建一个 4 分区、单副本的 orders topic（本机单 broker 只能 RF=1）
docker exec kafka /opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create --topic orders --partitions 4 --replication-factor 1

# 1. 生成一个集群 ID（KRaft 集群的唯一标识）
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

# 2. 用该 ID 格式化数据目录（KRaft 必需，缺这步 broker 启动即失败）
#    config/server.properties 已是 KRaft 单机默认（process.roles=broker,controller）
bin/kafka-storage.sh format --standalone \
  --cluster-id "$KAFKA_CLUSTER_ID" \
  --config config/server.properties

# 3. 启动 broker（前台运行；后台加 -daemon）
bin/kafka-server-start.sh config/server.properties

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class ReliableProducer {
    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());

        // —— 可靠性三件套 ——
        props.put(ProducerConfig.ACKS_CONFIG, "all");              // 等所有 ISR 确认
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 重试不produce重复、且保序
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE); // 配合幂等，失败就重试

        // try-with-resources 保证 close() 时 flush 掉缓冲区里未发完的记录
        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            for (int i = 0; i < 5; i++) {
                String orderId = "order-" + i;
                // key = orderId：同一订单的事件落同一分区，分区内有序
                ProducerRecord<String, String> record =
                    new ProducerRecord<>("orders", orderId, "{\"id\":\"" + orderId + "\",\"amount\":100}");

                // send 异步返回 Future；回调里能拿到 broker 分配的 partition / offset
                producer.send(record, (RecordMetadata md, Exception e) -> {
                    if (e != null) {
                        System.err.println("发送失败: " + e.getMessage());
                    } else {
                        System.out.printf("已确认 key=%s -> partition=%d offset=%d%n",
                            orderId, md.partition(), md.offset());
                    }
                });
            }
            producer.flush(); // 阻塞到上面 5 条都拿到 broker 的确认
        }
    }
}

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.StringDeserializer;

import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class ReliableConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor");   // 消费组名
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // —— 不丢消息的两个决定 ——
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);     // 关自动提交
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // 无已存 offset 时从头读

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("orders"));   // 加入组，由再平衡分到若干 partition

            while (true) {
                // poll 既拉数据，也向 broker 证明本消费者存活
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                for (ConsumerRecord<String, String> r : records) {
                    System.out.printf("收到 partition=%d offset=%d key=%s value=%s%n",
                        r.partition(), r.offset(), r.key(), r.value());
                    // ... 这里是真实业务：写库 / 调下游 ...
                }
                // 处理完整批后再同步提交 offset（at-least-once）
                if (!records.isEmpty()) {
                    consumer.commitSync();
                }
            }
        }
    }
}

# 终端 A：先启动消费者（持续运行，等待消息）
mvn -q compile exec:java -Dexec.mainClass=ReliableConsumer

# 终端 B：再运行生产者，发 5 条后退出
mvn -q compile exec:java -Dexec.mainClass=ReliableProducer

已确认 key=order-0 -> partition=2 offset=0
已确认 key=order-1 -> partition=0 offset=0
已确认 key=order-2 -> partition=2 offset=1
已确认 key=order-3 -> partition=1 offset=0
已确认 key=order-4 -> partition=3 offset=0
# 具体 partition 取决于 murmur2(key) % 4，因 key 而定、可复现；offset 在各分区内从 0 起。

收到 partition=0 offset=0 key=order-1 value={"id":"order-1","amount":100}
收到 partition=1 offset=0 key=order-3 value={"id":"order-3","amount":100}
收到 partition=2 offset=0 key=order-0 value={"id":"order-0","amount":100}
收到 partition=2 offset=1 key=order-2 value={"id":"order-2","amount":100}
收到 partition=3 offset=0 key=order-4 value={"id":"order-4","amount":100}
# partition 间到达顺序不保证；partition=2 内 offset 0 必在 1 之前——这就是"顺序只在分区内"。

// —— Producer 端 ——
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

// TODO ①：broker 端 topic 该配 min.insync.replicas 为几？（RF=3 的前提下）
//   提示：acks=all 的强度被当前 ISR 大小绑架，这个值决定 ISR 缩到多少时拒写。
//   —— 这是 broker/topic 配置，不是 producer 属性，用 kafka-configs.sh 设。

// —— Consumer 端 ——
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
        handle(r.value());

        // TODO ②：commit 放这里（每条一提交）还是放 for 循环之后（每批一提交）？
        //   两种都能跑通，但投递语义和吞吐不同。选一个，并说出代价。
    }
    // TODO ②（另一处）：还是放这里？
}

// TODO ③：orders 要求"同一订单的事件严格按 创建→支付→发货 顺序被消费"。
//   现在靠 key=orderId 的默认分区器够不够？要不要自定义 Partitioner？