RocketMQ的四種常用消息隊列及代碼演示
消息隊列
普通消息隊列
普通消息隊列是最基本的一種消息隊列,可以按照先進先出(FIFO)的順序存儲消息,并且可以被多個消費者同時消費??梢酝ㄟ^在生產(chǎn)者端指定主題名稱和標簽來創(chuàng)建普通消息隊列。
順序消息隊列
順序消息隊列可以保證相同主題和相同消息鍵的消息按照嚴格的順序被消費,例如可以用于訂單等需要保證處理順序的場景??梢酝ㄟ^在創(chuàng)建普通消息隊列時指定MessageQueueSelector對象和鍵來創(chuàng)建順序消息隊列。
延遲消息隊列
延遲消息隊列是一種可以在指定時間后被消費的消息隊列??梢栽谏a(chǎn)者端指定消息發(fā)送的時間戳和延遲時間,RocketMQ會根據(jù)這些信息將消息存儲到延遲消息隊列中,并在指定的時間后投遞消息到普通消息隊列中。
事務消息隊列
事務消息隊列是一種可以保證消息投遞的事務性消息隊列。在生產(chǎn)者端發(fā)送事務消息時,會先向RocketMQ發(fā)送一條預提交消息,然后在本地事務執(zhí)行成功后再提交或回滾事務。如果提交事務,則RocketMQ會將消息投遞到消費者,否則將不會投遞該消息??梢酝ㄟ^在創(chuàng)建事務消息隊列時指定本地事務執(zhí)行器來創(chuàng)建事務消息隊列。
除此之外,RocketMQ還支持多主題(Topic)、多消息生產(chǎn)者(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常用消息隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!
相關文章
SpringCloud+nacos部署在多ip環(huán)境下統(tǒng)一nacos服務注冊ip(親測有效)
在部署SpringCoud項目的時候分服務器部署注冊同一個nacos服務,但是在服務器有多個ip存在的同時(內(nèi)外網(wǎng)),就會出現(xiàn)注冊服務ip不同的問題,導致一些接口無法連接訪問,經(jīng)過多次排查終于找到問題并找到解決方法,需要的朋友可以參考下2023-04-04
springboot發(fā)送郵件功能的實現(xiàn)代碼
發(fā)郵件是一個很常見的功能,在java中實現(xiàn)需要依靠JavaMailSender這個接口,今天通過本文給大家分享springboot發(fā)送郵件功能的實現(xiàn)代碼,感興趣的朋友跟隨小編一起看看吧2021-07-07
Apache?Commons?BeanUtils:?JavaBean操作方法
這篇文章主要介紹了Apache?Commons?BeanUtils:?JavaBean操作的藝術,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2023-12-12
idea項目debug模式啟動,斷點失效,斷點紅點內(nèi)無對勾問題及解決
這篇文章主要介紹了idea項目debug模式啟動,斷點失效,斷點紅點內(nèi)無對勾問題及解決方案,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-10-10
SpringBoot解決循環(huán)調(diào)用問題
作者在將SpringBoot從1.5版本升級至2.6版本,并遷移至阿里云上運行后,遇到了循環(huán)調(diào)用問題,在Jetty容器中運行沒問題,但在Tomcat容器中就出現(xiàn)了循環(huán)引用問題,原因是SpringBoot 2.6不鼓勵循環(huán)引用,暴露出該問題,作者提供了兩種解決思路2024-10-10

