Java代码实现RocketMQ消息发送接收

末蓝、 2022-11-19 05:23 272阅读 0赞

配置RocketMQ参考：[https://blog.csdn.net/qq\_43037478/article/details/115670879][https_blog.csdn.net_qq_43037478_article_details_115670879]

所需的jar包

<dependency>
    			<groupId>org.apache.rocketmq</groupId>
    			<artifactId>rocketmq-common</artifactId>
    			<version>4.2.0</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.rocketmq</groupId>
    			<artifactId>rocketmq-client</artifactId>
    			<version>4.2.0</version>
    		</dependency>
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>fastjson</artifactId>
    			<version>1.2.29</version>
    		</dependency>

application.yml配置文件代码

rocketmq:
      consumer:
        #RocketMQ的请求地址
        namesrvAddr: 你配置的ip:9876
        #发送同一类消息的设置为同一个group，保证唯一,默认不需要设置，rocketmq会使用ip@pid(pid代表jvm名字)作为唯一标示
        groupName: test-demo
        #如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
        instanceName: consumer.demo
      producer:
        #发送消息超时时间
        sendMsgTimeout: 10000
        #maxMessageSize最大消息限制
        maxMessageSize: 999999999
        #消息大小超出配置对消息压缩
        compressOver: 40000
      topic: test-demo
      #不配置为订阅全部tag,多个行业用 "||" 分隔, 如："0||1||2"
      tag: "111"

配置producer和consumer

package com.example.demo.RocketMQ;
    
    import com.alibaba.fastjson.JSONException;
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class SpringConfig {
    
        @Value("${rocketmq.consumer.groupName}")
        private String groupName;
        @Value("${rocketmq.consumer.namesrvAddr}")
        private String namesrvAddr;
        @Value("${rocketmq.consumer.instanceName}")
        private String instanceName;
        @Value("${rocketmq.producer.sendMsgTimeout}")
        private int sendMsgTimeout;
        @Value("${rocketmq.producer.maxMessageSize}")
        private int maxMessageSize;
        @Value("${rocketmq.producer.compressOver}")
        private int compressOver;
        @Value("${rocketmq.topic}")
        private String topic;
        @Value("${rocketmq.tag}")
        private String tag;
    
        @Bean
        public DefaultMQProducer getRocketMQProducer() {
            DefaultMQProducer producer;
            producer = new DefaultMQProducer(this.groupName);
            producer.setNamesrvAddr(this.namesrvAddr);
            producer.setInstanceName(this.instanceName);
            producer.setSendMsgTimeout(this.sendMsgTimeout);
            producer.setCompressMsgBodyOverHowmuch(this.compressOver);
            producer.setMaxMessageSize(this.maxMessageSize);
            producer.setVipChannelEnabled(false);
    
            try {
                producer.start();
            } catch (MQClientException e) {
                System.out.println(e);
            }
            return producer;
        }
    
    
        @Bean
        public DefaultMQPushConsumer getRocketMQConsumer() {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.groupName);
            consumer.setNamesrvAddr(this.namesrvAddr);
            consumer.setInstanceName(this.instanceName);
            consumer.setConsumeMessageBatchMaxSize(1);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.setConsumeConcurrentlyMaxSpan(2000);  //单队列并行消费最大跨度，用于流控
            consumer.setPullThresholdForQueue(1000);       // 一个queue最大消费的消息个数，用于流控
            consumer.setPullInterval(1000);                //消息拉取时间间隔，默认为0，即拉完一次立马拉第二次，单位毫秒
            consumer.setMessageModel(MessageModel.CLUSTERING);  //消费模式，集群消费
    
            try {
                consumer.subscribe(this.topic, tag);
                consumer.registerMessageListener((MessageListenerConcurrently) (msgList, consumeConcurrentlyContext) -> {
                    try {
                        Message msg = null;
                        for (Message aMsgList : msgList) {
                            msg = aMsgList;
                            System.out.println("收到MQ消息："+msg);
                        }
    
                    } catch (JSONException e) {
                        System.out.println(e);
                    }
    
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                });
                consumer.start();
                System.out.println("已启动Conusmer【gruop:" + this.groupName + "，instance:" + this.instanceName
                        + "】，监听TOPIC-{" + this.topic + "},tag-{" + this.tag + "}");
            } catch (MQClientException e) {
                System.out.println(e);
            }
            return consumer;
        }
    }

发送消息测试，每10秒发送一条消息

package com.example.demo.RocketMQ;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.common.message.Message;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import java.util.UUID;
    
    @Component
    public class Task {
        @Resource
        private DefaultMQProducer producer;
    
        /**
         * 每10秒执行一次
         */
        @Scheduled(cron = "0/10 * *  * * ?")
        private void sendMsgToMq() {
            String str = "发送测试消息";
            Message msg;
            try {
                msg = new Message("test-demo"
                        , "111"
                        , UUID.randomUUID().toString()
                        , str.getBytes("utf-8"));
                SendResult result = producer.send(msg);
                if (result.getSendStatus() == SendStatus.SEND_OK) {
                    System.out.println("消息发送成功");
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

由于使用了定时器，需要在启动类上添加注解@EnableScheduling（开启定时任务）

@SpringBootApplication
    @EnableScheduling
    public class DemoApplication {
    
    	public static void main(String[] args) {
    		SpringApplication.run(DemoApplication.class, args);
    	}
    
    }

启动控制台打印

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM3NDc4_size_16_color_FFFFFF_t_70][]

[https_blog.csdn.net_qq_43037478_article_details_115670879]: https://blog.csdn.net/qq_43037478/article/details/115670879
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzMDM3NDc4_size_16_color_FFFFFF_t_70]: /images/20221022/910439e3a95d4decbfef65f152fe7fb5.png