RocketMQ消息隊(duì)列實(shí)現(xiàn)隨機(jī)消息發(fā)送當(dāng)做七夕禮物

更新時(shí)間：2022年08月22日 14:57:13 作者：奔跑的毛球

這篇文章主要為大家介紹了RocketMQ消息隊(duì)列實(shí)現(xiàn)隨機(jī)消息發(fā)送當(dāng)做七夕禮物，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

正文

都在過情人節(jié)，前端的小哥哥們給女朋友畫個(gè)頁面，美美的，寫個(gè)chrome插件，好看的，俺們后端程序員咋辦。

我給媳婦寫首詩，哈哈

我決定，把想對(duì)媳婦說的，今天發(fā)送到一個(gè)MQ里邊，然后在七夕當(dāng)天，打開消費(fèi)者，將這一段話給俺媳婦看。你看，這就是我好久前對(duì)你說的話，這就是我們后端程序員的浪漫。當(dāng)然也可以多發(fā)送幾個(gè)，到時(shí)候跟根據(jù)topic控制到底發(fā)什么，哈哈。

這里首先得用順序消息，當(dāng)然，消息過期時(shí)間得設(shè)置的長一點(diǎn)。

1 下載并啟動(dòng)RocketMQ

點(diǎn)擊下載，這是個(gè)windows版本的。

下載完成解壓后長這樣：

然后后還需要配置環(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ā)送的消息順序就亂了。一首情詩順序亂了，讀不下去，豈不是很尷尬。

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，單線程取消息這樣就可以通過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)類來處理從broker拉取回來的消息
        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ù)編成故事講出來，會(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)文章！

您可能感興趣的文章: