1
Properties props = new Properties();
2
props.put("bootstrap.servers", "localhost:9092");
3
props.put("acks", "0");              // 不等确认
4
props.put("enable.idempotence", "false");
5
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
6
producer.send(new ProducerRecord<>("logs", "cpu=90%"));  // 发完就忘
7
// 消费者自动提交 offset
8
props.put("enable.auto.commit", "true");
9
props.put("auto.commit.interval.ms", "1000");  // 每秒自动提交
10
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

1
import pika
2
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
3
channel = connection.channel()
4
channel.queue_declare(queue='metrics', durable=False)  # 不持久化
5
channel.basic_consume(queue='metrics', auto_ack=True,
6
    on_message_callback=lambda ch, method, properties, body:
7
        process_metric(body)  # 处理失败？消息已经没了
8
)
9
channel.start_consuming()

1
func produceAtMostOnce(writer *kafka.Writer, topic string, msg []byte) {
2
    // 不等待写入结果，直接发
3
    writer.WriteMessages(context.Background(), kafka.Message{
4
        Topic: topic,
5
        Value: msg,
6
    })
7
    // 不检查返回值 → 消息可能丢失
8
}

1
Properties producerProps = new Properties();
2
producerProps.put("bootstrap.servers", "localhost:9092");
3
producerProps.put("acks", "all");               // 等待所有副本确认
4
producerProps.put("retries", Integer.MAX_VALUE); // 无限重试
5
producerProps.put("max.in.flight.requests.per.connection", "1"); // 保证重试不乱序
6
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
7
// 消费者手动提交
8
Properties consumerProps = new Properties();
9
consumerProps.put("bootstrap.servers", "localhost:9092");
10
consumerProps.put("enable.auto.commit", "false");  // 关闭自动提交
11
consumerProps.put("group.id", "order-service");
12
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
13
while (true) {
14
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
15
    for (ConsumerRecord<String, String> record : records) {
16
        processOrder(record.value());  // 先处理
17
    }
18
    consumer.commitSync();  // 处理完再提交 → 如果处理失败，下次重新消费
19
}

1
import pika
2
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
3
channel = connection.channel()
4
channel.queue_declare(queue='orders', durable=True)  # 持久化队列
5
channel.basic_qos(prefetch_count=1)  # 一次只拿一条
6
def callback(ch, method, properties, body):
7
    try:
8
        process_order(body)
9
        ch.basic_ack(delivery_tag=method.delivery_tag)  # 处理成功，确认
10
    except Exception as e:
11
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
12
channel.basic_consume(queue='orders', on_message_callback=callback)
13
channel.start_consuming()

1
func produceAtLeastOnce(writer *kafka.Writer, topic string, key, value []byte) error {
2
    maxRetries := 3
3
    for i := 0; i < maxRetries; i++ {
4
        err := writer.WriteMessages(context.Background(), kafka.Message{
5
            Topic: topic,
6
            Key:   key,
7
            Value: value,
8
        })
9
        if err == nil {
10
            return nil
11
        }
12
        log.Printf("发送失败，第 %d 次重试: %v", i+1, err)
13
        time.Sleep(time.Second * time.Duration(i+1)) // 指数退避
14
    }
15
    return fmt.Errorf("重试 %d 次后仍然失败", maxRetries)
16
}

1
// 非幂等：每次调用增加余额
2
void deposit(String userId, int amount) {
3
    balance += amount;  // 重复调用 → 余额错误
4
}
5
// 幂等：基于唯一ID去重
6
void deposit(String userId, String txId, int amount) {
7
    if (processedTxs.contains(txId)) return;  // 已处理，跳过
8
    balance += amount;
9
    processedTxs.add(txId);
10
}

1
-- 数据库唯一约束实现幂等
2
CREATE TABLE orders (
3
    id BIGINT PRIMARY KEY,
4
    order_id VARCHAR(64) UNIQUE,  -- 业务唯一ID
5
    amount DECIMAL(10,2),
6
    status VARCHAR(16)
7
);
8
-- 幂等插入：重复插入会被唯一约束拒绝
9
INSERT INTO orders (id, order_id, amount, status)
10
VALUES (1, 'ORD-2026-001', 99.99, 'PAID')
11
ON CONFLICT (order_id) DO NOTHING;

1
import redis
2
import json
3
r = redis.Redis(host='localhost', port=6379, db=0)
4
def process_message_idempotent(msg_id, payload):
5
    """基于 Redis SETNX 的幂等消费"""
6
    key = f"msg:processed:{msg_id}"
7
    acquired = r.set(key, "1", nx=True, ex=86400)  # 24小时过期
8
    if not acquired:
9
        print(f"消息 {msg_id} 已处理，跳过")
10
        return
11
    try:
12
        process(payload)
13
    except Exception as e:
14
        r.delete(key)  # 处理失败，删除标记，允许重试
15
        raise e

1
-- 乐观锁：版本号控制
2
CREATE TABLE account (
3
    id BIGINT PRIMARY KEY,
4
    user_id VARCHAR(64),
5
    balance DECIMAL(10,2),
6
    version INT DEFAULT 0  -- 版本号
7
);
8
-- 幂等更新：版本号必须匹配
9
UPDATE account
10
SET balance = balance + 100, version = version + 1
11
WHERE user_id = 'user-123' AND version = 5;
12
-- 如果 version 不匹配（已被其他请求更新），影响行数为 0

1
// Java: 去重表 + 业务操作在同一事务中
2
@Transactional
3
public void processOrder(Message msg) {
4
    String msgId = msg.getId();
5
    // 1. 检查去重表
6
    if (dedupRepository.existsById(msgId)) {
7
        log.info("消息 {} 已处理，跳过", msgId);
8
        return;
9
    }
10
    // 2. 执行业务逻辑
11
    orderService.createOrder(parseOrder(msg));
12
    // 3. 写入去重表（同一事务，保证原子性）
13
    dedupRepository.save(new DedupRecord(msgId, Instant.now()));
14
}

1
// 启用幂等生产者
2
Properties props = new Properties();
3
props.put("bootstrap.servers", "localhost:9092");
4
props.put("enable.idempotence", "true");  // 开启幂等
5
// 以下参数会自动设置：
6
// acks = all
7
// retries = Integer.MAX_VALUE
8
// max.in.flight.requests.per.connection <= 5
9
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
10
// 发送消息，Broker 自动去重
11
producer.send(new ProducerRecord<>("orders", "order-1", "created"));
12
producer.send(new ProducerRecord<>("orders", "order-1", "created"));  // 重复，Broker 丢弃

1
// Kafka 事务性生产者：跨分区原子写入
2
Properties props = new Properties();
3
props.put("bootstrap.servers", "localhost:9092");
4
props.put("transactional.id", "order-tx-1");  // 事务ID，跨重启保持一致
5
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
6
// 初始化事务（只执行一次）
7
producer.initTransactions();
8
try {
9
    producer.beginTransaction();
10
    // 往多个 Topic 原子写入
11
    producer.send(new ProducerRecord<>("orders", "order-1", "created"));
12
    producer.send(new ProducerRecord<>("inventory", "order-1", "reserved"));
13
    producer.send(new ProducerRecord<>("notifications", "order-1", "pending"));
14
    // 提交事务：要么全部写入，要么全部不写入
15
    producer.commitTransaction();
16
} catch (Exception e) {
17
    // 回滚事务：所有消息对消费者不可见
18
    producer.abortTransaction();
19
}

1
Properties props = new Properties();
2
props.put("application.id", "order-processing");
3
props.put("bootstrap.servers", "localhost:9092");
4
props.put("processing.guarantee", "exactly_once_v2");  // Exactly-Once 语义
5
StreamsBuilder builder = new StreamsBuilder();
6
builder.stream("orders", Consumed.with(Serdes.String(), Serdes.String()))
7
    .process(() -> new Processor<String, String>() {
8
        @Override
9
        public void process(Record<String, String> record) {
10
            // 1. 消费订单消息
11
            Order order = parseOrder(record.value());
12
            // 2. 处理业务逻辑
13
            InventoryResult result = inventoryService.deduct(order);
14
            // 3. 输出结果（在同一事务中）
15
            context().forward(new Record<>(order.getId(), result.toString()));
16
        }
17
    })
18
    .to("inventory-results");
19
KafkaStreams streams = new KafkaStreams(builder.build(), props);
20
streams.start();

1
// 手动实现消费端 Exactly-Once
2
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
3
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
4
consumer.subscribe(Collections.singletonList("orders"));
5
while (true) {
6
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
7
    if (records.isEmpty()) continue;
8
    producer.beginTransaction();
9
    try {
10
        for (ConsumerRecord<String, String> record : records) {
11
            // 处理消息
12
            String result = processOrder(record.value());
13
            // 输出结果
14
            producer.send(new ProducerRecord<>("results", record.key(), result));
15
        }
16
        // 在同一事务中提交 offset
17
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
18
        for (ConsumerRecord<String, String> record : records) {
19
            offsets.put(new TopicPartition(record.topic(), record.partition()),
20
                       new OffsetAndMetadata(record.offset() + 1));
21
        }
22
        producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
23
        producer.commitTransaction();
24
    } catch (Exception e) {
25
        producer.abortTransaction();
26
    }
27
}

