RabbitMQ之项目中实战

时间:2022-08-27 12:09:30

  说了那么多,还不是为了在项目中进行实战吗,在实践中检验真理,不然我学他干嘛,不能解决项目中的实际问题的技术都是耍流氓。。。

一、后台管理系统发送消息

瞎咧咧:后台管理系统发送消息到交换机中,然后通知其他系统进行相应的操作,这岂不是美滋滋

1、导入依赖

注意:这里使用的是spring封装的rabbitMQ,只需要导入这一个依赖就可以了,这个版本有点老了

         <dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.4.0.RELEASE</version>
</dependency>

2、队列和交换机的绑定关系

实现:

2.1  在配置文件中将队列和交换机完成绑定

2.2  可以在管理界面中完成绑定

2.2.1 绑定关系如果发生变化,需要修改配置文件,并且服务需要重启

2.2.2 管理更加灵活

2.2.3 更容易对绑定关系的权限管理,流程管理

3、配置文件

注意:这个单独的放在一个配置文件中 applicationContext-rabbitmq.xml

 <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/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.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:topic-exchange name="TAOTAO-ITEM-EXCHANGE"
auto-declare="true" durable="true" /> <!-- 定义模板 -->
<rabbit:template id="rabbitTemplate"
connection-factory="connectionFactory" exchange="TAOTAO-ITEM-EXCHANGE" /> </beans>

4、具体代码实现

注意:这个实现是把发送消息的代码抽离出来了,具体发送的内容包括商品的id、操作类型(就是具体的删除、更新还是插入操作)、时间戳,那个ObjectMapper是jackson jar包中转json的类

     @Autowired
private RabbitTemplate RabbitTemplate; private static final ObjectMapper MAPPER = new ObjectMapper(); private void sendMsg(Long itemId, String type) {
try {
// 发送消息到MQ队列中,并通知其他系统
Map<String, Object> msg = new HashMap<String, Object>();
msg.put("itemId", itemId);
msg.put("type", type);
msg.put("date", System.currentTimeMillis());
this.RabbitTemplate.convertAndSend("item." + type, MAPPER.writeValueAsString(msg));
} catch (Exception e) {
e.printStackTrace();
}
} //例如:更新操作中,在方法的最后就直接调用此方法就可以了
// 发送消息到MQ队列中,并通知其他系统
sendMsg(item.getId(), "update");

 二、 前台系统接收消息

1、导入依赖

         <dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.4.0.RELEASE</version>
</dependency>

2、配置文件

 <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/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.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="TAOTAO-WEB-ITEM-QUEUE" auto-declare="true"
durable="true" /> <!-- 消费者对象 -->
<bean id="itemMQHandler" class="com.taotao.web.mq.handler.ItemMQHandler" /> <!-- 监听队列 -->
<rabbit:listener-container
connection-factory="connectionFactory">
<rabbit:listener ref="itemMQHandler" method="execute"
queue-names="TAOTAO-WEB-ITEM-QUEUE" />
</rabbit:listener-container> </beans>

3、具体处理逻辑

     @Autowired
private RedisService redisService; private static final ObjectMapper MAPPER = new ObjectMapper(); /**
* 删除缓存中的数据,完成数据同步
*
* @param msg
*/
public void execute(String msg) {
try {
JsonNode jsonNode = MAPPER.readTree(msg);
Long itemId = jsonNode.get("itemId").asLong();
String key = ItemService.REDIS_KEY + itemId;
this.redisService.del(key);
} catch (Exception e) {
e.printStackTrace();
} }

4、在界面管理工具中完成绑定关系

注意:这个操作是在 Exchanges菜单下面完成的

RabbitMQ之项目中实战

 三、搜索系统中接收消息

1、导入依赖

         <dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.4.0.RELEASE</version>
</dependency>

2、配置文件

注意:基本上是复制粘贴那个前台系统的

 <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/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.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="TAOTAO-SEARCH-ITEM-QUEUE" auto-declare="true"
durable="true" /> <!-- 消费者对象 -->
<bean id="itemMQHandler" class="com.taotao.search.mq.handler.ItemMQHandler" /> <!-- 监听队列 -->
<rabbit:listener-container
connection-factory="connectionFactory">
<rabbit:listener ref="itemMQHandler" method="execute"
queue-names="TAOTAO-SEARCH-ITEM-QUEUE" />
</rabbit:listener-container> </beans>

3、业务逻辑处理

 @Service
public class ItemService { @Autowired
private ApiService apiService; @Value("${TAOTAO_MANAGE_URL}")
private String TAOTAO_MANAGE_URL; private static final ObjectMapper MAPPER = new ObjectMapper(); public Item queryById(Long itemId){
try {
String url = TAOTAO_MANAGE_URL + "/rest/api/item/"+itemId;
String jsonData = this.apiService.doGet(url);
if(StringUtils.isNotEmpty(jsonData)){
return MAPPER.readValue(jsonData, Item.class);
}
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
}
 public class ItemMQHandler {

     private static final ObjectMapper MAPPER = new ObjectMapper();

     @Autowired
private HttpSolrServer httpSolrServer; @Autowired
private ItemService itemService; /**
* 处理消息,新增、修改、删除的消息,将商品数据同步到solr中
* 消息中并没有上商品的信息,需要通过商品id通过后台接口进行查询
*
* @param msg
*/
public void execute(String msg){
try {
JsonNode jsonNode = MAPPER.readTree(msg);
Long itemId = jsonNode.get("itemId").asLong();
String type = jsonNode.get("type").asText();
if(StringUtils.equals(type, "insert") || StringUtils.equals(type, "update")){
Item item = this.itemService.queryById(itemId);
this.httpSolrServer.addBean(item);
this.httpSolrServer.commit();
}else if(StringUtils.equals(type, "delete")){
this.httpSolrServer.deleteById(String.valueOf(itemId));
this.httpSolrServer.commit();
}
} catch (Exception e) {
e.printStackTrace();
}
}
}

4、在界面管理工具中完成绑定关系

RabbitMQ之项目中实战

四、总结

使用MQ实现商品数据的同步优势:

1、  降低系统间耦合度

2、  便于管理数据的同步