亲宝软件园·资讯

展开

RocketMQ消息队列随机消息

奔跑的毛球 人气:0

正文

都在过情人节,前端的小哥哥们给女朋友画个页面,美美的,写个chrome插件,好看的,俺们后端程序员咋办。

我给媳妇写首诗,哈哈

我决定,把想对媳妇说的,今天发送到一个MQ里边,然后在七夕当天,打开消费者,将这一段话给俺媳妇看。你看,这就是我好久前对你说的话,这就是我们后端程序员的浪漫。当然也可以多发送几个,到时候跟根据topic控制到底发什么,哈哈。

这里首先得用顺序消息,当然,消息过期时间得设置的长一点。

1 下载并启动RocketMQ

点击下载,这是个windows版本的。

下载完成解压后长这样:

然后后还需要配置环境变量

这个时候就可以进入到RocketMQ的bin目录启动MQ了

1.1 首先启动name server

start mqnamesrv.cmd

1.2 然后启动Broker

start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true

这个时候就启动成功了

2 生产者

需要注意的是,消息必须是顺序消息,不然发送的消息顺序就乱了。一首情诗顺序乱了,读不下去,岂不是很尴尬。

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 消费者

这里需要注意的是setConsumeThreadMaxsetConsumeThreadMin都需要设置成1,单线程取消息这样就可以通过sleep控制消息的显示速度,不然并发取消息就很快显示完了。不够唯美。

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 {
        //实例化消息消费者
        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","*");
        // 注册回调实现类来处理从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);
                }
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

发送的内容在这里自由编写哈,路径和文件名能对上就行,谢谢观看,最近突发奇想,把技术编成故事讲出来,会不会比较受欢迎呢。

加载全部内容

相关教程
猜你喜欢
用户评论