1
// RocketMQ 事务消息完整实现
2
TransactionMQProducer producer = new TransactionMQProducer("order-producer-group");
3
producer.setNamesrvAddr("localhost:9876");
4
// 事务回查线程池
5
ExecutorService executorService = new ThreadPoolExecutor(
6
    2, 5, 100, TimeUnit.SECONDS,
7
    new ArrayBlockingQueue<>(2000),
8
    r -> new Thread(r, "transaction-check-thread")
9
);
10
producer.setExecutorService(executorService);
11
// 事务监听器
12
producer.setTransactionListener(new TransactionListener() {
13
    // 执行本地事务
14
    @Override
15
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
16
        String orderId = msg.getKeys();
17
        try {
18
            // 本地事务：创建订单
19
            Order order = parseOrder(msg.getBody());
20
            orderService.createOrder(order);
21
            log.info("订单 {} 创建成功，提交消息", orderId);
22
            return LocalTransactionState.COMMIT_MESSAGE;
23
        } catch (DuplicateKeyException e) {
24
            // 订单已存在（幂等），提交消息
25
            log.info("订单 {} 已存在，提交消息", orderId);
26
            return LocalTransactionState.COMMIT_MESSAGE;
27
        } catch (Exception e) {
28
            log.error("订单 {} 创建失败，回滚消息", orderId, e);
29
            return LocalTransactionState.ROLLBACK_MESSAGE;
30
        }
31
    }
32
    // 回查本地事务状态
33
    @Override
34
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
35
        String orderId = msg.getKeys();
36
        Order order = orderService.getById(orderId);
37
        if (order != null) {
38
            log.info("回查：订单 {} 存在，提交消息", orderId);
39
            return LocalTransactionState.COMMIT_MESSAGE;
40
        } else {
41
            log.warn("回查：订单 {} 不存在，回滚消息", orderId);
42
            return LocalTransactionState.ROLLBACK_MESSAGE;
43
        }
44
    }
45
});
46
producer.start();
47
// 发送事务消息
48
Message msg = new Message("order-events", "TagA",
49
    "{\"orderId\":\"ORD-001\",\"amount\":99.9}".getBytes());
50
msg.setKeys("ORD-001");  // 业务唯一ID，用于回查
51
producer.sendMessageInTransaction(msg, null);

1
import pika
2
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
3
channel = connection.channel()
4
channel.confirm_delivery()
5
try:
6
    channel.basic_publish(
7
        exchange='',
8
        routing_key='orders',
9
        body='order-123',
10
        properties=pika.BasicProperties(delivery_mode=2)  # 持久化
11
    )
12
    print("消息确认到达 Broker")
13
except pika.exceptions.UnroutableError:
14
    print("消息无法路由，可能丢失")

1
// Java: RabbitMQ 异步确认
2
channel.confirmSelect();  // 开启确认模式
3
// 添加确认回调
4
channel.addConfirmListener(new ConfirmListener() {
5
    @Override
6
    public void handleAck(long deliveryTag, boolean multiple) {
7
        // 消息确认到达 Broker
8
        pendingMessages.remove(deliveryTag);
9
    }
10
    @Override
11
    public void handleNack(long deliveryTag, boolean multiple) {
12
        // 消息被拒绝，需要重发
13
        Message msg = pendingMessages.get(deliveryTag);
14
        resend(msg);
15
    }
16
});

1
import pika
2
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
3
channel = connection.channel()
4
channel.exchange_declare(exchange='dlx', exchange_type='direct')
5
channel.queue_declare(queue='orders-dlq')  # 死信队列
6
channel.queue_bind(queue='orders-dlq', exchange='dlx', routing_key='orders')
7
args = {
8
    'x-dead-letter-exchange': 'dlx',
9
    'x-dead-letter-routing-key': 'orders',
10
    'x-message-ttl': 60000,  # 消息 TTL 60秒
11
}
12
channel.queue_declare(queue='orders', durable=True, arguments=args)
13
max_retries = 3
14
def callback(ch, method, properties, body):
15
    headers = properties.headers or {}
16
    retry_count = headers.get('x-retry-count', 0)
17
    try:
18
        process_order(body)
19
        ch.basic_ack(delivery_tag=method.delivery_tag)
20
    except Exception as e:
21
        if retry_count < max_retries:
22
            ch.basic_publish(
23
                exchange='',
24
                routing_key='orders',
25
                body=body,
26
                properties=pika.BasicProperties(
27
                    delivery_mode=2,
28
                    headers={'x-retry-count': retry_count + 1}
29
                )
30
            )
31
            ch.basic_ack(delivery_tag=method.delivery_tag)  # 确认原消息
32
        else:
33
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
34
            log.error(f"消息处理失败，进入死信队列: {body}")
35
channel.basic_qos(prefetch_count=1)
36
channel.basic_consume(queue='orders', on_message_callback=callback)
37
channel.start_consuming()