快捷導(dǎo)航

RocketMQ實(shí)現(xiàn)隨緣分BUG小功能示例詳解

更新時(shí)間：2022年08月10日 15:47:47 作者：奔跑的毛球

這篇文章主要為大家介紹了RocketMQ實(shí)現(xiàn)隨緣分BUG小功能示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

正文

以前公司的產(chǎn)品已經(jīng)上線20多年了，主要是維護(hù)，也就是改bug。每周我們Team會(huì)從Jira上拿我們可以改的bug，因?yàn)槊總€(gè)團(tuán)隊(duì)負(fù)責(zé)的業(yè)務(wù)范圍不一樣，我們團(tuán)隊(duì)只能改我們自己業(yè)務(wù)范圍的。這樣每周大概有20個(gè)左右的新bug，假如團(tuán)隊(duì)一共10個(gè)人，那么均分就是每人兩個(gè)，改完下班。

但是這BUG肯定有難有簡(jiǎn)單，大家肯定都愿意改簡(jiǎn)單的，在家辦公，任務(wù)量完了不就等于放假么。開(kāi)始是自己給自己搶?zhuān)瓦?，是歐美項(xiàng)目，所以客服晚上報(bào)出來(lái)的bug多。有的哥們?cè)缟?點(diǎn)起來(lái)看有沒(méi)有新的簡(jiǎn)單的。

這對(duì)那些反應(yīng)慢的兄弟不公平。領(lǐng)導(dǎo)分的話，就是能者多勞了，累的累死，閑的閑死。所以我們就需要一個(gè)隨機(jī)的分bug小工具，正所謂聽(tīng)天由命,富貴在天。

實(shí)現(xiàn)過(guò)程

首先來(lái)n個(gè)消費(fèi)者，n就是需要分的人數(shù)。

這里需要將這些消費(fèi)者全部放入同一個(gè)group,這樣這些bug會(huì)均分給這些消費(fèi)者，而且當(dāng)某個(gè)同事請(qǐng)假?zèng)]有來(lái)的時(shí)候，那么就不啟動(dòng)自己的消費(fèi)者，這樣就是其他小伙伴均分。

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.setPullBatchSize(1);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //訂閱topic
        consumer.subscribe("topic_luke","*");
        // 注冊(cè)回調(diào)實(shí)現(xiàn)類(lèi)來(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");
    }
}

上邊的程序啟動(dòng)N個(gè)就行

生產(chǎn)者：

public class RocketMQProducer {
    public static void main(String[] args) throws Exception {
        //實(shí)例化消息生產(chǎn)者對(duì)象
        DefaultMQProducer producer = new DefaultMQProducer("group_luke");
        //設(shè)置消息過(guò)期時(shí)間
        producer.setSendMsgTimeout(300000);
        //設(shè)置NameSever地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //啟動(dòng)Producer實(shí)例
        producer.start();
        List<String> messages = new ArrayList<>();
        messages.add("WFD-11356");//添加bug號(hào)
        messages.add("WFD-11357");//添加bug號(hào)
        messages.add("WFD-11358");//添加bug號(hào)
        for (String message : messages) {
            //創(chuàng)建消息 指定topic和消息體
            Message msg = new Message("topic_luke", "tag", message.getBytes(StandardCharsets.UTF_8));
            System.out.println(msg.toString());
            //發(fā)送消息
            producer.sendOneway(msg);
        }
        //關(guān)閉producer
        producer.shutdown();
    }

當(dāng)生產(chǎn)者啟動(dòng)，將消息發(fā)送到RocketMQ的時(shí)候，各個(gè)消費(fèi)者將收到消息，這也就決定了這周改哪個(gè)bug。