RocketMq实现异步_代码实现

r囧r小猫 2023-08-17 15:56 162阅读 0赞

1、产生消息

package com.myfutech.common.rocketmq.producer;
    
    import com.myfutech.common.rocketmq.constants.RocketMQErrorEnum;
    import com.myfutech.common.rocketmq.exception.RocketMQException;
    import org.apache.commons.lang3.StringUtils;
    import org.apache.rocketmq.client.exception.MQBrokerException;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.exception.RemotingException;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.Properties;
    
    /**
     * Created by Liuxd on 2018/12/27.
     */
    public class RocketMqProvider {
    
        private static Logger LOGGER = LoggerFactory.getLogger(RocketMqProvider.class);
    
        private String groupName = "qd-rocketmq";
    
        private DefaultMQProducer defaultMQProducer;
    
        public RocketMqProvider(Properties props) {
    
            String namesrvAddr = props.getProperty("namesrvAddr");
            if (StringUtils.isEmpty(namesrvAddr)) {
                throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL, "namesrvAddr is blank", false);
            }
    
            defaultMQProducer = new DefaultMQProducer(groupName);
            defaultMQProducer.setNamesrvAddr(namesrvAddr);
            defaultMQProducer.setVipChannelEnabled(false);
            defaultMQProducer.setRetryTimesWhenSendFailed(5);
            defaultMQProducer.setRetryTimesWhenSendAsyncFailed(5);
            defaultMQProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
            defaultMQProducer.setMaxMessageSize(64);
            defaultMQProducer.setSendMsgTimeout(3000);
            //如果发送消息失败，设置重试次数，默认为2次
            defaultMQProducer.setRetryTimesWhenSendFailed(2);
            //如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
            //defaultMQProducer.setInstanceName(instanceName);
    
            try {
                defaultMQProducer.start();
    
                LOGGER.info(String.format("defaultMQProducer is start ! groupName:[%s],namesrvAddr:[%s]"
                        , props.getProperty("groupName"), props.getProperty("namesrvAddr")));
    
            } catch (MQClientException e) {
                LOGGER.error(String.format("defaultMQProducer is error {}", e.getMessage(), e));
                throw new RocketMQException(e);
            }
    
        }
    
        /**
         * 向rocketMq发送消息
         *
         * @param topic
         * @param tag
         * @param msg
         * @return
         * @throws MQClientException
         * @throws RemotingException
         * @throws MQBrokerException
         * @throws InterruptedException
         */
        public String send(String topic, String tag, String msg) {
            LOGGER.info("开始向rocketMq发送消息：topic:" + topic + ",tag:" + tag + ",msg:{}" + msg);
            Message sendMsg = new Message(topic, tag, msg.getBytes());
    
            SendResult sendResult = null;
            try {
                sendResult = defaultMQProducer.send(sendMsg);
            } catch (Exception e) {
                e.printStackTrace();
                return "error";
            }
    
    
            LOGGER.info("消息发送响应信息：" + sendResult.toString());
    
            return sendResult.toString();
        }
    }

2、消费消息

package com.myfutech.common.rocketmq.consumer;
    
    
    import com.myfutech.common.rocketmq.constants.RocketMQErrorEnum;
    import com.myfutech.common.rocketmq.exception.RocketMQException;
    import org.apache.commons.lang3.StringUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
    
    
    import java.util.Map;
    
    /**
     * Created by Liuxd on 2018/12/27.
     */
    public class RocketMqConsumer {
    
        private static Logger LOGGER = LoggerFactory.getLogger(RocketMqConsumer.class);
    
        private DefaultMQPushConsumer defaultMQPushConsumer;
    
        /**
         * 必填参数：6个
         * nameServerAddr，topic，tag，consumerGroup，mqMessageListenerProcessor，messageModel
         * @param map
         */
        public RocketMqConsumer(Map<String, Object> map) {
            String namesrvAddr = map.get("namesrvAddr") == null ? "" : map.get("namesrvAddr").toString().trim();
            if (StringUtils.isEmpty(namesrvAddr)) {
                throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL, "nameServerAddr is blank", false);
            }
            String topic = map.get("topic") == null ? "" : map.get("topic").toString().trim();
            if (StringUtils.isEmpty(topic)) {
                throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL, "topic is null !!!", false);
            }
            String tag = map.get("tag") == null ? "" : map.get("tag").toString().trim();
            if (StringUtils.isEmpty(tag)) {
                throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL, "tag is null !!!", false);
            }
            String consumerGroup = map.get("consumerGroup") == null ? "" : map.get("consumerGroup").toString().trim();
            if (StringUtils.isEmpty(consumerGroup)) {
                throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL, "consumerGroup is null !!!", false);
            }
            MessageListenerConcurrently mqMessageListenerProcessor = (MessageListenerConcurrently) map.get("mqMessageListenerProcessor");
    
            if (null == mqMessageListenerProcessor) {
                throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL, "mqMessageListenerProcessor is null !!!", false);
            }
    
            /**
             * 模式：广播或集群
             */
            MessageModel messageModel = (MessageModel) map.get("messageModel");
    
            if (null == messageModel) {
                throw new RocketMQException(RocketMQErrorEnum.PARAMM_NULL, "messageModel is null !!!", false);
            }
    
            defaultMQPushConsumer = new DefaultMQPushConsumer(consumerGroup);
            defaultMQPushConsumer.setNamesrvAddr(namesrvAddr);
            defaultMQPushConsumer.setConsumeThreadMin(20);
            defaultMQPushConsumer.setConsumeThreadMax(60);
    
            /**
             * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
             * 如果非第一次启动，那么按照上次消费的位置继续消费
             */
            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
    
            /**
             * 设置一次消费消息的条数，默认为1条
             */
            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
            //关闭VIP通道，避免接收不了消息
            defaultMQPushConsumer.setVipChannelEnabled(false);
    
    
            // 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
            // 如果非第一次启动，那么按照上次消费的位置继续消费
            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            //设置为集群消费(区别于广播消费)
            defaultMQPushConsumer.setMessageModel(messageModel);
    
            /**
             * 注册监听
             */
            defaultMQPushConsumer.registerMessageListener(mqMessageListenerProcessor);
    
            try {
    
                // 订阅Topic下Tag的消息
                defaultMQPushConsumer.subscribe(topic, tag);
    
                defaultMQPushConsumer.start();
                LOGGER.info("defaultMQPushConsumer is start !!! groupName:"+consumerGroup+",topic:"+topic+",namesrvAddr:"+namesrvAddr);
            } catch (MQClientException e) {
                LOGGER.error("defaultMQPushConsumer is start !!! groupName:"+consumerGroup+",topic:"+topic+",namesrvAddr:"+namesrvAddr);
                throw new RocketMQException(e);
            }
        }
    
    
    }

