在電商等系統中，下單后需要進行一系列的處理，包括庫存扣減、支付通知等。可以將訂單相關的消息發送到消息隊列中，由消費者異步地進行處理，從而提高系統的處理能力和可靠性。
Spring Boot 是一個基于 Spring 框架的快速開發框架，而 RabbitMQ 和 RocketMQ 則是常用的消息隊列中間件。下面是它們常用的一些用法和場景。
1、訂單處理
在電商等系統中，下單后需要進行一系列的處理，包括庫存扣減、支付通知等。可以將訂單相關的消息發送到消息隊列中，由消費者異步地進行處理，從而提高系統的處理能力和可靠性。
使用 RabbitMQ 實現訂單處理的示例代碼：
// 發送訂單消息
rabbitTemplate.convertAndSend("order-exchange", "order-routing-key", order);
// 消費訂單消息
@RabbitListener(queues = "order-queue")
public void handleOrderMessage(Order order) {
// 處理訂單相關的業務邏輯
}
使用 RocketMQ 實現訂單處理的示例代碼：
// 發送訂單消息
DefaultMQProducer producer = new DefaultMQProducer("order-producer-group");
producer.setNamesrvAddr("localhost:9876");
Message message = new Message("order-topic", "order-tag", order.toString().getBytes());
producer.send(message);
// 消費訂單消息
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order-consumer-group");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("order-topic", "order-tag");
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
for (MessageExt message : messages) {
// 處理訂單相關的業務邏輯
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
2、日志處理
在分布式系統中，各個節點產生的日志需要進行集中處理和存儲，以便后續的分析和監控。可以使用消息隊列將日志消息發送到中心化的日志系統中。
使用 RabbitMQ 實現日志處理的示例代碼：
// 發送日志消息
rabbitTemplate.convertAndSend("log-exchange", "log-routing-key", log);
// 消費日志消息
@RabbitListener(queues = "log-queue")
public void handleLogMessage(Log log) {
// 處理日志相關的業務邏輯
}
使用 RocketMQ 實現日志處理的示例代碼：
// 發送日志消息
DefaultMQProducer producer = new DefaultMQProducer("log-producer-group");
producer.setNamesrvAddr("localhost:9876");
Message message = new Message("log-topic", "log-tag", log.toString().getBytes());
producer.send(message);
// 消費日志消息
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("log-consumer-group");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("log-topic", "log-tag");
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
for (MessageExt message : messages) {
// 處理日志相關的業務邏輯
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
3、分布式事務
在分布式系統中，不同的服務之間需要進行事務管理，以保證數據的一致性和可靠性。可以使用消息隊列來實現分布式事務的消息確認和回滾。
使用 RabbitMQ 實現分布式事務的示例代碼：
// 發送訂單消息和庫存消息
rabbitTemplate.invoke(new RabbitCallback<Void>() {
public Void doInRabbit(Channel channel) throws Exception {
channel.txSelect();
channel.basicPublish("order-exchange", "order-routing-key", null, order.toString().getBytes());
channel.basicPublish("stock-exchange", "stock-routing-key", null, stock.toString().getBytes());
channel.txCommit();
return null;
}
});
// 消費訂單消息和庫存消息
@RabbitListener(queues = "order-queue")
@Transactional
public void handleOrderMessage(Order order) {
// 處理訂單相關的業務邏輯
}
@RabbitListener(queues = "stock-queue")
@Transactional
public void handleStockMessage(Stock stock) {
// 處理庫存相關的業務邏輯
}
使用 RocketMQ 實現分布式事務的示例代碼：
// 發送訂單消息和庫存消息
TransactionMQProducer producer = new TransactionMQProducer("transaction-producer-group");
producer.setNamesrvAddr("localhost:9876");
producer.setTransactionListener(new TransactionListener() {
public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
// 執行本地事務
return LocalTransactionState.COMMIT_MESSAGE;
}
public LocalTransactionState checkLocalTransaction(MessageExt message) {
// 檢查本地事務狀態
return LocalTransactionState.COMMIT_MESSAGE;
}
});
Message message1 = new Message("order-topic", "order-tag", order.toString().getBytes());
Message message2 = new Message("stock-topic", "stock-tag", stock.toString().getBytes());
TransactionSendResult result = producer.sendMessageInTransaction(Arrays.asList(message1, message2), null);
// 消費訂單消息和庫存消息
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transaction-consumer-group");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("order-topic", "order-tag");
consumer.subscribe("stock-topic", "stock-tag");
consumer.registerMessageListener(new MessageListenerOrderly() {
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messages, ConsumeOrderlyContext context) {
for (MessageExt message : messages) {
// 處理訂單或庫存相關的業務邏輯
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
4、消息隊列比較
RabbitMQ 和 RocketMQ 都是常用的消息隊列中間件，它們在特點和使用上有些區別。
- RabbitMQ 采用 AMQP（高級消息隊列協議），支持多種編程語言和多種操作系統。RabbitMQ 的消息傳輸可靠性較高，但性能較低。
- RocketMQ 采用自定義的協議，適合 Java 應用，支持高吞吐量和高可用性。RocketMQ 的消息傳輸可靠性較低，但性能較高。
