欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识！
您现在的位置是： 首页

RocketMq实践

程序员文章站 2022-07-13 09:50:31
...

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

RocketMq实践

  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;
    
    @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);
        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.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;
	}

	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;
	}

	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("发送异步消息发生异常");
		}
	}

	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;
	}
}

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.logback.xml文件：

<?xml version="1.0" encoding="UTF-8"?>
<configuration>

    <conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter" />
    <conversionRule conversionWord="wex" converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter" />
    <conversionRule conversionWord="wEx" converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter" />

    <!-- 彩色日志格式 -->
    <property name="CONSOLE_LOG_PATTERN"
              value="${CONSOLE_LOG_PATTERN:-%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}" />

    <!-- 控制台打印日志的相关配置 -->
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>${CONSOLE_LOG_PATTERN}<!--%d{yyyy-MM-dd HH:mm:ss.SSS} [%level] [%class:%line] - %m%n--></pattern>
            <charset>utf8</charset>
        </encoder>
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>debug</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>ACCEPT</onMismatch>
        </filter>
    </appender>

    <!-- 文件保存日志的相关配置 -->
    <appender name="file" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>logs/rocketMq.log</file>
        <encoder>
            <pattern>${CONSOLE_LOG_PATTERN}<!--%d{yyyy-MM-dd HH:mm:ss.SSS} [%level] [%class:%line] - %m%n--></pattern>
            <charset>utf8</charset>
        </encoder>
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>info</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>ACCEPT</onMismatch>
        </filter>
        <!-- 循环政策：基于时间创建日志文件 -->
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>logs/%d{yyyy-MM-dd}/tms_svc.%d{yyyy-MM-dd}.log</fileNamePattern>
            <maxHistory>30</maxHistory>
        </rollingPolicy>
    </appender>

    <logger name="com.yj.mq" level="DEBUG" />
    <!-- 基于dubug处理日志：具体控制台或者文件对日志级别的处理还要看所在appender配置的filter，如果没有配置filter，则使用root配置 -->
    <root level="info">
        <appender-ref ref="console" />
        <appender-ref ref="file" />
    </root>
</configuration>

13.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

出现结果：

RocketMq实践

②访问：

http://127.0.0.1:9001/sendlist

出现结果：

RocketMq实践

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

③访问：

http://127.0.0.1:9001/sendOrderlist

出现结果：

RocketMq实践

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

 

 

 

相关标签： RocketMq SpringBoot

上一篇： Java数据结构与算法：图、图的概念、深度优先搜索DFS、广度优先搜索BFS、思路分析、代码实现

下一篇： Nginx 负载均衡策略与配置大结局

推荐阅读