RocketMQ的四種常用消息隊列及代碼演示

更新時間：2024年01月13日 10:33:43 作者：小湯湯湯湯

這篇文章主要介紹了RocketMQ的四種常用消息隊列及代碼演示,普通消息隊列是最基本的一種消息隊列,可以按照先進先出(FIFO)的順序存儲消息,并且可以被多個消費者同時消費,可以通過在生產者端指定主題名稱和標簽來創(chuàng)建普通消息隊列,需要的朋友可以參考下

消息隊列

普通消息隊列

普通消息隊列是最基本的一種消息隊列，可以按照先進先出（FIFO）的順序存儲消息，并且可以被多個消費者同時消費。可以通過在生產者端指定主題名稱和標簽來創(chuàng)建普通消息隊列。

順序消息隊列

順序消息隊列可以保證相同主題和相同消息鍵的消息按照嚴格的順序被消費，例如可以用于訂單等需要保證處理順序的場景?？梢酝ㄟ^在創(chuàng)建普通消息隊列時指定MessageQueueSelector對象和鍵來創(chuàng)建順序消息隊列。

延遲消息隊列

延遲消息隊列是一種可以在指定時間后被消費的消息隊列?？梢栽谏a者端指定消息發(fā)送的時間戳和延遲時間，RocketMQ會根據(jù)這些信息將消息存儲到延遲消息隊列中，并在指定的時間后投遞消息到普通消息隊列中。

事務消息隊列

事務消息隊列是一種可以保證消息投遞的事務性消息隊列。在生產者端發(fā)送事務消息時，會先向RocketMQ發(fā)送一條預提交消息，然后在本地事務執(zhí)行成功后再提交或回滾事務。如果提交事務，則RocketMQ會將消息投遞到消費者，否則將不會投遞該消息?？梢酝ㄟ^在創(chuàng)建事務消息隊列時指定本地事務執(zhí)行器來創(chuàng)建事務消息隊列。

除此之外，RocketMQ還支持多主題（Topic）、多消息生產者（Producer）和多消費者組（Consumer Group）的概念，可以為不同的業(yè)務場景創(chuàng)建不同的消息隊列。

代碼演示

普通消息隊列

@Service
public class MyProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    public void sendMessage(String message) {
        rocketMQTemplate.convertAndSend("myTopic", message);
    }
}

順序消息隊列

@Service
public class MyProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    public void sendOrderMessage(String message, int orderId) {
        rocketMQTemplate.setMessageQueueSelector(new OrderMessageQueueSelector());
        rocketMQTemplate.convertAndSend("myTopic", message, orderId);
    }
}
class OrderMessageQueueSelector implements MessageQueueSelector {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message message, Object orderId) {
        int index = (int) orderId % mqs.size();
        return mqs.get(index);
    }
}

延遲消息隊列

@Service
public class MyProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    public void sendDelayMessage(String message, long delayTime) {
        rocketMQTemplate.syncSend("myTopic", MessageBuilder.withPayload(message)
                .build(), 3000, 2, delayTime);
    }
}

事務消息隊列

@Service
public class MyTransactionListener implements RocketMQLocalTransactionListener {
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        // 執(zhí)行本地事務
        // 如果本地事務執(zhí)行成功，則返回RocketMQLocalTransactionState.COMMIT
        // 如果本地事務執(zhí)行失敗，則返回RocketMQLocalTransactionState.ROLLBACK
        return RocketMQLocalTransactionState.UNKNOWN;
    }
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        // 檢查本地事務狀態(tài)
        // 如果本地事務執(zhí)行成功，則返回RocketMQLocalTransactionState.COMMIT
        // 如果本地事務執(zhí)行失敗，則返回RocketMQLocalTransactionState.ROLLBACK
        // 如果本地事務狀態(tài)未知，則返回RocketMQLocalTransactionState.UNKNOWN
        return RocketMQLocalTransactionState.UNKNOWN;
    }
}
@Service
public class MyProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private MyTransactionListener transactionListener;
    public void sendTransactionMessage(String message) {
        rocketMQTemplate.setTransactionListener(transactionListener);
        rocketMQTemplate.sendMessageInTransaction("myTransactionGroup", "myTopic",
                MessageBuilder.withpayload(message).build(), null);
}
}

到此這篇關于RocketMQ的四種常用消息隊列及代碼演示的文章就介紹到這了,更多相關RocketMQ常用消息隊列內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: