RocketMQ消息隊(duì)列實(shí)現(xiàn)隨機(jī)消息發(fā)送當(dāng)做七夕禮物
正文
都在過(guò)情人節(jié),前端的小哥哥們給女朋友畫個(gè)頁(yè)面,美美的,寫個(gè)chrome插件,好看的,俺們后端程序員咋辦。
我給媳婦寫首詩(shī),哈哈
我決定,把想對(duì)媳婦說(shuō)的,今天發(fā)送到一個(gè)MQ里邊,然后在七夕當(dāng)天,打開消費(fèi)者,將這一段話給俺媳婦看。你看,這就是我好久前對(duì)你說(shuō)的話,這就是我們后端程序員的浪漫。當(dāng)然也可以多發(fā)送幾個(gè),到時(shí)候跟根據(jù)topic
控制到底發(fā)什么,哈哈。
這里首先得用順序消息,當(dāng)然,消息過(guò)期時(shí)間得設(shè)置的長(zhǎng)一點(diǎn)。
1 下載并啟動(dòng)RocketMQ
點(diǎn)擊下載,這是個(gè)windows版本的。
下載完成解壓后長(zhǎng)這樣:
然后后還需要配置環(huán)境變量
這個(gè)時(shí)候就可以進(jìn)入到RocketMQ的bin目錄啟動(dòng)MQ了
1.1 首先啟動(dòng)name server
start mqnamesrv.cmd
1.2 然后啟動(dòng)Broker
start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true
這個(gè)時(shí)候就啟動(dòng)成功了
2 生產(chǎn)者
需要注意的是,消息必須是順序消息,不然發(fā)送的消息順序就亂了。一首情詩(shī)順序亂了,讀不下去,豈不是很尷尬。
import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.MessageQueueSelector; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageQueue; import java.util.List; public class RocketMQOrderProducer { public static void main(String[] args) throws Exception { DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name"); producer.setNamesrvAddr("127.0.0.1:9876"); producer.start(); String[] tags = new String[]{"TagA", "TagC", "TagD"}; //讀取文件 List<String> messages = getMessages(); for (int i = 0; i < messages.size(); i++) { String body = messages.get(i); Message msg = new Message("topic_luke", tags[i % tags.length], "KEY" + i, body.getBytes()); SendResult sendResult = producer.send(msg, new MessageQueueSelector() { @Override public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { Long id = ((Integer)arg).longValue(); long index = id % mqs.size(); return mqs.get((int) index); } }, i); } producer.shutdown(); } static List<String> getMessages() throws IOException { String temp = null; File f = new File("E:\Code\online-taxi-three\demo\luke.txt"); InputStreamReader read = new InputStreamReader(new FileInputStream(f)); ArrayList readList = new ArrayList(); BufferedReader reader = new BufferedReader(read); while ((temp = reader.readLine()) != null && !"".equals(temp)) { readList.add(temp); } return readList; } }
3 消費(fèi)者
這里需要注意的是setConsumeThreadMax
和setConsumeThreadMin
都需要設(shè)置成1,單線程取消息這樣就可以通過(guò)sleep控制消息的顯示速度,不然并發(fā)取消息就很快顯示完了。不夠唯美。
import lombok.SneakyThrows; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.common.message.MessageExt; import java.util.List; import java.util.concurrent.TimeUnit; public class RockerMQConsumer { public static void main(String[] args) throws Exception { //實(shí)例化消息消費(fèi)者 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_luke"); //指定nameserver地址 consumer.setNamesrvAddr("127.0.0.1:9876"); consumer.setConsumeThreadMax(1); consumer.setConsumeThreadMin(1); consumer.setPullBatchSize(1); //訂閱topic consumer.subscribe("topic_luke","*"); // 注冊(cè)回調(diào)實(shí)現(xiàn)類來(lái)處理從broker拉取回來(lái)的消息 consumer.registerMessageListener(new MessageListenerConcurrently() { @SneakyThrows @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { for (MessageExt msg : msgs) { System.out.println(new String(msg.getBody())); TimeUnit.SECONDS.sleep(3); } // 標(biāo)記該消息已經(jīng)被成功消費(fèi) return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); // 啟動(dòng)消費(fèi)者實(shí)例 consumer.start(); System.out.printf("Consumer Started.%n"); } }
發(fā)送的內(nèi)容在這里自由編寫哈,路徑和文件名能對(duì)上就行,謝謝觀看,最近突發(fā)奇想,把技術(shù)編成故事講出來(lái),會(huì)不會(huì)比較受歡迎呢。
以上就是RocketMQ消息隊(duì)列實(shí)現(xiàn)隨機(jī)消息發(fā)送當(dāng)做七夕禮物的詳細(xì)內(nèi)容,更多關(guān)于RocketMQ消息隊(duì)列隨機(jī)消息的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
spring boot 配置動(dòng)態(tài)刷新實(shí)現(xiàn)詳解
這篇文章主要介紹了spring boot 配置動(dòng)態(tài)刷新實(shí)現(xiàn)詳解,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-09-09關(guān)于MyBatis中Mapper?XML熱加載優(yōu)化
大家好,本篇文章主要講的是關(guān)于MyBatis中Mapper?XML熱加載優(yōu)化,感興趣的同學(xué)趕快來(lái)看一看吧,對(duì)你有幫助的話記得收藏一下2022-01-01SpringBoot使用Scheduling實(shí)現(xiàn)定時(shí)任務(wù)的示例代碼
Spring Boot提供了一種方便的方式來(lái)實(shí)現(xiàn)定時(shí)任務(wù),即使用Spring的@Scheduled注解,通過(guò)在方法上添加@Scheduled注解,我們可以指定方法在何時(shí)執(zhí)行,本文我們就給大家介紹一下SpringBoot如何使用Scheduling實(shí)現(xiàn)定時(shí)任務(wù),需要的朋友可以參考下2023-08-08Java實(shí)戰(zhàn)項(xiàng)目 健身管理系統(tǒng)
本文是一個(gè)Java語(yǔ)言編寫的實(shí)戰(zhàn)項(xiàng)目,是一個(gè)健身管理系統(tǒng),主要用到了ssm+springboot等技術(shù),技術(shù)含量筆記高,感興趣的童鞋跟著小編往下看吧2021-09-09Spring使用@Conditional進(jìn)行條件裝配的實(shí)現(xiàn)
在spring中有些bean需要滿足某些環(huán)境條件才創(chuàng)建某個(gè)bean,這個(gè)時(shí)候可以在bean定義上使用@Conditional注解來(lái)修飾,所以本文給大家介紹了Spring使用@Conditional進(jìn)行條件裝配的實(shí)現(xiàn),文中通過(guò)代碼示例給大家介紹的非常詳細(xì),需要的朋友可以參考下2023-12-12解決MyEclipse中的Building workspace問(wèn)題的三個(gè)方法
這篇文章主要介紹了解決MyEclipse中的Building workspace問(wèn)題的三個(gè)方法,需要的朋友可以參考下2015-11-11