小东子的个人技术专栏

RabbitMQ工作原理和Spring的集成

发表于   |   字数统计: 2,108(字)   |   阅读时长: 9(分) |

在介绍RadditMQ之前，先介绍一下AMQP。

1.什么是AMPQ？

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。Erlang中的实现有 RabbitMQ等。

2.AMPQ的特征

AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

3.AMPQ的实现

3.1 OpenAMQ

AMQP的开源实现，用C语言编写，运行于Linux、AIX、Solaris、Windows、OpenVMS

3.2 Apache Qpid

Apache的开源项目，支持C++、Ruby、Java、JMS、Python和.NET

3.3 Redhat Enterprise MRG

实现了AMQP的最新版本0-10，提供了丰富的特征集，比如完全管理、联合、Active-Active集群，有Web控制台，还有许多企业级特征，客户端支持C++、Ruby、Java、JMS、Python和.NET

3.4 RabbitMQ

一个独立的开源实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。RabbitMQ发布在Ubuntu、FreeBSD、Centos中的实现

4.RabbitMQ是什么

RabbitMQ是一个遵循AMQP协议的消息中间件，它从生产者接收消息并递送给消费者，在这个过程中，根据规则进行路由，缓存与持久化。

5.RabbitMQ能为你做些什么？

消息系统允许软件、应用相互连接和扩展．这些应用可以相互链接起来组成一个更大的应用，或者将用户设备和数据进行连接。消息系统通过将消息的发送和接收分离来实现应用程序的异步和解偶。

或许你正在考虑进行数据投递，非阻塞操作或推送通知。或许你想要实现发布／订阅，异步处理，或者工作队列。所有这些都可以通过消息系统实现。

在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。

6.RabbitMQ的结构图

这里写图片描述

几个概念说明：

  • Broker：简单来说就是消息队列服务器实体。
  • Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
  • Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
  • Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
  • Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
  • vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
  • Producer：消息生产者，就是投递消息的程序。
  • Consumer：消息消费者，就是接受消息的程序。
  • Channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

消息队列的使用过程大概如下：

  1. 客户端连接到消息队列服务器，打开一个channel。
  2. 客户端声明一个exchange，并设置相关属性。
  3. 客户端声明一个queue，并设置相关属性。
  4. 客户端使用routing key，在exchange和queue之间建立好绑定关系。
  5. 客户端投递消息到exchange。
  6. exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里

exchange也有几个类型，完全根据key进行投递的叫做Direct交换机，例如，绑定时设置了routing key为”abc”，那么客户端提交的消息，只有设置了key为”abc”的才会投递到队列。对key进行模式匹配后进行投递的叫做Topic交换机，符号”#”匹配一个或多个词，符号””匹配正好一个词。例如”abc.#”匹配”abc.def.ghi”，”abc.”只匹配”abc.def”。还有一种不需要key的，叫做Fanout交换机，它采取广播模式，一个消息进来时，投递到与该交换机绑定的所有队列。

RabbitMQ支持消息的持久化，也就是数据写在磁盘上，为了数据安全考虑，我想大多数用户都会选择持久化。消息队列持久化包括3个部分：

  1. exchange持久化，在声明时指定durable => 1
  2. queue持久化，在声明时指定durable => 1
  3. 消息持久化，在投递时指定delivery_mode => 2（1是非持久化）

如果exchange和queue都是持久化的，那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化，一个非持久化，就不允许建立绑定。

7.RabbitMQ之交换机

消息(Message)由Client发送，RabbitMQ接收到消息之后通过交换机转发到对应的队列上面。Worker会从队列中获取未被读取的数据处理。

RabbitMQ包含四种不同的交换机类型：

  1. Direct exchange：直连交换机，转发消息到routigKey指定的队列，如果消息的routigKey和binding的routigKey直接匹配的话，消息将会路由到该队列
  2. Fanout exchange：扇形交换机，转发消息到所有绑定队列（速度最快），不管消息的routigKey息和binding的参数表头部信息和值是什么，消息将会路由到所有的队列
  3. Topic exchange：主题交换机，按规则转发消息（最灵活），如果消息的routigKey和binding的routigKey符合通配符匹配的话，消息将会路由到该队列
  4. Headers exchange：首部交换机 ，如果消息的头部信息和binding的参数表中匹配的话，消息将会路由到该队列。

