RocketMq之简单实践

╰半夏微凉° 2022-05-22 08:04 208阅读 0赞

1、项目总体结构，如下图所示

![70][]

1.  RocketMqConfig文件：

package com.yj.mq.config;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RocketMqConfig {
        @Value("${rocketmq.producer.groupName}")
        private String producerGroupName;
        @Value("${rocketmq.consumer.groupName}")
        private String consumerGroupName;
        @Value("${rocketmq.consumer.order.groupName}")
        private String consumerOrderGroupName;
        @Value("${rocketmq.namesrv.addr}")
        private String nameServerAddress;
        @Value("${rocketmq.topic}")
        private String topic;
        @Value("${rocketmq.order.topic}")
        private String orderTopic;
        @Value("${rocketmq.sendMsg.timeOut:30000}")
        private Integer sendMsgTimeOut;
        @Value("${rocketmq.retry.times:3}")
        private Integer retryTimes;
        
        @Autowired
        private MessageListenerConcurrently messageListener;
        
        @Autowired
        private MessageListenerOrderly orderMessageListener;
    
        @Bean(initMethod = "start", destroyMethod = "shutdown")
        public DefaultMQProducer producer() {
            DefaultMQProducer producer = new DefaultMQProducer(producerGroupName);
            //VipChannel阿里内部使用版本才用，开源版本没有，默认为true，占用10909端口，此时虚拟机需要开放10909端口，否则会报 ：connect to <：10909> failed异常，可以直接设置为false
            //producer.setVipChannelEnabled(false);
            producer.setNamesrvAddr(nameServerAddress);
            producer.setSendMsgTimeout(sendMsgTimeOut);
            producer.setRetryTimesWhenSendFailed(retryTimes);
            return producer;
        }
    
        @Bean(initMethod = "start", destroyMethod = "shutdown")
        public DefaultMQPushConsumer consumer() {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroupName);
            consumer.setNamesrvAddr(nameServerAddress);
            consumer.setMaxReconsumeTimes(0);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.registerMessageListener(messageListener);
            //consumer.setMessageModel(MessageModel.CLUSTERING);集群模式(默认)
            //consumer.setMessageModel(MessageModel.BROADCASTING);//广播模式
            try {
                consumer.subscribe(topic, "*");
            } catch (MQClientException e) {
                e.printStackTrace();
            }
            return consumer;
        }
        
        @Bean(initMethod = "start", destroyMethod = "shutdown")
        public DefaultMQPushConsumer orderConsumer() {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerOrderGroupName);
            consumer.setNamesrvAddr(nameServerAddress);
            consumer.setMaxReconsumeTimes(0);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //consumer.setMessageModel(MessageModel.CLUSTERING);集群模式(默认)
            //consumer.setMessageModel(MessageModel.BROADCASTING);//广播模式
            consumer.registerMessageListener(orderMessageListener);
            try {
                consumer.subscribe(orderTopic, "*");
            } catch (MQClientException e) {
                e.printStackTrace();
            }
            return consumer;
        }
    }

2.MessageProcessor文件：

package com.yj.mq.consumer;
    
    import org.apache.rocketmq.common.message.MessageExt;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    
    @Service
    public class MessageProcessor {
    	
    	@Value("${server.port}")  
        private String port; 
    
    	private static final Logger LOGGER = LoggerFactory.getLogger(MessageProcessor.class);
    
    	public Boolean handleMessage(MessageExt messageExt) {
    		try {
    			String body = null;
    			// 消费者消费
    			if (null == messageExt || null == messageExt.getBody()) {
    				LOGGER.info("消息体为空");
    				return false;
    			}
    			body = new String(messageExt.getBody());
    			LOGGER.info("port:"+port+",消费Tag:" + messageExt.getTags() + ",body:" + body);
    			//Order order = JSON.parseObject(body, Order.class);
    		} catch (Exception e) {
    			e.printStackTrace();
    			LOGGER.error(e.getMessage(), e);
    			return false;
    		}
    		return true;
    	}
    }

3.OrderMessageProcessor文件：

