# 确认 Kafka Broker
docker logs kafka | tail -10
# 预期：[KafkaRaftServer] Kafka Server started

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>

spring:
  kafka:
    bootstrap-servers: localhost:9092
    # ===== Producer 配置 =====
    producer:
      # Key 和 Value 的序列化器——Spring 自动注入
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      # ACK 级别：all = 所有 ISR 副本确认（最可靠）
      acks: all
      # 发送重试次数
      retries: 3
      # 批量发送大小（字节）——凑满 16KB 才发送
      batch-size: 16384
      # 批量等待时间（ms）——即使没凑满，等 10ms 也发
      linger-ms: 10
      # 压缩算法——none / gzip / snappy / lz4 / zstd
      compression-type: snappy
    # ===== Consumer 配置 =====
    consumer:
      # Key 和 Value 的反序列化器
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      # ConsumerGroup——同一组内 Partition 只会分配给一个实例
      group-id: order-consumer-group
      # 第一次连 Kafka 时从哪里开始消费
      auto-offset-reset: earliest
      # 是否自动提交 offset（false = 手动提交）
      enable-auto-commit: false
      # 每次 poll 最多拉取多少条
      max-poll-records: 50
    # ===== Listener 配置 =====
    listener:
      # 手动提交 offset
      ack-mode: manual

@Data
@NoArgsConstructor
@AllArgsConstructor
public class OrderMessage implements Serializable {
    private Long orderId;
    private Long userId;
    private String productName;
    private BigDecimal amount;
    private String action;    // created / paid / cancelled / shipped
    private LocalDateTime createTime;
}

@Service
public class OrderMessageService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    // ===== 1. 发送——默认异步，不管结果 =====
    public void send(OrderMessage msg) {
        // send 是异步的——返回 ListenableFuture，不调 .get() 就不阻塞
        kafkaTemplate.send("order-topic", String.valueOf(msg.getOrderId()), msg);
    }

    // ===== 2. 同步发送——等待 Broker 确认 =====
    public void sendSync(OrderMessage msg) throws Exception {
        SendResult<String, Object> result =
                kafkaTemplate.send("order-topic",
                        String.valueOf(msg.getOrderId()), msg)
                        .get(5, TimeUnit.SECONDS);  // 阻塞最多 5 秒

        RecordMetadata meta = result.getRecordMetadata();
        System.out.printf("发送成功: topic=%s, partition=%d, offset=%d%n",
                meta.topic(), meta.partition(), meta.offset());
    }

    // ===== 3. 异步发送 + 回调 =====
    public void sendWithCallback(OrderMessage msg) {
        ListenableFuture<SendResult<String, Object>> future =
                kafkaTemplate.send("order-topic",
                        String.valueOf(msg.getOrderId()), msg);

        future.addCallback(
            // 成功回调
            result -> {
                RecordMetadata meta = result.getRecordMetadata();
                System.out.printf("异步发送成功: offset=%d, partition=%d%n",
                        meta.offset(), meta.partition());
            },
            // 失败回调
            ex -> {
                System.err.println("异步发送失败: " + ex.getMessage());
                // 补偿逻辑：写 DB 重试表、发告警...
            }
        );
    }

    // ===== 4. 指定 Partition =====
    public void sendToPartition(OrderMessage msg, int partition) {
        kafkaTemplate.send("order-topic", partition,
                String.valueOf(msg.getOrderId()), msg);
    }

    // ===== 5. 指定 Topic + Key（同一个 Key 的消息进入同一个 Partition） =====
    public void sendWithKey(OrderMessage msg) {
        // Kafka 对 Key 取哈希 % Partition 数
        // 同一个 orderId → 同一个哈希 → 同一个 Partition → 有序！
        kafkaTemplate.send("order-topic",
                String.valueOf(msg.getOrderId()), msg);
    }
}

// 最简形式：只指定 Topic
send(topic, data)

// 指定 Key——同一个 Key 的消息哈希到同一个 Partition（保证顺序）
send(topic, key, data)