8.Spring的集成RabbitMQ

8.1 在pom.xml 配置依赖的包

1
2
3
4
5
6
7
8
9
10
11
12
<!-- rabbitmq -->
 <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>${rabbitmq-client.version}</version>
 </dependency>
<!-- spring-rabbit -->
<dependency>
                <groupId>org.springframework.amqp</groupId>
                <artifactId>spring-rabbit</artifactId>
                <version>${spring-rabbit.version}</version>
</dependency>

8.2 rabbit.properties 配置文件

1
2
3
4
rabbit_username=guest
rabbit_password=guest
rabbit_host=127.0.0.1
rabbit_port=5672

8.3 创建生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
package com.lidong.dubbo.core.spittle.service;
import com.lidong.dubbo.api.spittle.service.IMessageProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
/**
 * @项目名称:lidong-dubbo
 * @类名:MessageProducerImp
 * @类的描述:
 * @作者:lidong
 * @创建时间:2017/2/4 上午10:01
 * @公司:chni
 * @QQ:1561281670
 * @邮箱:lidong1665@163.com
 */
@Service
public class MessageProducerServiceImp implements IMessageProducer {
    
    
    private Logger logger = LoggerFactory.getLogger(MessageProducerServiceImp.class);
    @Resource
    private AmqpTemplate amqpTemplate;
    @Override
    public void sendMessage(Object message) {
        logger.info("发送消息");
        logger.info("to send message:",message);
        amqpTemplate.convertAndSend("queueTestKey",message);
    }
}

8.4 创建消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
package com.lidong.dubbo.core.util.customer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
/**
 * @项目名称:lidong-dubbo
 * @类名:MessageConsumer
 * @类的描述: RabbitMQ 消息消费者
 * @作者:lidong
 * @创建时间:2017/2/4 上午9:33
 * @公司:chni
 * @QQ:1561281670
 * @邮箱:lidong1665@163.com
 */
public class MessageConsumer implements MessageListener {
    private Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
	@Override
	public void onMessage(Message message) {
		logger.info("------消费者处理消息------");
		logger.info("receive message",message);
	}
}

8.5 配置spring-rabbitmq.xml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
<?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/beans
     http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
     http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
     http://www.springframework.org/schema/rabbit
     http://www.springframework.org/schema/rabbit/spring-rabbit-1.6.xsd">
    <!--配置connection-factory，指定连接rabbit server参数 -->
    <rabbit:connection-factory id="rabbitConnectionFactory"
                               username="${rabbit_username}"
                               password="${rabbit_password}"
                               host="${rabbit_host}"
                               port="${rabbit_port}" />
    <!--定义rabbit template用于数据的接收和发送 -->
    <rabbit:template id="amqpTemplate"  connection-factory="rabbitConnectionFactory"
                     exchange="exchangeTest" />
    <!--通过指定下面的admin信息，当前producer中的exchange和queue会在rabbitmq服务器上自动生成 -->
    <rabbit:admin connection-factory="rabbitConnectionFactory" />
    <!--定义queue -->
    <rabbit:queue name="queueTest" durable="true" auto-delete="false" exclusive="false" />
    <!-- 定义direct exchange，绑定queueTest -->
    <rabbit:direct-exchange name="exchangeTest" durable="true" auto-delete="false">
        <rabbit:bindings>
            <rabbit:binding
                    queue="queueTest"
                    key="queueTestKey">
            </rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>
    <!-- 消息接收者 -->
    <bean id="messageReceiver" class="com.lidong.dubbo.core.util.customer.MessageConsumer"></bean>
    <!-- queue litener  观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象-->
    <rabbit:listener-container
            connection-factory="rabbitConnectionFactory">
        <rabbit:listener queues="queueTest"
                         ref="messageReceiver"/>
    </rabbit:listener-container>
</beans>

8.6 测试消息

1
2
3
4
5
6
7
8
9
10
11
@RequestMapping("/sendRabbitMessage")
@ResponseBody
public  void  sendRabbitMessage(){
    logger.info("--------------------");
    User user = new User();
    user.setName("lidong");
    user.setAge(25);
    user.setPassword("123456");
    user.setId(12);
    iMessageProducer.sendMessage(user);
}

注意：本例子采用的direct exchange交换机 。

代码地址