RocketMQ——入门代码案例

男娘i 2021-09-26 23:58 266阅读 0赞

该案例基于springboot，和普通的spring项目的主要差别在于大大减少了配置话，不了解的同学可以先参考的一下我的[springboot干货专栏][springboot]，学习之后你会知道什么叫爽歪歪，什么叫美滋滋，再也不用话太多时间在项目的配置上了。

1.新建一个springboot项目

2.引入对应的rocketmq的jar包，这里以比较经典的3.2.6的版本为例

<dependency>  
                <groupId>com.alibaba.rocketmq</groupId>  
                <artifactId>rocketmq-client</artifactId>  
                <version>3.2.6</version>  
            </dependency>

3.新建Producer.Class

package com.gwd.rocketmq;
    
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
    import com.alibaba.rocketmq.client.producer.SendResult;
    import com.alibaba.rocketmq.common.message.Message;
    
    public class Producer {
    	public static void main(String[] args) throws MQClientException, InterruptedException {
    		//声明并初始化一个producer
            //需要一个producer group名字作为构造方法的参数，这里为producer1
            DefaultMQProducer producer = new DefaultMQProducer("producer1");
            
            //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
            //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
            producer.setNamesrvAddr("192.168.140.128:9876;192.168.140.129:9876");
       //     producer.setVipChannelEnabled(false);//3.2。6的版本没有该设置，在更新或者最新的版本中务必将其设置为false，否则会有问题
            //调用start()方法启动一个producer实例
            producer.start();
    
            //发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值
            for (int i = 0; i < 10; i++) {
                try {
                    Message msg = new Message("TopicTest",// topic
                            "TagA",// tag
                            ("Hello RocketMQ " + i).getBytes("utf-8")// body
                    );
                    
                    //调用producer的send()方法发送消息
                    //这里调用的是同步的方式，所以会有返回结果
                    SendResult sendResult = producer.send(msg);
                    System.out.println(sendResult.getSendStatus()); //发送结果状态
                    //打印返回结果，可以看到消息发送的状态以及一些相关信息
                    System.out.println(sendResult);
                } catch (Exception e) {
                    e.printStackTrace();
                    Thread.sleep(1000);
                }
            }
    
            //发送完消息之后，调用shutdown()方法关闭producer
            producer.shutdown();
    	}
    }

4.新建Consumer.Class

package com.gwd.rocketmq;
    
    import java.util.List;
    
    import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import com.alibaba.rocketmq.client.exception.MQClientException;
    import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
    import com.alibaba.rocketmq.common.message.MessageExt;
    
    public class Consumer {
    	public static void main(String[] args) throws MQClientException {
    		
    		//声明并初始化一个consumer
    	     //需要一个consumer group名字作为构造方法的参数，这里为consumer1
    	     DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");
    
    	     //同样也要设置NameServer地址
    	     consumer.setNamesrvAddr("192.168.140.128:9876;192.168.140.129:9876");
    
    	     //这里设置的是一个consumer的消费策略
    	     //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
    	     //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
    	     //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
    	     consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
    	     //设置consumer所订阅的Topic和Tag，*代表全部的Tag
    	     consumer.subscribe("TopicTest", "*");
    
    	     //设置一个Listener，主要进行消息的逻辑处理
    	     consumer.registerMessageListener(new MessageListenerConcurrently() {
    
    	         @Override
    	         public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
    	                                                         ConsumeConcurrentlyContext context) {
    
    	            // System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
    	        	 for (MessageExt messageExt : msgs) {  
    	        		  
                  //       System.out.println("messageExt: " + messageExt);//输出消息内容  
                        String messageBody = new String(messageExt.getBody());  
             //            System.out.println(messageExt.getStoreHost());//获取对应rocketmq服务器ip地址
    	        		 System.out.println("消费响应：msgId : " + messageExt.getMsgId() + ",  msgBody : " + messageBody);//输出消息内容  
                     }  
    	             
    	             //返回消费状态
    	             //CONSUME_SUCCESS 消费成功
    	             //RECONSUME_LATER 消费失败，需要稍后重新消费
    	             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    	         }
    	     });
    
    	     //调用start()方法启动consumer
    	     consumer.start();
    
    	     System.out.println("Consumer Started.");
    	 }
    	 
    }

5.先运行Consumer再运行Producer即可

[springboot]: https://blog.csdn.net/gwd1154978352/article/details/80651332