// 指定 Partition——绕过哈希，强制发到某个 Partition
send(topic, partition, key, data)

// 指定时间戳——用于按时间查询
send(topic, partition, timestamp, key, data)

@Component
public class OrderMessageListener {

    private static final Logger log = LoggerFactory.getLogger(OrderMessageListener.class);

    // ===== 基本用法：单条消费 =====
    @KafkaListener(
        topics = "order-topic",
        groupId = "order-consumer-group"
    )
    public void onMessage(OrderMessage msg) {
        log.info("收到订单消息: orderId={}, action={}, amount={}",
                msg.getOrderId(), msg.getAction(), msg.getAmount());
        // 处理业务...
    }

    // ===== 拿到完整 ConsumerRecord =====
    @KafkaListener(topics = "order-topic", groupId = "order-consumer-group")
    public void onRecord(ConsumerRecord<String, OrderMessage> record) {
        log.info("收到: topic={}, partition={}, offset={}, key={}, value={}",
                record.topic(), record.partition(), record.offset(),
                record.key(), record.value());
    }

    // ===== 批量消费——一次 poll 拿到一批消息 =====
    @KafkaListener(
        topics = "order-topic",
        groupId = "order-batch-consumer-group"
    )
    public void onBatch(List<OrderMessage> messages) {
        log.info("批量收到 {} 条消息", messages.size());
        for (OrderMessage msg : messages) {
            processOrder(msg);
        }
    }

    // ===== 拿到完整的 ConsumerRecord 列表 =====
    @KafkaListener(topics = "order-topic", groupId = "order-batch-consumer-group")
    public void onBatchRecords(List<ConsumerRecord<String, OrderMessage>> records) {
        log.info("批量收到 {} 条消息", records.size());
        for (ConsumerRecord<String, OrderMessage> record : records) {
            log.info("  partition={}, offset={}, key={}",
                    record.partition(), record.offset(), record.key());
            processOrder(record.value());
        }
    }

    private void processOrder(OrderMessage msg) {
        // 实际业务逻辑
    }
}

// Spring Kafka 根据方法参数类型自动注入：

// 1. 只有消息体
@KafkaListener(topics = "order-topic")
public void handle(OrderMessage msg)

// 2. 消息体 + Acknowledgment（手动提交 offset）
@KafkaListener(topics = "order-topic")
public void handle(OrderMessage msg, Acknowledgment ack)

// 3. ConsumerRecord（包含 topic/partition/offset/headers 全部信息）
@KafkaListener(topics = "order-topic")
public void handle(ConsumerRecord<String, OrderMessage> record)

// 4. ConsumerRecord + Acknowledgment
@KafkaListener(topics = "order-topic")
public void handle(ConsumerRecord<String, OrderMessage> record, Acknowledgment ack)

// 5. 批量：List<消息体>
@KafkaListener(topics = "order-topic")
public void handle(List<OrderMessage> messages, Acknowledgment ack)

// 6. 批量：List<ConsumerRecord>
@KafkaListener(topics = "order-topic")
public void handle(List<ConsumerRecord<String, OrderMessage>> records)

@Component
public class ManualCommitListener {

    private static final Logger log = LoggerFactory.getLogger(ManualCommitListener.class);

    // 单条消费 + 手动提交
    @KafkaListener(topics = "order-topic", groupId = "order-manual-commit")
    public void handle(OrderMessage msg, Acknowledgment ack) {
        try {
            processOrder(msg);
            // 处理成功 → 提交 offset
            ack.acknowledge();
        } catch (Exception e) {
            log.error("处理失败: orderId={}", msg.getOrderId(), e);
            // 不调用 acknowledge() → 下次 poll 时重新拉取这条消息
            // 注意：不是"立即"重新消费，而是下次 poll 时
        }
    }

    // 批量消费 + 手动提交
    @KafkaListener(topics = "order-topic", groupId = "order-batch-manual")
    public void handleBatch(List<OrderMessage> messages, Acknowledgment ack) {
        for (OrderMessage msg : messages) {
            processOrder(msg);
        }
        // 整批处理完后提交
        ack.acknowledge();
    }

