背景:
为了减轻服务器的压力,现在原有项目的基础上集成消息队列来异步处理消息!
此项目是企业真实需求,项目的代码属于线上生产代码,直接用于生产即可!
此项目采用MQ发送消息模式为:订阅模式,如果对RabbitMQ不熟悉,请学习RabbitMQ专栏进行相关知识点的学习!遇到问题,可以给我留言!看到后定会回复!
文章目录
- 一、RabbitMQ 订阅模式快速入门
- 1. RabbitMQ 订阅模式简述
- 2. RabbitMQ 订阅模式图示
- 3. MQ角色组成
- 二、SSM集成RabbitMQ
- 2.1. 引依赖
- 2.2. 生产者配置文件
- 2.3. 消费者配置文件
- 2.4. 连接配置文件
- 2.5. 生产者代码
- 2.6 消费者 ①
- 2.7 消费者 ②
- 2.8 MQ工具类
- 三、启动项目验证
- 3.1. 启动tomcat7插件
- 3.2. 清空控制台
- 四、管控台队列绑定交换机
- 4.1. 复制队列名称
- 4.2. 队列绑定交换机
- 五、请求验证测试
- 5.1. 生产者①请求
- 4.5. 生产者②请求
- 五、启动RabbitMQ
- 5.1. 进入sbin目录,双击运行
- 5.2. 启动图示
一、RabbitMQ 订阅模式快速入门
1. RabbitMQ 订阅模式简述
什么是发布/订阅模式(Publish/Subscribe)
简单解释就是,可以将消息发送给不同类型的消费者。做到发布一次,消费多个。
重要知识点:订阅模式不走路由routingKey,消息队列只绑定交换机。
2. RabbitMQ 订阅模式图示
解读:
1、1个生产者,多个消费者
2、每一个消费者都有自己的一个队列
3、生产者没有将消息直接发送到队列,而是发送到了交换机
4、每个队列都要绑定到交换机
5、生产者发送的消息,经过交换机,到达队列,实现,一个消息被多个消费者获取的目的
3. MQ角色组成
序号 | 说明 | 角色 |
---|---|---|
① | 生产者 | PRODUCER |
② | 消费者 | CONSUMER |
③ | 消息队列 | QUEUE |
④ | 交换机 | EXCHANGE |
⑤ | 交换机和队列绑定 | EXCHANGE和ROUTINGKEY绑定 |
二、SSM集成RabbitMQ
2.1. 引依赖
<!--spring整合rabbitmq--><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>1.4.0.RELEASE</version></dependency>
2.2. 生产者配置文件
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsdhttp://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.1.xsd"><!--生产者者配置如下:--><!-- 定义RabbitMQ的连接工厂 --><rabbit:connection-factory id="connectionFactory"host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}"password="${rabbitmq.password}" virtual-host="${rabbitmq.vhost}"/><!-- 管理消息队列 --><rabbit:admin connection-factory="connectionFactory"/><!--此处为配置文件方式 管控台配置模式需要注释 默认模式管控台 Start--><!-- 定义一个队列或者多个队列 自动声明--><!-- <rabbit:queue name="ORDER-CATEGORY-GBLFY-QUEUE" auto-declare="true" durable="true"/><rabbit:queue name="ORDER-USER-MENU-QUEUE" auto-declare="true" durable="true"/><rabbit:fanout-exchange name="ORDER-TRACE-EXCHANGE"><rabbit:bindings><!– 可绑定多个队列,发送的时候指定key进行发送 –><rabbit:binding queue="ORDER-CATEGORY-GBLFY-QUEUE"/><rabbit:binding queue="ORDER-USER-MENU-QUEUE"/></rabbit:bindings></rabbit:fanout-exchange>--><!--此处为配置文件方式 管控台配置模式需要注释 默认模式管控台 End--><!-- 定义交换机 自动声明--><rabbit:fanout-exchange name="ORDER-TRACE-EXCHANGE"auto-declare="true" durable="true"/><!-- 定义MQ消息模板 --><rabbit:template id="rabbitTemplate"connection-factory="connectionFactory" exchange="ORDER-TRACE-EXCHANGE"/>
</beans>
2.3. 消费者配置文件
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsdhttp://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.1.xsd"><!--消费者配置如下:--><!-- 定义RabbitMQ的连接工厂 --><rabbit:connection-factory id="connectionFactory"host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}"password="${rabbitmq.password}" virtual-host="${rabbitmq.vhost}"/><!-- 管理消息队列 --><rabbit:admin connection-factory="connectionFactory"/><!-- 定义一个队列或者多个队列 --><rabbit:queue name="ORDER-USER-MENU-QUEUE" auto-declare="true" durable="true"/><rabbit:queue name="ORDER-CATEGORY-GBLFY-QUEUE" auto-declare="true" durable="true"/><!-- 声明多个消费者对象 --><bean id="mQSimpleMsgHandler" class="com.gblfy.order.mqhandler.MQSimpleMsgHandler"/><bean id="orderMQMonitorMsgHandler" class="com.gblfy.order.mqhandler.OrderMQMonitorMsgHandler"/><!-- 监听队列 --><rabbit:listener-container connection-factory="connectionFactory"><rabbit:listener ref="mQSimpleMsgHandler" method="execute" queue-names="ORDER-CATEGORY-GBLFY-QUEUE"/><rabbit:listener ref="orderMQMonitorMsgHandler" method="execute" queue-names="ORDER-USER-MENU-QUEUE"/></rabbit:listener-container>
</beans>
2.4. 连接配置文件
rabbitmq.host=127.0.0.1
rabbitmq.port=5672
rabbitmq.username=admin
rabbitmq.password=admin
rabbitmq.vhost=/admin
2.5. 生产者代码
package com.gblfy.order.controller;import com.gblfy.order.utils.MQSendMsgUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;@Controller
@Slf4j
public class FanoutMQMsgSendController {@Autowiredprivate MQSendMsgUtils mqSendMsgUtils;@RequestMapping(value = "/snedUserMQMsg", method = RequestMethod.GET)@ResponseBodypublic String snedUserMQMsg() {/*** 模拟发送数据* 1. serviceName 接口名称* 2. type 路由routingKey*/String serviceName = "my name serviceName";String type = "user";//发送消息到MQ的交换机,通知其他系统mqSendMsgUtils.sendMsg(serviceName, type);return "snedUserMQMsg success !!!";}@RequestMapping(value = "/sendMenuMQMsg", method = RequestMethod.GET)@ResponseBodypublic String sendMenuMQMsg() {/*** 模拟发送数据* 1. serviceName 接口名称* 2. type 路由routingKey*/String serviceName = "my name serviceName";String type = "menu";//发送消息到MQ的交换机,通知其他系统mqSendMsgUtils.sendMsg(serviceName, type);return "sendMenuMQMsg success !!!";}@RequestMapping(value = "/snedCategoryMQMsg", method = RequestMethod.GET)@ResponseBodypublic String snedCategoryMQMsg() {/*** 模拟发送数据* 1. serviceName 接口名称* 2. type 路由routingKey*/String serviceName = "my name serviceName2";String type = "category.gblfy";//发送消息到MQ的交换机,通知其他系统mqSendMsgUtils.sendMsg(serviceName, type);return "snedCategoryMQMsg success !!!";}public static void main(String[] args) {DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println(dateFormat.format(new Date()));}
}
2.6 消费者 ①
package com.gblfy.order.mqhandler;import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;@Slf4j
public class MQSimpleMsgHandler {private static final ObjectMapper MAPPER = new ObjectMapper();/*** 接收MQ消息** @param msg*/public void execute(String msg) {try {JsonNode jsonNode = MAPPER.readTree(msg);String serviceName = jsonNode.get("serviceName").asText();String routingKey = jsonNode.get("routingKey").asText();String currentDate = jsonNode.get("currentDate").asText();log.info("接口名称:" + serviceName);log.info("路由routingKey:" + routingKey);log.info("当前时间:" + currentDate);} catch (Exception e) {e.printStackTrace();}}
}
2.7 消费者 ②
package com.gblfy.order.mqhandler;import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;@Slf4j
public class OrderMQMonitorMsgHandler {private static final ObjectMapper MAPPER = new ObjectMapper();/*** 接收MQ消息** @param msg*/public void execute(String msg) {try {JsonNode jsonNode = MAPPER.readTree(msg);String serviceName = jsonNode.get("serviceName").asText();String routingKey = jsonNode.get("routingKey").asText();String currentDate = jsonNode.get("currentDate").asText();log.info("接口名称:" + serviceName);log.info("路由routingKey:" + routingKey);log.info("当前时间:" + currentDate);} catch (Exception e) {e.printStackTrace();}}
}
2.8 MQ工具类
package com.gblfy.order.utils;import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gblfy.order.pojo.FisCallingTrace;
import com.gblfy.order.pojo.RequestInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;/*** MQ发送消息公用工具类* <p>* MQ发送消息模式采用 通配符模式* order.* 区配一个词* order.# 区配一个或者多个词* <p>** @author gblfy*/
@Component
@Slf4j
public class MQSendMsgUtils {//格式化时间public static final DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");// 日期格式//引入json工具类private static final ObjectMapper MAPPER = new ObjectMapper();@Autowired//注入发送消息模板private RabbitTemplate rabbitTemplate;/*** 发送MQ消息公用类** @param mFisCallingTrace 轨迹保存的数据对象* @param type 路由routingKey 字符串拼接* @param reqXml 请求报文* @param resXml 响应报文* @param uuid 随机生成的uuid*/public void sendMsg(FisCallingTrace mFisCallingTrace, String type, String reqXml, String resXml, String uuid) {try {RequestInfo requestInfo = new RequestInfo().builder().fisCallingTrace(mFisCallingTrace).mReqXml(reqXml).mResXml(resXml).mUUID(uuid).serviceName(mFisCallingTrace.getServicename()).type(type).build();//发送消息到MQ的交换机,通知其他系统String jsonStr = JSON.toJSONString(requestInfo);rabbitTemplate.convertAndSend("order." + type, jsonStr);} catch (Exception e) {e.printStackTrace();}}/*** 发送MQ消息公用类** @param serviceName 接口名称* @param type 路由routingKey*/public void sendMsg(String serviceName, String type) {try {//发送消息到MQ的交换机,通知其他系统Map<String, Object> msg = new HashMap<String, Object>();msg.put("serviceName", serviceName);msg.put("routingKey", type);msg.put("currentDate", dateFormat.format(new Date()));rabbitTemplate.convertAndSend("order." + type, MAPPER.writeValueAsString(msg));} catch (Exception e) {e.printStackTrace();}}// public static void main(String[] args) {
// //使用fastjson 实体类对象转jsonStr
// User ly = new User().builder()
// .id(1)
// .name("ly")
// .build();
// String jsonStr = JSON.toJSONString(ly);
// log.info("转换后jsonStr的用户:" + jsonStr);
//
// //使用fastjson 进行jsonObject转实体类对象
// String userString = "{\"id\":1,\"name\":\"ly\"}";
//
// JSONObject userJson = JSONObject.parseObject(userString);
// User user = JSON.toJavaObject(userJson, User.class);
//
// log.info("用户姓名:" + user.getName());
// log.info("用户ID:" + user.getId());
// }
}
三、启动项目验证
3.1. 启动tomcat7插件
3.2. 清空控制台
四、管控台队列绑定交换机
4.1. 复制队列名称
4.2. 队列绑定交换机
点击声明的交换机
配置文件中声明的交换机
在交换机中绑定【ORDER-CATEGORY-GBLFY-QUEUE】队列
在交换机中绑定【ORDER-CATEGORY-GBLFY-QUEUE】队列
在交换机中绑定【ORDER-USER-MENU-QUEUE】队列
在交换机菜单中查看,绑定的队列
在队列菜单中查看,绑定的交换机
五、请求验证测试
5.1. 生产者①请求
http://localhost:8888/snedUserMQMsg
4.5. 生产者②请求
http://localhost:8888/sendMenuMQMsg