RocketMQ实现随缘分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");
    }
}

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();
    }