rabbitMQ 与spring、springmvc框架相结合的一个工程示例

浅浅的花香味﹌ 2022-07-15 13:00 155阅读 0赞

最近公司的项目中使用到的消息中间件是**rabbitMQ**，所以对**rabbitMQ**做了一些了解，有不足之处，欢迎大家指点。

# 一、rabbitMQ简介 #

**1.1、rabbitMQ的优点（适用范围）**  
1. 基于erlang语言开发具有高可用高并发的优点，适合集群服务器。  
2. 健壮、稳定、易用、跨平台、支持多种语言、文档齐全。  
3. 有消息确认机制和持久化机制，可靠性高。  
4. 开源  
其他MQ的优势：  
1. Apache ActiveMQ曝光率最高，但是可能会丢消息。  
2. ZeroMQ延迟很低、支持灵活拓扑，但是不支持消息持久化和崩溃恢复。

**1.2、几个概念说明**  
producer&Consumer  
producer指的是消息生产者，consumer消息的消费者。  
Queue  
消息队列，提供了FIFO的处理机制，具有缓存消息的能力。rabbitmq中，队列消息可以设置为持久化，临时或者自动删除。  
设置为持久化的队列，queue中的消息会在server本地硬盘存储一份，防止系统crash，数据丢失  
设置为临时队列，queue中的数据在系统重启之后就会丢失  
设置为自动删除的队列，当不存在用户连接到server，队列中的数据会被自动删除Exchange

Exchange类似于数据通信网络中的交换机，提供消息路由策略。rabbitmq中，producer不是通过信道直接将消息发送给queue，而是先发送给Exchange。一个Exchange可以和多个Queue进行绑定，producer在传递消息的时候，会传递一个ROUTING\_KEY，Exchange会根据这个ROUTING\_KEY按照特定的路由算法，将消息路由给指定的queue。和Queue一样，Exchange也可设置为持久化，临时或者自动删除。  
Exchange有4种类型：direct(默认)，fanout, topic, 和headers，不同类型的Exchange转发消息的策略有所区别：  
Direct  
直接交换器，工作方式类似于单播，Exchange会将消息发送完全匹配ROUTING\_KEY的Queue  
fanout  
广播是式交换器，不管消息的ROUTING\_KEY设置为什么，Exchange都会将消息转发给所有绑定的Queue。  
topic  
主题交换器，工作方式类似于组播，Exchange会将消息转发和ROUTING\_KEY匹配模式相同的所有队列，比如，ROUTING\_KEY为user.stock的Message会转发给绑定匹配模式为 \* .stock,user.stock， \* . \* 和\#.user.stock.\#的队列。（ \* 表是匹配一个任意词组，\#表示匹配0个或多个词组）  
headers  
消息体的header匹配（ignore）  
Binding  
所谓绑定就是将一个特定的 Exchange 和一个特定的 Queue 绑定起来。Exchange 和Queue的绑定可以是多对多的关系。  
virtual host  
在rabbitmq server上可以创建多个虚拟的message broker，又叫做virtual hosts (vhosts)。每一个vhost本质上是一个mini-rabbitmq server，分别管理各自的exchange，和bindings。vhost相当于物理的server，可以为不同app提供边界隔离，使得应用安全的运行在不同的vhost实例上，相互之间不会干扰。producer和consumer连接rabbit server需要指定一个vhost。

# 二、Rbbitmq与Spring、springMVC结合使用实例 #

新建一个maven工程，工程目录结构是：

![Image 1][]

1、消息生产者：

package cn.zyy.mq.test.productor;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    /**
     * 功能概要：消息产生，提交到队列中去
     * @author zhangyunye
     *
     */
    public class Productor {
    
    	private RabbitTemplate rabbitTemplate;
    	
    	public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
    		this.rabbitTemplate = rabbitTemplate;
    	}
    	
    	public void sendMessage(Object message){	
    		System.out.println("productor:" + message);
    		rabbitTemplate.convertAndSend("RECON_START", message);
    	}
    }

2、消息消费者：

package cn.zyy.mq.test.consumor;
    /**
     * 功能概要：消费接收
     * @author zhangyunye
     *
     */
    public class Consumor {
    
    	public void onMessage(String message){
    		System.out.println("consumor:" + message);
    	}	
    }

3、spring配置文件（配置rabbitmq的相关参数）

<?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:aop="http://www.springframework.org/schema/aop"
    	xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:jee="http://www.springframework.org/schema/jee" xmlns:mvc="http://www.springframework.org/schema/mvc"
    	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
    	xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
            http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
            http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee.xsd
            http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd
            http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd
            http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.3.xsd">
            
        
    	<rabbit:connection-factory id="connectionFactory"
    		username="zrj" password="zrj2015" host="192.168.1.237" port="5555"
    		virtual-host="/" />
    
    	
    	<rabbit:template id="amqTemplate" connection-factory="connectionFactory"
    		exchange="E_PAY_PPMS_CMD"></rabbit:template>
    
    	
    	<rabbit:admin connection-factory="connectionFactory" />
    
    	
    	<rabbit:listener-container
    		connection-factory="connectionFactory">
    		<rabbit:listener queue-names="Q_PAY_PPMS_RECON"
    			ref="consumor" method="onMessage" />
    	</rabbit:listener-container>
    
    	<bean id = "consumor" class = "cn.zyy.mq.test.consumor.Consumor"></bean>
    	<bean id = "produtor" class = "cn.zyy.mq.test.productor.Productor">
    		<property name="rabbitTemplate" ref = "amqTemplate"></property>
    	</bean>
    </beans>

这里需要注意一下，因为我们公司的mq服务器中已经对相关的exchange、queue进行了配置，所以在配置文件中，我不需要配置比如：定义queue、定义exchange和绑定queue

这些内容。但是为了通用性，我把相关配置内容写出来（本工程中没有用这些配置）：

<rabbit:queue name="Q_PAY_PPMS_RECON" durable="true" auto-delete="false" exclusive="false" />
    
    	
    	<rabbit:direct-exchange name="E_PAY_PPMS_CMD" durable="true" auto-delete="false">
    		<rabbit:bindings>
    			<rabbit:binding queue="Q_PAY_PPMS_RECON" key="RECON_START"></rabbit:binding>
    		</rabbit:bindings>
    	</rabbit:direct-exchange>

这就是本示例工程的核心代码。

整个工程的详细代码地址如下：http://download.csdn.net/download/u011991249/9685807

[Image 1]: