RocketMQ实现随缘分BUG小功能示例详解

2022-11-13 14:47

短信预约 -IT技能 免费直播动态提醒

正文

以前公司的产品已经上线20多年了，主要是维护，也就是改bug。每周我们Team会从Jira上拿我们可以改的bug，因为每个团队负责的业务范围不一样，我们团队只能改我们自己业务范围的。这样每周大概有20个左右的新bug，假如团队一共10个人，那么均分就是每人两个，改完下班。

但是这BUG肯定有难有简单，大家肯定都愿意改简单的，在家办公，任务量完了不就等于放假么。开始是自己给自己抢，就忒卷，是欧美项目，所以客服晚上报出来的bug多。有的哥们早上5点起来看有没有新的简单的。

这对那些反应慢的兄弟不公平。领导分的话，就是能者多劳了，累的累死，闲的闲死。所以我们就需要一个随机的分bug小工具，正所谓听天由命,富贵在天。

实现过程

首先来n个消费者，n就是需要分的人数。

这里需要将这些消费者全部放入同一个group,这样这些bug会均分给这些消费者，而且当某个同事请假没有来的时候，那么就不启动自己的消费者，这样就是其他小伙伴均分。

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.setPullBatchSize(1);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //订阅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");
    }
}

上边的程序启动N个就行

生产者：

public class RocketMQProducer {
    public static void main(String[] args) throws Exception {
        //实例化消息生产者对象
        DefaultMQProducer producer = new DefaultMQProducer("group_luke");
        //设置消息过期时间
        producer.setSendMsgTimeout(300000);
        //设置NameSever地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //启动Producer实例
        producer.start();
        List<String> messages = new ArrayList<>();
        messages.add("WFD-11356");//添加bug号
        messages.add("WFD-11357");//添加bug号
        messages.add("WFD-11358");//添加bug号
        for (String message : messages) {
            //创建消息 指定topic和消息体
            Message msg = new Message("topic_luke", "tag", message.getBytes(StandardCharsets.UTF_8));
            System.out.println(msg.toString());
            //发送消息
            producer.sendOneway(msg);
        }
        //关闭producer
        producer.shutdown();
    }

当生产者启动，将消息发送到RocketMQ的时候，各个消费者将收到消息，这也就决定了这周改哪个bug。