3、监听

package com.myfutech.common.rocketmq.test;
    
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.apache.commons.collections4.CollectionUtils;
    
    
    
    import java.util.List;
    
    /**
     * 消费者消费消息路由
    
     */
    public class MQConsumeMsgListenerProcessor implements MessageListenerConcurrently{
    
    	/**
    	 *  默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息<br/>
    	 *  不要抛异常，如果没有return CONSUME_SUCCESS ，consumer会重新消费该消息，直到return CONSUME_SUCCESS
    	 */
    	@Override
    	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    		if(CollectionUtils.isEmpty(msgs)){
    			System.out.println("接受到的消息为空，不处理，直接返回成功");
    			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    		}
    
    		MessageExt messageExt = msgs.get(0);
    		System.out.println("接受到的消息为："+messageExt.toString());
    
    		if(messageExt.getTopic().equals("DemoTopic")){
    			if(messageExt.getTags().equals("demoTag")){
    				//TODO 判断该消息是否重复消费（RocketMQ不保证消息不重复，如果你的业务需要保证严格的不重复消息，需要你自己在业务端去重）
    				String str = new String (messageExt.getBody());
    				System.out.println("接到消息："+str);
    				int reconsume = messageExt.getReconsumeTimes();
    				if(reconsume ==3){//消息已经重试了3次，如果不需要再次消费，则返回成功
    					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    				}
    				//TODO 处理对应的业务逻辑
    			}
    		}
    		// 如果没有return success ，consumer会重新消费该消息，直到return success
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    	}
    }

4、测试产生消息

package com.myfutech.common.rocketmq.test;
    
    import com.myfutech.common.rocketmq.producer.RocketMqProvider;
    import org.junit.Test;
    
    import java.io.InputStream;
    import java.util.Properties;
    
    /**
     * Created by Liuxd on 2018/12/28.
     */
    public class RocketMqProviderTest {
    
        @Test
        public void testRocketMqProducer() throws Exception{
            InputStream is = getClass().getClassLoader().getResourceAsStream("rocketMq.properties");
            Properties properties = new Properties();
            properties.load(is);
            RocketMqProvider rocketMqProvider = new RocketMqProvider(properties);
    
    
            String msg = "客户端测试发送消息";
            System.out.println("开始发送消息："+msg);
            String topic = "DemoTopic";
            String tag = "demoTag";
    
    
            String sendResult = rocketMqProvider.send(topic,tag,msg);
            System.out.println("消息发送响应信息："+sendResult);
        }
    }

5、测试消费消息

package com.myfutech.common.rocketmq.test;
    
    import com.myfutech.common.rocketmq.consumer.RocketMqConsumer;
    import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
    import org.junit.Test;
    
    import java.io.InputStream;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Properties;
    
    /**
     * Created by Liuxd on 2018/12/28.
     */
    
    public class RocketMqConsumerTest {
    
        @Test
        public void TestRocketMqConsumer() throws Exception{
            InputStream is = getClass().getClassLoader().getResourceAsStream("rocketMq.properties");
            Properties properties = new Properties();
            properties.load(is);
    
            Map<String, Object> map = new HashMap<String, Object>((Map) properties);
            MQConsumeMsgListenerProcessor mqConsumeMsgListenerProcessor = new MQConsumeMsgListenerProcessor();
            map.put("mqMessageListenerProcessor",mqConsumeMsgListenerProcessor);
            map.put("messageModel", MessageModel.CLUSTERING);
    
            RocketMqConsumer rocketMqConsumer = new RocketMqConsumer(map);
    
            Thread.sleep(100000);
    
        }
    }

6、监听

7、配置文件

groupName=qd-rocketmq
    namesrvAddr=192.168.30.150:9876
    topic=DemoTopic
    tag=demoTag
    consumerGroup=qd-rocketmq-test

8、因篇幅原因，忽略了两个自定义的异常类和Errocode的自定义类