    private void processOrder(OrderMessage msg) { /* ... */ }
}

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        # 告诉 Deserializer 反序列化成哪个类
        spring.json.value.default.type: com.example.demo.OrderMessage
        # 信任的包——只有这些包下的类才允许反序列化（安全机制）
        spring.json.trusted.packages: com.example.demo

// Provider 发送时，JsonSerializer 自动在消息 Header 中写入 __TypeId__
// 不需要任何额外配置——JsonSerializer 默认就写 Header
kafkaTemplate.send("order-topic", String.valueOf(msg.getOrderId()), msg);
// Header 中自动包含: __TypeId__ = com.example.demo.OrderMessage

// Consumer 接收时，JsonDeserializer 从 Header 中读取 __TypeId__ 确定目标类型
// 只需要配 spring.json.trusted.packages

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        # 信任 demo 包下的所有类
        spring.json.trusted.packages: com.example.demo
        # 如果 Header 中没有 __TypeId__，回退到这个默认类型
        spring.json.value.default.type: com.example.demo.OrderMessage

spring:
  kafka:
    consumer:
      properties:
        spring.json.trusted.packages: com.example.demo

@RestController
@RequestMapping("/api/order")
public class OrderController {

    @Autowired
    private OrderMessageService orderMessageService;

    // 创建订单 → 同步发送（等 Broker 确认）
    @PostMapping("/create")
    public String createOrder(@RequestBody OrderMessage msg) {
        msg.setAction("created");
        msg.setCreateTime(LocalDateTime.now());
        try {
            orderMessageService.sendSync(msg);
            return "订单消息已发送（同步确认）: " + msg.getOrderId();
        } catch (Exception e) {
            return "发送失败: " + e.getMessage();
        }
    }

    // 付款通知 → 异步回调发送
    @PostMapping("/pay")
    public String payOrder(@RequestBody OrderMessage msg) {
        msg.setAction("paid");
        orderMessageService.sendWithCallback(msg);
        return "支付消息已异步发送: " + msg.getOrderId();
    }

    // 日志 → 纯异步发送（丢了无所谓）
    @PostMapping("/log")
    public String logOrder(@RequestBody OrderMessage msg) {
        msg.setAction("log");
        orderMessageService.send(msg);
        return "日志已异步发送（不管结果）: " + msg.getOrderId();
    }

    // 发货通知 → 指定 Partition 发送
    @PostMapping("/ship/{orderId}")
    public String shipOrder(@PathVariable Long orderId) {
        OrderMessage msg = new OrderMessage();
        msg.setOrderId(orderId);
        msg.setAction("shipped");
        msg.setCreateTime(LocalDateTime.now());
        // 根据 orderId 哈希决定 Partition，保证同一订单的消息有序
        orderMessageService.sendWithKey(msg);
        return "发货消息已发送: " + orderId;
    }
}

# 1. 先手动创建 Topic（生产环境 autoCreateTopicEnable=false）
docker exec -it kafka \
  /opt/kafka/bin/kafka-topics.sh --create \
  --topic order-topic \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

# 2. 启动应用
mvn spring-boot:run

# 3. 发送创建订单请求
curl -X POST http://localhost:8080/api/order/create \
  -H "Content-Type: application/json" \
  -d '{
    "orderId": 10001,
    "userId": 2001,
    "productName": "iPhone 15",
    "amount": 6999.00
  }'

# 4. 观察控制台
# 生产者侧: 发送成功: topic=order-topic, partition=1, offset=0
# 消费者侧: 收到订单消息: orderId=10001, action=created, amount=6999.00

# 5. 测试批量消费
# 先发 10 条消息
for i in $(seq 1 10); do
  curl -X POST http://localhost:8080/api/order/log \
    -H "Content-Type: application/json" \
    -d "{\"orderId\": 1000$i, \"userId\": 2001, \"amount\": 99.00}"
done
# 批量消费者一次收到最多 50 条（max-poll-records 配置）