package com.yj.mq.consumer;
    
    import org.apache.rocketmq.common.message.MessageExt;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    
    @Service
    public class OrderMessageProcessor {
    	
    	@Value("${server.port}")  
        private String port; 
    
    	private static final Logger LOGGER = LoggerFactory.getLogger(OrderMessageProcessor.class);
    
    	public Boolean handleMessage(MessageExt messageExt) {
    		try {
    			String body = null;
    			// 消费者消费
    			if (null == messageExt || null == messageExt.getBody()) {
    				LOGGER.info("消息体为空");
    				return false;
    			}
    			body = new String(messageExt.getBody());
    			LOGGER.info("port:"+port+",消费Tag:" + messageExt.getTags() + ",body:" + body);
    			//Order order = JSON.parseObject(body, Order.class);
    		} catch (Exception e) {
    			e.printStackTrace();
    			LOGGER.error(e.getMessage(), e);
    			return false;
    		}
    		return true;
    	}
    }

4.MQController文件：

package com.yj.mq.controller;
    
    import java.util.ArrayList;
    import java.util.List;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.web.bind.annotation.RequestBody;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
    import com.alibaba.fastjson.JSON;
    import com.yj.mq.entity.BusinessException;
    import com.yj.mq.entity.Order;
    import com.yj.mq.producer.MQProducer;
    
    @RestController
    public class MQController {
    
    	private static final Logger LOGGER = LoggerFactory.getLogger(MQController.class);
    
    	@Autowired
    	private MQProducer producer;
    
    	@Value("${rocketmq.topic}")
    	private String topic;
    
    	@Value("${rocketmq.order.topic}")
    	private String orderTopic;
    
    	@RequestMapping("/send")
    	public SendResult send(@RequestBody String desc) {
    		SendResult result = null;
    		try {
    			Order order = new Order();
    			order.setOrderId(1L);
    			order.setDesc(desc);
    			String userStr = JSON.toJSONString(order);
    
    			Message message = new Message();
    			message.setTopic(topic);
    			message.setBody(userStr.getBytes());
    			message.setTags(desc + "_tag");
    			result = producer.syncSend(message);
    			LOGGER.info("SendResult:" + result);
    		} catch (BusinessException e) {
    			LOGGER.info("e:" + e.getMessage());
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    		return result;
    	}
    
    	@RequestMapping("/sendlist")
    	public SendResult sendlist(@RequestBody String name) {
    		SendResult result = null;
    		try {
    			List<Order> orderList = buildOrders();
    			for (int i = 0; i < orderList.size(); i++) {
    				String body = JSON.toJSONString(orderList.get(i));
    				Message msg = new Message(topic, body.getBytes());
    				result = producer.syncSend(msg);
    				LOGGER.info("SendResult:" + result);
    			}
    		} catch (BusinessException e) {
    			LOGGER.info("e:" + e.getMessage());
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    		return result;
    	}
    
    	// 发送有序消息
    	@RequestMapping("/sendOrderlist")
    	public SendResult sendOrderlist(@RequestBody String name) {
    		SendResult result = null;
    		try {
    			List<Order> orderList = buildOrders();
    			for (int i = 0; i < orderList.size(); i++) {
    				String body = JSON.toJSONString(orderList.get(i));
    				Message msg = new Message(orderTopic, body.getBytes());
    				result = producer.sendOrder(msg, orderList.get(i).getOrderId());
    				LOGGER.info("SendResult:" + result);
    			}
    		} catch (BusinessException e) {
    			LOGGER.info("e:" + e.getMessage());
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    		return result;
    	}
    
    	@RequestMapping("/sendDelay")
    	public SendResult sendDelay(@RequestBody String desc, @RequestParam("level") int level) {
    		SendResult result = null;
    		try {
    			Order order = new Order();
    			order.setOrderId(1L);
    			order.setDesc(desc);
    			String userStr = JSON.toJSONString(order);
    
    			Message message = new Message();
    			message.setTopic(topic);
    			message.setBody(userStr.getBytes());
    			message.setTags(desc + "_tag");
    			result = producer.delaySend(message, level);
    			LOGGER.info("SendResult:" + result);
    		} catch (BusinessException e) {
    			LOGGER.info("e:" + e.getMessage());
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    		return result;
    	}
    
    	private List<Order> buildOrders() {
    		List<Order> orderList = new ArrayList<Order>();
    		for (int i = 0; i < 20; i++) {
    			Order orderDemo = new Order();
    			orderDemo.setOrderId(new Long((long) i));
    			orderDemo.setDesc("创建订单");
    			orderList.add(orderDemo);
    
    			orderDemo = new Order();
    			orderDemo.setOrderId(new Long((long) i));
    			orderDemo.setDesc("订单付款");
    			orderList.add(orderDemo);
    
    			orderDemo = new Order();
    			orderDemo.setOrderId(new Long((long) i));
    			orderDemo.setDesc("付款成功");
    			orderList.add(orderDemo);
    
    			orderDemo = new Order();
    			orderDemo.setOrderId(new Long((long) i));
    			orderDemo.setDesc("订单完成");
    			orderList.add(orderDemo);
    		}
    		return orderList;
    	}
    }

5.BusinessException文件：

package com.yj.mq.entity;
    
    public class BusinessException extends RuntimeException {
    	private static final long serialVersionUID = -8138602623241348983L;
    	private String errorMessage = null;
    
    	public BusinessException() {
    		super();
    	}
    
    	public BusinessException(String errorMessage) {
    		super(errorMessage);
    		this.errorMessage = errorMessage;
    	}
    
    	public String getMessage() {
    		if (errorMessage != null) {
    			return errorMessage;
    		}
    		if (super.getMessage() != null)
    			return super.getMessage();
    		return errorMessage;
    	}
    }

6.Order文件：

package com.yj.mq.entity;
    
    import java.io.Serializable;
    
    public class Order implements Serializable{
    	private static final long serialVersionUID = -2203625690315456938L;
    	private Long orderId;
    	private String desc;
    	public Long getOrderId() {
    		return orderId;
    	}
    	public void setOrderId(Long orderId) {
    		this.orderId = orderId;
    	}
    	public String getDesc() {
    		return desc;
    	}
    	public void setDesc(String desc) {
    		this.desc = desc;
    	}
    	@Override
    	public String toString() {
    		return "Order [orderId=" + orderId + ", desc=" + desc + "]";
    	}
    }

7.MQListener文件：

package com.yj.mq.listener;
    
    import java.util.List;
    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.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import com.yj.mq.consumer.MessageProcessor;
    
    @Component
    public class MessageListener implements MessageListenerConcurrently {
    
    	@Autowired
    	private MessageProcessor messageProcessor;
    
    
    	@Override
    	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
    		for (MessageExt messageExt : list) {
    			boolean result = messageProcessor.handleMessage(messageExt);
    			if (!result) {
    				return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    			}
    		}
    		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    	}
    }

8.OrderMessageListener文件：

package com.yj.mq.listener;
    
    import java.util.List;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import com.yj.mq.consumer.MessageProcessor;
    
    @Component
    public class OrderMessageListener implements MessageListenerOrderly{
    	
    	@Autowired
    	private MessageProcessor messageProcessor;
    
    	@Override
    	public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {//有序监听
    		context.setAutoCommit(true); 
            for (MessageExt messageExt: msgs) {    
            	boolean result = messageProcessor.handleMessage(messageExt);
                if (!result) {
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            }   
            return ConsumeOrderlyStatus.SUCCESS;    
    	}
    }

9.MQProducer文件：

package com.yj.mq.producer;
    
    import java.util.List;
    
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.MessageQueueSelector;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.common.message.MessageQueue;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import com.yj.mq.entity.BusinessException;
    
    @Component
    public class MQProducer {
    	private static final Logger logger = LoggerFactory.getLogger(MQProducer.class);
    
    	@Autowired
    	private DefaultMQProducer producer;
    
    	/*
    	 * 可靠同步发送 原理：同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式。
    	 * 场景：此种方式应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。 类似推拉的形式 发送 ->同步返回 ->发送 ->同步返回
    	 */
    	public SendResult syncSend(Message message) throws BusinessException {
    		SendResult result = null;
    		try {
    			result = producer.send(message);
    		} catch (Exception e) {
    			e.printStackTrace();
    			throw new BusinessException("发送同步消息发生异常");
    		}
    		return result;
    	}
    
    	/*
    	 * 可靠异步发送 原理：异步发送是指发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。 MQ
    	 * 的异步发送，需要用户实现异步发送回调接口（SendCallback）。消息发送方在发送了一条消息后，不需要等待服务器响应即可返回，
    	 * 进行第二条消息发送。发送方通过回调接口接收服务器响应，并对响应结果进行处理。
    	 * 场景：异步发送一般用于链路耗时较长，对响应时间较为敏感的业务场景，例如用户视频上传后通知启动转码服务，转码完成后通知推送转码结果等。 耗时比较长的
    	 * 可以不需要同步返回给用户的
    	 */
    	public void asyncSend(Message message) throws BusinessException {
    		try {
    			producer.send(message, new SendCallback() {
    				@Override
    				public void onSuccess(SendResult result) {
    					logger.info("消息发布成功,messageId：" + result.getMsgId());
    				}
    
    				@Override
    				public void onException(Throwable throwable) {
    					logger.error("消息发布失败：" + throwable.getMessage(), throwable);
    				}
    			});
    		} catch (Exception e) {
    			e.printStackTrace();
    			throw new BusinessException("发送异步消息发生异常");
    		}
    	}
    
    	/*
    	 * 单向（Oneway）发送 原理：单向（Oneway）发送特点为发送方只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。
    	 * 此方式发送消息的过程耗时非常短，一般在微秒级别。 场景：适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。
    	 */
    	public void oneWaySend(Message message) throws BusinessException {
    		try {
    			producer.sendOneway(message);
    		} catch (Exception e) {
    			e.printStackTrace();
    			throw new BusinessException("发送单向消息发生异常");
    		}
    	}
    
    	// 有序发送
    	public SendResult sendOrder(Message message, Long orderId) throws BusinessException {
    		SendResult result = null;
    		try {
    			result = producer.send(message, new MessageQueueSelector() {
    				@Override
    				public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
    					Long id = (Long) arg;
    					long index = id % mqs.size();
    					return mqs.get((int) index);
    				}
    			}, orderId);
    		} catch (Exception e) {
    			e.printStackTrace();
    			throw new BusinessException("发送有序消息发生异常");
    		}
    		return result;
    	}
    
    	/*
    	 * 延时消息，简单来说就是当 producer 将消息发送到 broker 后，会延时一定时间后才投递给 consumer 进行消费。
    	 * RcoketMQ的延时等级为：1s，5s，10s，30s，1m，2m，3m，4m，5m，6m，7m，8m，9m，10m，20m，30m，1h，2h
    	 * 。level=0，表示不延时。level=1，表示 1 级延时，对应延时 1s。level=2 表示 2 级延时，对应5s，以此类推。
    	 */
    	public SendResult delaySend(Message message, int level) throws BusinessException {
    		SendResult result = null;
    		try {
    			message.setDelayTimeLevel(level);
    			result = producer.send(message);
    		} catch (Exception e) {
    			e.printStackTrace();
    			throw new BusinessException("发送同步消息发生异常");
    		}
    		return result;
    	}
    }

10.RocketMqApplication文件：

package com.yj.mq;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class RocketMqApplication {
    	public static void main(String[] args) {
    		SpringApplication.run(RocketMqApplication.class, args);
    	}
    }

11.application.properties文件：

server.port=9001
    ## rocketmq
    rocketmq.namesrv.addr=192.168.81.130:9876
    rocketmq.producer.groupName=PRODUCER_GROUP_CSDN
    rocketmq.consumer.groupName=CONSUMER_GROUP_CSDN
    rocketmq.consumer.order.groupName=CONSUMER_ORDER_GROUP_CSDN
    rocketmq.topic=CSDN
    rocketmq.order.topic=CSDN_ORDER

12.pom.xml文件：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    
    	<groupId>com.yj</groupId>
    	<artifactId>RocketMq</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    	<packaging>jar</packaging>
    
    	<name>RocketMq</name>
    	<url>http://maven.apache.org</url>
    
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    	</properties>
    
    	<parent>
    		<groupId>org.springframework.boot</groupId>
    		<artifactId>spring-boot-starter-parent</artifactId>
    		<version>1.5.2.RELEASE</version>
    		<relativePath />
    	</parent>
    
    	<dependencies>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-web</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-test</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.rocketmq</groupId>
    			<artifactId>rocketmq-client</artifactId>
    			<version>4.2.0</version>
    		</dependency>
    		<dependency>
    			<groupId>com.google.code.gson</groupId>
    			<artifactId>gson</artifactId>
    		</dependency>
    	</dependencies>
    </project>

2、验证

①访问：

http://127.0.0.1:9001/send

出现结果：

![70 1][]

②访问：

http://127.0.0.1:9001/sendlist

出现结果：

![70 2][]

不同订单号之间会出现乱序，相同的订单号内的订单，也发现有乱序消费的情况。

③访问：

http://127.0.0.1:9001/sendOrderlist

出现结果：

![70 3][]

不同订单号之间会出现乱序，但是相同的订单号内的订单，未发现有乱序消费的情况，都是顺序消费的。

[70]: /images/20220522/9f73907483f349deafea1075397c5a7b.png
[70 1]: /images/20220522/55167c5885c54e4a8a951e0460d25ae5.png
[70 2]: /images/20220522/c1543e448ded4cdfac9323c2ba0717ac.png
[70 3]: /images/20220522/1b30b8da7eeb440e84b800d2d568b2c1.png