Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 关键配置：为了追求极致吞吐，设置了 acks=0
props.put("acks", "0");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("order-events", orderId, orderEventJson);
// 使用异步发送，不关心结果
producer.send(record);
producer.close();

props.put("acks", "all"); // 最强持久性保证
props.put("retries", 3); // 合理重试次数
props.put("delivery.timeout.ms", 30000); // 总交付超时时间
props.put("max.in.flight.requests.per.connection", 1); // 为开启幂等性，或保证分区内顺序时可能需要设为1

// 发送并获取Future，可阻塞等待或添加回调
Future<RecordMetadata> future = producer.send(record);
// 方式一：同步等待（影响吞吐，可根据业务选择）
// RecordMetadata metadata = future.get();
// 方式二：异步回调
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        log.error("消息发送失败: {}", exception.getMessage());
        // 此处应加入业务级的补偿逻辑，如落本地库、告警等
    } else {
        log.info("消息发送成功，主题:{}，分区:{}，偏移量:{}",
                 metadata.topic(), metadata.partition(), metadata.offset());
    }
});

Properties props = new Properties();
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000"); // 5秒自动提交一次

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("order-events"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        try {
            // 处理业务逻辑：扣减库存、创建支付单
            processOrderEvent(record.value());
        } catch (Exception e) {
            log.error("处理消息失败，跳过", e);
            // 注意：此处发生异常，但偏移量依然会在后续被提交！
        }
    }
}

props.put("enable.auto.commit", "false"); // 关闭自动提交

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        try {
            // 1. 处理业务逻辑（要求幂等）
            processOrderEvent(record.value());
            // 2. 处理成功，准备提交偏移量。这里采用异步提交提高性能，可配合回调处理错误。
            consumer.commitAsync();
        } catch (BusinessException e) {
            // 3. 业务逻辑失败，记录日志并跳入死信队列。不应提交当前偏移量。
            log.error("业务处理失败，消息转入死信队列: {}", record.value(), e);
            sendToDlq(record);
            // 可以选择提交偏移量，跳过此条坏消息，但必须确保死信队列处理可靠。
            consumer.commitAsync();
        } catch (Exception e) {
            // 4. 系统异常（如DB连接失败），应停止消费，等待恢复，而不是提交偏移量。
            log.error("系统异常，暂停消费", e);
            break; // 或进入重试循环
        }
    }
}

// 生产者配置
props.put("enable.idempotence", true); // 幂等性是事务的基础
props.put("transactional.id", "unique-order-transaction-id"); // 唯一事务ID，用于故障恢复

producer.initTransactions(); // 初始化事务

try {
    producer.beginTransaction();
    // 发送订单创建事件
    producer.send(new ProducerRecord<>("order-events", orderEvent));
    // 发送关联的物流事件（到不同主题或分区）
    producer.send(new ProducerRecord<>("logistics-events", logisticsEvent));
    // 可以结合消费者偏移量提交（消费-处理-生产模式）
    producer.sendOffsetsToTransaction(currentOffsets, consumerGroupId);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
    throw e;
}