// ===== 最重要的配置 =====
props.put(ProducerConfig.ACKS_CONFIG, "all");

// 失败重试次数（Kafka 2.1+ 默认 Integer.MAX_VALUE，配合 delivery.timeout.ms 使用）
props.put(ProducerConfig.RETRIES_CONFIG, 3);

// 重试间隔（默认 100ms）
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 200);

// 消息发送最大超时时间（默认 120000ms = 2分钟）
// retries 用尽前，只要在这个时间内，会一直重试
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000);

// 开启幂等性（Kafka 3.0+ 默认开启）
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

// 开启幂等性会自动设置：
// acks = all
// retries > 0
// max.in.flight.requests.per.connection <= 5

// 批量发送的最大字节数（默认 16KB）
// 消息积累到 batch.size 后立即发送
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768); // 32KB

// 等待更多消息的最长时间（默认 0ms，即不等待）
// 设置 5-10ms 可显著提升吞吐量，略微增加延迟
props.put(ProducerConfig.LINGER_MS_CONFIG, 5);

// 发送缓冲区总大小（默认 32MB）
// 当所有 Batch 总大小达到此值，send() 会阻塞
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // 32MB

// 缓冲区满后的最长等待时间（默认 60000ms）
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000);

// 压缩算法（默认 none）
// 推荐 lz4（平衡压缩率和性能），大数据场景用 zstd
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

// 完整的生产推荐配置
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092,kafka3:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// 可靠性（核心业务用 all，日志用 1）
props.put(ProducerConfig.ACKS_CONFIG, "all");

// 幂等（推荐开启）
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

// 性能（根据业务延迟要求调整）
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);      // 32KB
props.put(ProducerConfig.LINGER_MS_CONFIG, 5);           // 5ms
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864); // 64MB
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

// 消费者组 ID（相同 group.id 共享消费）
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");

// 没有初始 offset 时的策略
// earliest: 从最早的消息开始消费（不会遗漏）
// latest: 从最新的消息开始消费（会遗漏已有消息）
// none: 没有 offset 就抛异常
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

// ===== 核心选择：自动提交 vs 手动提交 =====

// 自动提交（不推荐生产使用）
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000"); // 每1秒提交一次

// 手动提交（推荐）
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
// 代码中通过 consumer.commitSync() 或 consumer.commitAsync() 提交

// 单次 poll 最多返回的消息数（默认 500）
// 根据每条消息处理时间调整：处理慢则调小，处理快则调大
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);

// 两次 poll 之间的最长间隔（默认 5 分钟）
// 如果业务处理时间超过此值，Consumer 会被踢出 Group（触发 Rebalance）
// 一定要大于批量处理最大耗时！
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // 5分钟

// Consumer 与 Broker 的心跳超时（默认 10 秒）
// Consumer 在此时间内没发心跳会被认为已死，触发 Rebalance
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);

// 心跳间隔（默认 3 秒，必须小于 session.timeout.ms / 3）
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 10000);

// 单次 fetch 的最少字节数（默认 1 字节，即有消息就返回）
// 增大可以减少网络请求次数（提升吞吐），但会增加延迟
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024); // 1KB

// 等待 fetch.min.bytes 满足的最长时间（默认 500ms）
// 超过此时间后即使不满足 fetch.min.bytes 也会返回
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);

// 单次 fetch 的最大字节数（默认 50MB）
props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 52428800); // 50MB

// 生产推荐配置（手动提交 + 稳定不 Rebalance）
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1:9092,kafka2:9092,kafka3:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-consumer-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

// Offset 策略
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

// 性能与稳定性（根据业务处理速度调整）
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 200);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 10000);