RocketMQ怎么保证消息不丢失

1.生产者端

（1）方式1：同步发送//

适用于重要消息，需要确保消息一定到达Broker
public void sendImportantMessage() {
Message msg = new Message(“TopicTest”, “TagA”, “Important message”.getBytes());
try {
SendResult sendResult = producer.send(msg);
// 等待发送结果
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
// 发送失败处理
handleSendFail(msg);
}
} catch (Exception e) {
// 异常重试处理
handleRetry(msg);
}
}

2）方式2：开启事务消息机制

// 适用于需要保证本地事务和消息发送原子性的场景
// 例如：下单扣减库存，确保库存扣减成功才发送订单消息
public void sendTransactionMessage() {
TransactionMQProducer producer = new TransactionMQProducer(“tx_producer_group”);
producer.setTransactionListener(new TransactionListener() {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
try {
// 执行本地事务（如扣减库存）
boolean result = deductInventory();
if (result) {
return LocalTransactionState.COMMIT_MESSAGE;
} else {
return LocalTransactionState.ROLLBACK_MESSAGE;
}
} catch (Exception e) {
return LocalTransactionState.UNKNOW;
}
}

@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
// 回查本地事务状态
return checkTransactionStatus(msg);
}
});
}

选择建议：

（1）一般业务消息：使用同步发送即可

（2）分布式事务场景：使用事务消息

（3）非重要消息：可以使用异步发送