Springboot：集成RocketMQ使用延迟消息队列

深碍√TFBOYSˉ_ 2023-02-20 04:56 34阅读 0赞

*  application.yml配置文件

rocketmq:
      consumer:
        consumer-group: TestConsumerGroup
      producer:
        producer-group: TestProducerGroup
      namesrv-addr: 127.0.0.1:9876

*  ProducerSchedule 生产者

@Component
    public class ProducerSchedule { 
    
        private DefaultMQProducer producer;
    
        @Value("${rocketmq.producer.producer-group}")
        private String producerGroup;
    
        @Value("${rocketmq.namesrv-addr}")
        private String namesrvAddr;
    
        public ProducerSchedule() { 
    
        }
    
        @PostConstruct
        public void defaultMQProducer() { 
            if (this.producer == null) { 
                this.producer = new DefaultMQProducer(this.producerGroup);
                this.producer.setNamesrvAddr(this.namesrvAddr);
            }
            try { 
                this.producer.start();
                System.out.println("-------producer start");
            } catch (MQClientException e) { 
                e.printStackTrace();
            }
        }
    
        public String send(String topic, String messageText) throws Exception { 
            Message message = new Message(topic, messageText.getBytes());
    		// messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    		// 这里设置4，即30s的延迟
            message.setDelayTimeLevel(4);
    
            SendResult result = this.producer.send(message);
            System.out.println(result.getMsgId());
            System.out.println(result.getSendStatus());
            return result.getMsgId();
        }
    
    
    }

*  ConsumerSchedule 消费者

@Component
    public class ConsumerSchedule implements CommandLineRunner { 
    
        @Value("${rocketmq.consumer.consumer-group}")
        private String consumerGroup;
    
        @Value("${rocketmq.namesrv-addr}")
        private String namesrvAddr;
    
        public void messageListener() throws MQClientException { 
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
    
            consumer.setNamesrvAddr(namesrvAddr);
    
            consumer.subscribe("TopicTest", "*");
    
            // 每次消费的消息数
            consumer.setConsumeMessageBatchMaxSize(1);
    
            consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> { 
                for (Message message : messages) { 
                    System.out.println("消息：" + new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
    
            consumer.start();
        }
    
    
        @Override
        public void run(String... args) throws Exception { 
            this.messageListener();
        }
    }

*  测试类

@Autowired
    private ProducerSchedule producerSchedule;
        
    public void pushMessageToMQ() throws Exception { 
        producerSchedule.send("TopicTest", "test");
    }