rabbirMQ学习笔记

Mac安装Rabbit

1.使用brew来安装 RabbitMQ

brew install rabbitmq

• 1

看到如下的代码表示RabbitMQ安装成功

localhost:~ lidong$ brew install rabbitmq
Updating Homebrew...
==> Installing dependencies for rabbitmq: jpeg, libpng, libtiff, wxmac, erlang
==> Installing rabbitmq dependency: jpeg
==> Downloading https://homebrew.bintray.com/bottles/jpeg-8d.sierra.bottle.2.tar
######################################################################## 100.0%
==> Pouring jpeg-8d.sierra.bottle.2.tar.gz
��  /usr/local/Cellar/jpeg/8d: 19 files, 708.3K
==> Installing rabbitmq dependency: libpng
==> Downloading https://homebrew.bintray.com/bottles/libpng-1.6.28.sierra.bottle
######################################################################## 100.0%
==> Pouring libpng-1.6.28.sierra.bottle.tar.gz
��  /usr/local/Cellar/libpng/1.6.28: 26 files, 1.2M
==> Installing rabbitmq dependency: libtiff
==> Downloading https://homebrew.bintray.com/bottles/libtiff-4.0.7_1.sierra.bott
######################################################################## 100.0%
==> Pouring libtiff-4.0.7_1.sierra.bottle.tar.gz
��  /usr/local/Cellar/libtiff/4.0.7_1: 248 files, 3.4M
==> Installing rabbitmq dependency: wxmac
==> Downloading https://homebrew.bintray.com/bottles/wxmac-3.0.2_4.sierra.bottle
######################################################################## 100.0%
==> Pouring wxmac-3.0.2_4.sierra.bottle.tar.gz
��  /usr/local/Cellar/wxmac/3.0.2_4: 810 files, 24.6M
==> Installing rabbitmq dependency: erlang
==> Downloading https://homebrew.bintray.com/bottles/erlang-19.2.sierra.bottle.t
######################################################################## 100.0%
==> Pouring erlang-19.2.sierra.bottle.tar.gz
==> Caveats
Man pages can be found in:
  /usr/local/opt/erlang/lib/erlang/man

Access them with `erl -man`, or add this directory to MANPATH.
==> Summary
��  /usr/local/Cellar/erlang/19.2: 7,310 files, 280.9M
==> Installing rabbitmq 
==> Using the sandbox
==> Downloading https://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitm
######################################################################## 100.0%
==> /usr/bin/unzip -qq -j /usr/local/Cellar/rabbitmq/3.6.6/plugins/rabbitmq_mana
==> Caveats
Management Plugin enabled by default at http://localhost:15672

Bash completion has been installed to:
  /usr/local/etc/bash_completion.d

To have launchd start rabbitmq now and restart at login:
  brew services start rabbitmq
Or, if you don't want/need a background service you can just run:
  rabbitmq-server
==> Summary
��  /usr/local/Cellar/rabbitmq/3.6.6: 188 files, 5.8M, built in 9 minutes 12 seconds

• 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
• 50
• 51

注意： rabbitmq的安装目录： /usr/local/Cellar/rabbitmq/3.6.6

3.RabbitMQ 的启动

进入到 /usr/local/Cellar/rabbitmq/3.6.6，执行

localhost:3.6.6 lidong$ sbin/rabbitmq-server

• 1

4.RabbitMQ 启动插件

待RabbitMQ 的启动完毕之后，另起终端进入cd /Users/lidong/javaEE/rabbitmq_server-3.6.6/sbin 。启动插件：

 sudo ./rabbitmq-plugins enable rabbitmq_management（执行一次以后不用再次执行）

• 1

5.登陆管理界面

浏览器输入：http://localhost:15672/

• 1
• 2

账号密码初始默认都为guest

 

1.Windows下安装RabbitMQ需要以下几个步骤

   (1)：下载erlang，原因在于RabbitMQ服务端代码是使用并发式语言erlang编写的，下载地址：http://www.erlang.org/downloads，双击.exe文件进行安装就好，安装完成之后创建一个名为ERLANG_HOME的环境变量，其值指向erlang的安装目录，同时将%ERLANG_HOME%\bin加入到Path中，最后打开命令行，输入erl，如果出现erlang的版本信息就表示erlang语言环境安装成功；

                      

 

   (2)：下载RabbitMQ，下载地址：http://www.rabbitmq.com/，同样双击.exe进行安装就好(这里需要注意一点，默认的安装目录是C:/Program Files/....，这个目录中是存在空格符的，我们需要改变安装目录，貌似RabbitMQ安装目录中是不允许有空格的，我之前踩过这个大坑)；

   (3)：安装RabbitMQ-Plugins，这个相当于是一个管理界面，方便我们在浏览器界面查看RabbitMQ各个消息队列以及exchange的工作情况，安装方法是：打开命令行cd进入rabbitmq的sbin目录(我的目录是：E:\software\rabbitmq\rabbitmq_server-3.6.5\sbin)，输入：rabbitmq-plugins enable rabbitmq_management命令，稍等会会发现出现plugins安装成功的提示，默认是安装6个插件，如果你在安装插件的过程中出现了下面的错误：        

   解决方法是：首先在命令行输入：rabbitmq-service stop，接着输入rabbitmq-service remove，再接着输入rabbitmq-service install，接着输入rabbitmq-service start，最后重新输入rabbitmq-plugins enable rabbitmq_management试试，我是这样解决的；

   (4)：插件安装完之后，在浏览器输入http://localhost:15672进行验证，你会看到下面界面，输入用户名：guest，密码：guest你就可以进入管理界面，当然用户名密码你都可以变的；

2.安装完RabbitMQ之后，我们先来简单了解下RabbitMQ中涉及到的几个概念

    producer：消息生产者

    consumer：消息消费者

     virtual host：虚拟主机，在RabbitMQ中，用户只能在虚拟主机的层面上进行一些权限设置，比如我可以访问哪些队列，我可以处理哪些请求等等；

     broker：消息转发者，也就是我们RabbitMQ服务端充当的功能了，那么消息是按照什么规则进行转发的呢？需要用到下面几个概念；

     exchange：交换机，他是和producer直接进行打交道的，有点类似于路由器的功能，主要就是进行转发操作的呗，那么producer到底用哪个exchange进行路由呢？这个取决于routing key(路由键)，每个消息都有这个键，我们也可以自己设定，其实就是一字符串；

     queue：消息队列，用于存放消息，他接收exchange路由过来的消息，我们可以对队列内容进行持久化操作，那么queue到底接收那个exchange路由的消息呢？这个时候就要用到binding key(绑定键)了，绑定键会将队列和exchange进行绑定，至于绑定方式，RabbitMQ提供了多种方式，大家可以看看鸿洋大神的RabbitMQ博客系列(点击查看)；

     以上就是RabbitMQ涉及到的一些概念了，用一张图表示这些概念之间的关系就是：

3.RabbitMQ简单使用

   producer(生产者)端步骤：

    (1)：创建ConnectionFactory，并且设置一些参数，比如hostname,portNumber等等

    (2)：利用ConnectionFactory创建一个Connection连接

    (3)：利用Connection创建一个Channel通道

    (4)：创建queue并且和Channel进行绑定

    (5)：创建消息，并且发送到队列中

     注意，在我们当前的例子中，并没有用到exchange交换机，RabbitMQ默认情况下是会创建一个空字符串名字的exchange的，如果我们没有创建自己的exchange的话，默认就是使用的这个exchange；

     producer端代码：

 

1. public class Sender {

2. private final static String QUEUE_NAME = "MyQueue";

3.  

4. public static void main(String[] args) {

5. send();

6. }

7.  

8. public static void send()

9. {

10. ConnectionFactory factory = null;

11. Connection connection = null;

12. Channel channel = null;

13. try {

14. factory = new ConnectionFactory();

15. factory.setHost("localhost");

16. connection = factory.newConnection();

17. channel = connection.createChannel();

18. channel.queueDeclare(QUEUE_NAME, false, false, false, null);

19. String message = "my first message .....";

20. channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));

21. System.out.println("已经发送消息....."+message);

22. } catch (IOException e) {

23. e.printStackTrace();

24. } catch (TimeoutException e) {

25. e.printStackTrace();

26. }finally{

27. try {

28. //关闭资源

29. channel.close();

30. connection.close();

31. } catch (IOException e) {

32. e.printStackTrace();

33. } catch (TimeoutException e) {

34. e.printStackTrace();

35. }

36. }

37. }

38. }

 

     consumer(消费者)端步骤：

     (1)：创建ConnectionFactory，并且设置一些参数，比如hostname,portNumber等等

     (2)：利用ConnectionFactory创建一个Connection连接

     (3)：利用Connection创建一个Channel通道

     (4)：将queue和Channel进行绑定，注意这里的queue名字要和前面producer创建的queue一致

     (5)：创建消费者Consumer来接收消息，同时将消费者和queue进行绑定

     consumer端代码：

 

1. public class Receiver {

2. private final static String QUEUE_NAME = "MyQueue";

3.  

4. public static void main(String[] args) {

5. receive();

6. }

7.  

8. public static void receive()

9. {

10. ConnectionFactory factory = null;

11. Connection connection = null;

12. Channel channel = null;

13.  

14. try {

15. factory = new ConnectionFactory();

16. factory.setHost("localhost");

17. connection = factory.newConnection();

18. channel = connection.createChannel();

19. channel.queueDeclare(QUEUE_NAME, false, false, false, null);

20. Consumer consumer = new DefaultConsumer(channel){

21. @Override

22. public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,

23. byte[] body) throws IOException {

24. System.out.println("11111111111");

25. String message = new String(body, "UTF-8");

26. System.out.println("收到消息....."+message);

27. }};

28. channel.basicConsume(QUEUE_NAME, true,consumer);

29. } catch (IOException e) {

30. e.printStackTrace();

31. } catch (TimeoutException e) {

32. e.printStackTrace();

33. }finally{

34. try {

35. //关闭资源

36. channel.close();

37. connection.close();

38. } catch (IOException e) {

39. e.printStackTrace();

40. } catch (TimeoutException e) {

41. e.printStackTrace();

42. }

43. }

44. }

45. }

   好了，这篇先到这了，下一篇我会简单介绍点更深入的东西，后续也会对RabbitMQ原生API进行封装，便于我们自己开发；

 

与Spring集成

spring配置RabbitMQ的发送和接收demo

首先，整个demo包含6个文件： 
spring-rabbitmq.xml：配置rabbitmq的queue，exchange，已经发送和接受的bean； 
MQProducer.java：发送消息到指定队列的接口； 
MQProducerImpl.java：发送消息到指定队列的实现类； 
QueueListenter.java：监听queue消息并接收消费； 
TestQueue.java：测试类； 
sys.properties：配置rabbitmq的地址，账户，密码，端口等

具体内容如下： 
1、spring-rabbitmq.xml

<?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"
    xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemaLocation="
    http://www.springframework.org/schema/beans  
    http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/rabbit 
    http://www.springframework.org/schema/rabbit/spring-rabbit-1.6.xsd
    http://www.springframework.org/schema/beans 
    http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
    http://www.springframework.org/schema/context 
    http://www.springframework.org/schema/context/spring-context-3.1.xsd
    http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd">

    <description>rabbitmq 连接服务配置</description>

    <!-- 基于注解的根包扫描路径 -->
    <context:component-scan base-package="com.mdl.activemq" />
    <context:component-scan base-package="com.mdl.activemq.services" />

    <!-- 加载内网系统配置 -->
    <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="properties" ref="configProperties" />  
    </bean> 

    <bean id="configProperties" class="org.springframework.beans.factory.config.PropertiesFactoryBean">
        <property name="locations">
            <list>
                <value>classpath:sys.properties</value>
            </list>
        </property>
    </bean>
    <!-- 消息对象json转换类 -->
    <bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /> 

    <!-- 连接配置 -->
    <rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}" password="${mq.password}" port="5672" />

    <rabbit:admin connection-factory="connectionFactory"/>

    <!-- spring template声明-->
    <rabbit:template id="amqpTemplate" exchange="xk-test-mq-exchange" connection-factory="connectionFactory"  message-converter="jsonMessageConverter"/>

    <!-- 声明一个Que -->
    <!-- durable:是否持久化 ；      exclusive: 仅创建者可以使用的私有队列，断开后自动删除；    auto_delete: 当所有消费客户端连接断开后，是否自动删除队列  --> 
    <rabbit:queue id="xk_test_queue" name="xk_test_queue" durable="true" auto-delete="false" exclusive="false" />

     <!-- 声明一个Exchange -->
    <rabbit:direct-exchange name="xk-test-mq-exchange" durable="true" auto-delete="false" id="xk-test-mq-exchange">
        <rabbit:bindings>
            <rabbit:binding queue="xk_test_queue" key="xk_test_queue"/>
        </rabbit:bindings>
    </rabbit:direct-exchange>
    <!--
    rabbit:direct-exchange：定义exchange模式为direct，意思就是消息与一个特定的路由键完全匹配，才会转发。
    rabbit:binding：设置消息queue匹配的key
    -->



    <!-- 监听器 -->
    <bean id="queueListenter" class="com.mdl.activemq.listener.QueueListenter"/>

    <!-- 配置监听queue -->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto">
            <rabbit:listener queues="xk_test_queue" ref="queueListenter"/>
    </rabbit:listener-container>


</beans>

• 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
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72

2、MQProducer.java

package com.mdl.activemq.services;

public interface MQProducer {
    /**
     * 发送消息到指定队列
     * @param queueKey
     * @param object
     */
    void sendDataToQueue(String queueKey, Object object);
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11

3、MQProducerImpl.java

package com.mdl.activemq.services.impl;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

import com.mdl.activemq.services.MQProducer;

//或者@Repository
@Service
public class MQProducerImpl implements MQProducer {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Override
    public void sendDataToQueue(String queueKey, Object object) {
        try {
            System.out.println("=========发送消息开始=============消息：" + object.toString());
            amqpTemplate.convertAndSend(queueKey, object);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

• 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

4、QueueListenter.java

package com.mdl.activemq.listener;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;

public class QueueListenter implements MessageListener {

    @Override
    public void onMessage(Message message) {
        String str = "";
        try
        {
            str = new String(message.getBody(), "UTF-8");
            System.out.println("=============监听【QueueListenter】消息"+message);
            System.out.print("=====获取消息"+str);

        }catch(Exception e)
        {
            e.printStackTrace();
        }

    }

}

• 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

5、sys.properties

#rabbitmq
mq.host=172.29.231.70
mq.username=admin
mq.password=admin
mq.port=5672

• 1
• 2
• 3
• 4
• 5
• 6

结果：

log4j:WARN No appenders could be found for logger (org.springframework.test.context.junit4.SpringJUnit4ClassRunner).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
=========发送消息开始=============消息：{data=hello,rabbmitmq!}
=============监听【QueueListenter】消息(Body:'{"data":"hello,rabbmitmq!"}' MessageProperties [headers={__ContentTypeId__=java.lang.Object, __KeyTypeId__=java.lang.Object, __TypeId__=java.util.HashMap}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/json, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=xk-test-mq-exchange, receivedRoutingKey=xk_test_queue, receivedDelay=null, deliveryTag=1, messageCount=0, consumerTag=amq.ctag-fPcPLEkIxiq0zUgwvOP_eQ, consumerQueue=xk_test_queue])
=====获取消息{"data":"hello,rabbmitmq!"}

 

SpringBoot-RabbitMQ（消息队列）

RabbitMQ简介

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

相关概念

Queue

Queue（队列）是RabbitMQ的内部对象，用于存储消息。

queue

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。

queue-one

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

queue-two

Exchange

通常我们谈到队列服务，会有三个概念：发消息者、队列、收消息者，RabbitMQ在这个基本概念之上，多做了一层抽象，在发消息者和队列之间，加入了交换器（Exchange）。这样发消息者和队列就没有直接联系，转而变成发消息者把消息给交换器，交换器根据调度策略再把消息再给队列。

exchange

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers四种。

fanout

把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

exchange-fanout

• 生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

direct

把消息路由到bindingKey与routingKey完全匹配的Queue中。

exchange-direct

• routingKey=”error”发送消息，则会同时路由到Queue1（amqp.gen-S9b…）和Queue2（amqp.gen-Agl…）
• routingKey=”info”或routingKey=”warning”发送消息，则只会路由到Queue2
• 以其他routingKey发送消息，则不会路由到这两个Queue中

topic

把消息路由到bindingKey与routingKey模糊匹配的Queue中，匹配规则如下：

• routingKey为一个句点号“.”分隔的字符串（被句点号“.”分隔开的每一段独立的字符串称为一个单词）
• bindingKey与routingKey一样也是句点号“.”分隔的字符串
• bindingKey中可以存在两种特殊字符“”与“#”，用于做模糊匹配，其中“”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）

exchange-topic

• routingKey=”quick.orange.rabbit”发送信息，则会同时路由到Q1与Q2
• routingKey=”lazy.orange.fox”发送信息，则只会路由到Q1
• routingKey=”lazy.brown.fox”发送消息，则只会路由到Q2
• routingKey=”lazy.pink.rabbit”发送消息，则只会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）
• routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”发送消息，则会被丢弃，它们并没有匹配任何bindingKey

headers

headers类型的Exchange不依赖于routingKey与bindingKey的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。

Web开发中的"路由

路由就是URL到函数的映射。

在web开发中，“route”是指根据url， 分配到对应的处理程序。

路由： 
就是一个路径的解析，根据客户端提交的路径，将请求解析到相应的控制器上； 
从 URL 找到处理这个 URL 的类和函数。

参考文献： 
1.理解Web路由 
2.Web开发中的”路由”是什么意思？ 
3.web 开发中的路由是什么意思

RPC

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

rpc

RabbitMQ中实现RPC的机制是：

客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14中properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）

• 服务器端收到消息并处理
• 服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性
• 客户端之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理

安装RabbitMQ

此处不进行详细说明。

默认访问路径：http://localhost:15672

默认账号/密码：guest/guest

SpringBoot集成RabbitMQ

首先创建两个springboot工程,这是为了模拟分布式应用系统中,两个应用之间互相交流的过程,一个发送者(Sender),一个接收者(Receiver)

接收端的pom文件,application.properties,Queue配置类,App启动类都是一致的。

Direct模式相当于一对一模式,一个消息被发送者发送后,会被转发到一个绑定的消息队列中,然后被一个接收者接收!

HelloWorld

1. pom.xml添加依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2. application.properties配置

 

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

3. RabbitConfig队列配置

@Configuration
public class RabbitConfig {

    @Bean
    public Queue helloQueue() {
        return new Queue("hello");
    }

}

4. HelloSender发送者

@Component
public class HelloSender {

    private static Logger logger = Logger.getLogger(HelloSender.class);

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send(String message) {
        rabbitTemplate.convertAndSend("hello", message);
        logger.info(String.format("send message: %s", message));
    }

}

5. HelloReceiver接收者

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {

    private static Logger logger = Logger.getLogger(HelloReceiver.class);

    @RabbitHandler
    public void process(String message) {
        logger.info(String.format("receive message: %s", message));
    }

}

6. HelloTest测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class HelloTest {

    @Autowired
    private HelloSender helloSender;

    @Test
    public void test() {
        helloSender.send("hello world");
    }

}

7. 测试结果

send message: hello world
receive message: hello world

多对多使用

默认情况下，M个发送者，N个接收者，消息会均匀的发送到N个接收者中。

1. ManyTest测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class ManyTest {

    @Autowired
    private MessageSender1 messageSender1;
    @Autowired
    private MessageSender2 messageSender2;

    @Test
    public void manyToMany() {
        for (int i = 0; i < 10; i++) {
            messageSender1.send(String.format("hi(%d)", i + 1));
            messageSender2.send(String.format("hi(%d)", i + 1));
        }
    }

}

2. 测试结果

接收者1、2均匀接收到信息。

receive(1) message: hi(1)
receive(2) message: hi(1)
receive(1) message: hi(3)
receive(2) message: hi(3)
receive(1) message: hi(4)
receive(2) message: hi(4)
receive(1) message: hi(2)
receive(2) message: hi(2)
receive(1) message: hi(5)
receive(2) message: hi(5)
receive(1) message: hi(6)
receive(2) message: hi(6)
receive(1) message: hi(7)
receive(2) message: hi(7)
receive(1) message: hi(8)
receive(2) message: hi(8)
receive(1) message: hi(9)
receive(2) message: hi(9)
receive(1) message: hi(10)
receive(2) message: hi(10)

对象的支持

1. 发送者与接收者

// 发送者
public void send(User user) {
    rabbitTemplate.convertAndSend("object", user);
    logger.info(String.format("send object: %s", user));
}

...

// 接收者
@RabbitListener(queues = "object")
public void process(User user) {
    logger.info(String.format("receive object: %s", user));
}

2. 测试结果

SpringBoot完美的支持对象的发送和接收，不需要格外的配置。

send object: User{id=1, name=ConanLi}
receive object: User{id=1, name=ConanLi}

FanoutExchange

Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

1. FanoutRabbitConfig队列配置

@Configuration
public class FanoutRabbitConfig {

    @Bean
    public Queue fanoutAQueue() {
        return new Queue("fanout.a");
    }

    @Bean
    public Queue fanoutBQueue() {
        return new Queue("fanout.b");
    }

    @Bean
    public Queue fanoutCQueue() {
        return new Queue("fanout.c");
    }

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    public Binding bindingExchangeFanoutA(Queue fanoutAQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutAQueue).to(fanoutExchange);
    }

    @Bean
    public Binding bindingExchangeFanoutB(Queue fanoutBQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutBQueue).to(fanoutExchange);
    }

    @Bean
    public Binding bindingExchangeFanoutC(Queue fanoutCQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutCQueue).to(fanoutExchange);
    }

}

2. 发送者与接收者

// 发送者
public void send(String message) {
    rabbitTemplate.convertAndSend("fanoutExchange", "", message);
    logger.info(String.format("send fanout message: %s", message));
}

...

// 接收者A
@RabbitListener(queues = "fanout.a")
public void process(String message) {
    logger.info(String.format("receive fanout a message: %s", message));
}

...

// 接收者B
@RabbitListener(queues = "fanout.b")
public void process(String message) {
    logger.info(String.format("receive fanout b message: %s", message));
}

...

// 接收者C
@RabbitListener(queues = "fanout.c")
public void process(String message) {
    logger.info(String.format("receive fanout c message: %s", message));
}

3. 测试结果

接收者A、B、C均接收到信息。

send fanout message: email
receive fanout b message: email
receive fanout a message: email
receive fanout c message: email

TopicExchange

Topic是RabbitMQ中最灵活的一种方式，可以根据routing_Key*的绑定不同的队列。

1. TopicRabbitConfig队列配置

@Configuration
public class TopicRabbitConfig {

    @Bean
    public Queue topicAQueue() {
        return new Queue("topic.a");
    }

    @Bean
    public Queue topicAnyQueue() {
        return new Queue("topic.any");
    }

    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("topicExchange");
    }

    @Bean
    public Binding bindingExchangeTopicA(Queue topicAQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(topicAQueue).to(topicExchange).with("topic.a");
    }

    @Bean
    public Binding bindingExchangeTopicAny(Queue topicAnyQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(topicAnyQueue).to(topicExchange).with("topic.#");
    }

}

2. 发送者与接收者

// 发送者A
public void send(String message) {
    rabbitTemplate.convertAndSend("topicExchange", "topic.a", message);
    logger.info(String.format("send topic a message: %s", message));
}

...

// 发送者B
public void send(String message) {
    rabbitTemplate.convertAndSend("topicExchange", "topic.b", message);
    logger.info(String.format("send topic b message: %s", message));
}

...

// 发送者Any
public void send(String message) {
    rabbitTemplate.convertAndSend("topicExchange", "topic.any", message);
    logger.info(String.format("send topic any message: %s", message));
}

...

// 接收者A
@RabbitListener(queues = "topic.a")
public void process(String message) {
    logger.info(String.format("receive topic a message: %s", message));
}

...

// 接收者Any
@RabbitListener(queues = "topic.any")
public void process(String message) {
    logger.info(String.format("receive topic any message: %s", message));
}

3. 测试结果

接收到A接收到发送者A的信息。

接收者Any接收到发送者A、发送者B、发送者Any的信息。

send topic a message: tag
receive topic a message: tag
receive topic any message: tag

...

send topic b message: tag
receive topic any message: tag

...

send topic any message: tag
receive topic any message: tag

完整示例：GitHub
PS：本文使用的是spring-boot-1.4.4.RELEASE、rabbit-3.6.6

 

springboot整合RabbitMQ

一 RabbitMQ的介绍　　

　　RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿里巴巴公司的,现已经转让给apache).

　　消息中间件的工作过程可以用生产者消费者模型来表示.即,生产者不断的向消息队列发送信息,而消费者从消息队列中消费信息.具体过程如下:

　　从上图可看出,对于消息队列来说,生产者,消息队列,消费者是最重要的三个概念,生产者发消息到消息队列中去,消费者监听指定的消息队列,并且当消息队列收到消息之后,接收消息队列传来的消息,并且给予相应的处理.消息队列常用于分布式系统之间互相信息的传递.

　　对于RabbitMQ来说,除了这三个基本模块以外,还添加了一个模块,即交换机(Exchange).它使得生产者和消息队列之间产生了隔离,生产者将消息发送给交换机,而交换机则根据调度策略把相应的消息转发给对应的消息队列.那么RabitMQ的工作流程如下所示:

　　紧接着说一下交换机.交换机的主要作用是接收相应的消息并且绑定到指定的队列.交换机有四种类型,分别为Direct,topic,headers,Fanout.

　　Direct是RabbitMQ默认的交换机模式,也是最简单的模式.即创建消息队列的时候,指定一个BindingKey.当发送者发送消息的时候,指定对应的Key.当Key和消息队列的BindingKey一致的时候,消息将会被发送到该消息队列中.

　　topic转发信息主要是依据通配符,队列和交换机的绑定主要是依据一种模式(通配符+字符串),而当发送消息的时候,只有指定的Key和该模式相匹配的时候,消息才会被发送到该消息队列中.

　　headers也是根据一个规则进行匹配,在消息队列和交换机绑定的时候会指定一组键值对规则,而发送消息的时候也会指定一组键值对规则,当两组键值对规则相匹配的时候,消息会被发送到匹配的消息队列中.

　　Fanout是路由广播的形式,将会把消息发给绑定它的全部队列,即便设置了key,也会被忽略.

二.SpringBoot整合RabbitMQ(Direct模式)

　　SpringBoot整合RabbitMQ非常简单!感觉SpringBoot真的极大简化了开发的搭建环境的时间..这样我们程序员就可以把更多的时间用在业务上了,下面开始搭建环境:

　　首先创建两个maven工程,这是为了模拟分布式应用系统中,两个应用之间互相交流的过程,一个发送者(Sender),一个接收者(Receiver)

　　紧接着,配置pom.xml文件,注意其中用到了springboot对于AMQP(高级消息队列协议,即面向消息的中间件的设计)

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.4.0.RELEASE</version>
    </parent>
    <properties>
        <java.version>1.7</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <optional>true</optional>
            <scope>true</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <!-- 添加springboot对amqp的支持 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-tomcat</artifactId>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.tomcat.embed</groupId>
            <artifactId>tomcat-embed-jasper</artifactId>
            <scope>provided</scope>
        </dependency>
    </dependencies>

　　紧接着,我们编写发送者相关的代码.首先毫无疑问,要书写启动类:

@SpringBootApplication
public class App{
    public static void main(String[] args) {
        SpringApplication.run(App.class, args);
    }
}

　　接着在application.properties中,去编辑和RabbitMQ相关的配置信息,配置信息的代表什么内容根据键就能很直观的看出了.这里端口是5672,不是15672...15672是管理端的端口!

spring.application.name=spirng-boot-rabbitmq-sender
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

　　随后,配置Queue(消息队列).那注意由于采用的是Direct模式,需要在配置Queue的时候,指定一个键,使其和交换机绑定.

@Configuration
public class SenderConf {
     @Bean
     public Queue queue() {
          return new Queue("queue");
     }
}

　　接着就可以发送消息啦!在SpringBoot中,我们使用AmqpTemplate去发送消息!代码如下:

@Component
public class HelloSender {
    @Autowired
    private AmqpTemplate template;
    
    public void send() {
    template.convertAndSend("queue","hello,rabbit~");
    }
}

　　编写测试类!这样我们的发送端代码就编写完了~

@SpringBootTest(classes=App.class)
@RunWith(SpringJUnit4ClassRunner.class)
public class TestRabbitMQ {
    
    @Autowired
    private HelloSender helloSender;

    @Test
    public void testRabbit() {
        helloSender.send();
    }
}

　　接着我们编写接收端.接收端的pom文件,application.properties(修改spring.application.name),Queue配置类,App启动类都是一致的!这里省略不计.主要在于我们需要配置监听器去监听绑定到的消息队列,当消息队列有消息的时候,予以接收,代码如下:

@Component
public class HelloReceive {

    @RabbitListener(queues="queue")    //监听器监听指定的Queue
    public void processC(String str) {
        System.out.println("Receive:"+str);
    }

}

　　接下来就可以测试啦,首先启动接收端的应用,紧接着运行发送端的单元测试,接收端应用打印出来接收到的消息,测试即成功!

　　需要注意的地方,Direct模式相当于一对一模式,一个消息被发送者发送后,会被转发到一个绑定的消息队列中,然后被一个接收者接收!

　　实际上RabbitMQ还可以支持发送对象:当然由于涉及到序列化和反序列化,该对象要实现Serilizable接口.HelloSender做出如下改写:

public void send() {

        User user=new User();    //实现Serializable接口
        user.setUsername("hlhdidi");
        user.setPassword("123");
        template.convertAndSend("queue",user);
    
}

　　HelloReceiver做出如下改写:

@RabbitListener(queues="queue")    //监听器监听指定的Queue
    public void process1(User user) {    //用User作为参数
        System.out.println("Receive1:"+user);
    }

三.SpringBoot整合RabbitMQ(Topic转发模式)

　　首先我们看发送端,我们需要配置队列Queue,再配置交换机(Exchange),再把队列按照相应的规则绑定到交换机上:

@Configuration
public class SenderConf {

        @Bean(name="message")
        public Queue queueMessage() {
            return new Queue("topic.message");
        }

        @Bean(name="messages")
        public Queue queueMessages() {
            return new Queue("topic.messages");
        }

        @Bean
        public TopicExchange exchange() {
            return new TopicExchange("exchange");
        }

        @Bean
        Binding bindingExchangeMessage(@Qualifier("message") Queue queueMessage, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
        }

        @Bean
        Binding bindingExchangeMessages(@Qualifier("messages") Queue queueMessages, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");//*表示一个词,#表示零个或多个词
        }
}

 　　而在接收端,我们配置两个监听器,分别监听不同的队列:

@RabbitListener(queues="topic.message")    //监听器监听指定的Queue
    public void process1(String str) {    
        System.out.println("message:"+str);
    }
    @RabbitListener(queues="topic.messages")    //监听器监听指定的Queue
    public void process2(String str) {
        System.out.println("messages:"+str);
    }

　　好啦!接着我们可以进行测试了!首先我们发送如下内容:

　　方法的第一个参数是交换机名称,第二个参数是发送的key,第三个参数是内容,RabbitMQ将会根据第二个参数去寻找有没有匹配此规则的队列,如果有,则把消息给它,如果有不止一个,则把消息分发给匹配的队列(每个队列都有消息!),显然在我们的测试中,参数2匹配了两个队列,因此消息将会被发放到这两个队列中,而监听这两个队列的监听器都将收到消息!那么如果把参数2改为topic.messages呢?显然只会匹配到一个队列,那么process2方法对应的监听器收到消息!

四.SpringBoot整合RabbitMQ(Fanout Exchange形式)

　　那前面已经介绍过了,Fanout Exchange形式又叫广播形式,因此我们发送到路由器的消息会使得绑定到该路由器的每一个Queue接收到消息,这个时候就算指定了Key,或者规则(即上文中convertAndSend方法的参数2),也会被忽略!那么直接上代码,发送端配置如下:

@Configuration
public class SenderConf {

        @Bean(name="Amessage")
        public Queue AMessage() {
            return new Queue("fanout.A");
        }


        @Bean(name="Bmessage")
        public Queue BMessage() {
            return new Queue("fanout.B");
        }

        @Bean(name="Cmessage")
        public Queue CMessage() {
            return new Queue("fanout.C");
        }

        @Bean
        FanoutExchange fanoutExchange() {
            return new FanoutExchange("fanoutExchange");//配置广播路由器
        }

        @Bean
        Binding bindingExchangeA(@Qualifier("Amessage") Queue AMessage,FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(AMessage).to(fanoutExchange);
        }

        @Bean
        Binding bindingExchangeB(@Qualifier("Bmessage") Queue BMessage, FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(BMessage).to(fanoutExchange);
        }

        @Bean
        Binding bindingExchangeC(@Qualifier("Cmessage") Queue CMessage, FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(CMessage).to(fanoutExchange);
        }
    
}

 

　　发送端使用如下代码发送:

　　接收端监听器配置如下:

@Component
public class HelloReceive {
    @RabbitListener(queues="fanout.A")
    public void processA(String str1) {
        System.out.println("ReceiveA:"+str1);
    }
    @RabbitListener(queues="fanout.B")
    public void processB(String str) {
        System.out.println("ReceiveB:"+str);
    }
    @RabbitListener(queues="fanout.C")
    public void processC(String str) {
        System.out.println("ReceiveC:"+str);
    }
    
    
}

　　运行测试代码,发现三个监听器都接收到了数据,测试成功!

Spring AMQP

目录

• Message
• Exchange
• Queue
• Binding
• ConnectionFactory
• Routing Connection Factory
• AmqpTemplate
• Sending messages
• Receiving messages
• Asynchronous Consumer异步接受消息
  • MessageListener 异步接收消息，MessageListener 定义比较简单。
  • 利用注解@RabbitListener实现异步接受消息
• Message Converters
  • SimpleMessageConverter
  • ContentTypeDelegatingMessageConverter
• 代码示例

 

---

Spring AMQP 是基于 Spring 框架的AMQP消息解决方案，提供模板化的发送和接收消息的抽象层，提供基于消息驱动的 POJO的消息监听等，很大方便我们使用RabbitMQ程序的相关开发。

Spring AMQP包含一些模块，如：spring-amqp, spring-rabbit and spring-erlang等，每个模块分别由独立的一些Jar包组成.

Spring AMQP模块主要包含org.springframework.amqp.core这个包中。这个包定义的相关类主要是与前面讲的AMQP模型相对应。Spring AMQP的目的是提供不依赖于任何特定的AMQP代理实现或客户端库通用的抽象。最终用户代码将很容易实现更易替换、添加和删除AMQP，因为它可以只针对抽象层来开发。这可以很方便我们选择和使用哪一个具体的broker实现，如sping-rabbit实现。

回到顶部

Message

Spring AMQP定义了Message类, 它定义了一个更一般的AMQP域模型,它是Spring AMQP的很重要的一部分，Message消息是当前模型中所操纵的基本单位，它由Producer产生，经过Broker被Consumer所消费。它是生产者和消费者发送和处理的对象。Message封装了body及MessageProperties，很方便了API的调用，它的定义如下:

public class Message {

　　private final MessageProperties messageProperties;
　　private final byte[] body;
 
　　public Message(byte[] body, MessageProperties messageProperties) {
　　　　this.body = body;
　　　　this.messageProperties = messageProperties;
　　}
 
　　public byte[] getBody() {
　　　　return this.body;
　　}
 
　　public MessageProperties getMessageProperties() {
　　　　return this.messageProperties;
　　}
}

MessageProperties 定义了多种常用的属性，如：messageId', 'timestamp', 'contentType'等等。

回到顶部

Exchange

Exchange接口代表了AMQP Exchange，它是消息发送的地方。在虚拟主机的消息协商器(Broker)中，每个Exchange都有唯一的名字。

Exchange包含4种类型：Direct, Topic, Fanout, Headers。不同的类型，他们如何处理绑定到队列方面的行为会有所不同。

1)Direct类型: 允许一个队列通过一个固定的Routing-key(通常是队列的名字)进行绑定。 Direct交换器将消息根据其routing-key属性投递到包含对应key属性的绑定器上。

2)Topic类型: 支持消息的Routing-key用*或#的模式，进行绑定。*匹配一个单词，#匹配0个或者多个单词。例如，binding key *.user.# 匹配routing key为 usd.user和eur.user.db，但是不匹配user.hello。

3)Fanout类型：它只是将消息广播到所有绑定到它的队列中,而不考虑routing key的值。

4)Header类型： 它根据应用程序消息的特定属性进行匹配，这些消息可能在binding key中标记为可选或者必选

注意：AMQP规范要求，任何的消息协商器(broker)需要提供一个没有名称的"default" Direct Exchange。

它的相关定义如下：

public interface Exchange {
　　String getName();
　　String getExchangeType();
　　boolean isDurable();
　　boolean isAutoDelete();
　　Map getArguments();
}

回到顶部

Queue

Queue,队列，它代表了Message Consumer接收消息的地方，它用来保存消息直到发送给消费者。Queue有以下一些重要的属性。

1)持久性：如果启用，队列将会在消息协商器(Broker)重启前都有效。

2)自动删除：如果启用，那么队列将会在所有的消费者停止使用之后自动删除掉自身。

3)惰性：如果没有声明队列，那么在执行到使用的时候会导致异常，并不会主动声明。

4)排他性：如果启用，队列只能被声明它的消费者使用。

public class Queue  {
 
    private final String name;
    private volatile boolean durable;
    private volatile boolean exclusive;
    private volatile boolean autoDelete;
    private volatile Map arguments;
 
    public Queue(String name) {
    this(name, true, false, false);
    }
 
    // Getters and Setters omitted for brevity    

回到顶部

Binding

生产者发送消息到Exchange,接收者从Queue接收消息，而绑定(Binging)是生产者和消费者消息传递的重要连接，它是连接生产者和消费者进行信息交流的关键。

Binging实例本身仅仅是代表持有连接的数据信息。不过它可以被AmqpAdmin这个类用来实际触发broker上的绑定操作。同时它可以在程序启动时，简化 Queues, Exchanges, and Bindings的定义及一些操作。

1、下面我们看一下绑定Queue到Exchange的一些基本选项及例子。

你可以用一个固定的RoutingKey将Queue绑定到DirectExchange 

例如：new Binding(someQueue, someDirectExchange, "foo.bar")

2、也可以用*、#模式匹配的RoutingKey方式,绑定到TopicExchange。

例如:  new Binding(someQueue, someTopicExchange, "foo.*")

3、绑定到一个广播的FanoutExchange,它没有RoutingKey

例如:  new Binding(someQueue, someFanoutExchange)

4、当然你也可以用fluent API的方式。

Binding b = BindingBuilder.bind(someQueue).to(someTopicExchange).with("foo.*");

回到顶部

ConnectionFactory

Spring AMQP中连接和资源的管理，尤其是spring-rabbit这个模块，因为spring-rabbit是 RabbitMQ 的唯一支持的实现。

在spring-rabbit中，管理消息协商器(broker)连接的核心组件是ConnectionFactory这个接口。 ConnectionFactory提供了

org.springframework.amqp.rabbit.connection.Connection(com.rabbitmq.client.Connection的包装类)实例的连接与管理。而CachingConnectionFactory是ConnectionFactory的在Spring AMQP中唯一实现，它创建一个连接代理，使程序可以共享的连接。

Connection 提供一个createChannel的方法。CachingConnectionFactory 的实现能支持channels的缓存,并且能根据区分是事务性或非事务性各自独立。同时，CachingConnectionFactory也提供hostname的构造函数，并且可以设置username、password、setChannelCacheSize等方法。CachingConnectionFactory  默认channel cache 大小为1，如果想改变可以用setChannelCacheSize设置channel cache size的大小。

@Bean
public ConnectionFactory rabbitConnectionFactory() {
    CachingConnectionFactory factory = new CachingConnectionFactory(
            environment.getProperty("rabbitmq.host"),
            environment.getProperty("rabbitmq.port", Integer.class)
    );
    factory.setUsername(environment.getProperty("rabbitmq.username"));
    factory.setPassword(environment.getProperty("rabbitmq.password"));
    return factory;
}
 Connection connection = factory.createConnection();

回到顶部

Routing Connection Factory

从spring-rabbit 1.3版本开始，AbstractRoutingConnectionFactory 被引入进来，它提供了一个这样的途径来配置许多的不同的Connection Factory的映射，并且能够根据运行时的lookupKey(通过绑定线程上下文的方式) 来决定使用哪个具体的Connection Factory。

 为了方便Spring AMQP提供了 AbstractRoutingConnectionFactory 的具体实现SimpleRoutingConnectionFactory。它是从SimpleResourceHolder中获得当前线程绑定的lookupKey。

    Map factories = new HashMap<Object, org.springframework.amqp.rabbit.connection.ConnectionFactory>(2);
    factories.put("foo", connectionFactory1);
    factories.put("bar", connectionFactory2);

    AbstractRoutingConnectionFactory connectionFactory = new SimpleRoutingConnectionFactory();
    connectionFactory.setTargetConnectionFactories(factories);

    final RabbitTemplate template = new RabbitTemplate(connectionFactory);
    Expression expression = new SpelExpressionParser()
        .parseExpression("T(org.springframework.amqp.rabbit.core.RabbitTemplateTests)" +
            ".LOOKUP_KEY_COUNT.getAndIncrement() % 2 == 0 ? 'foo' : 'bar'");

回到顶部

AmqpTemplate

Spring AMQP提供了一个发送和接收消息的操作模板类AmqpTemplate。 AmqpTemplate它定义包含了发送和接收消息等的一些基本的操作功能。RabbitTemplate是AmqpTemplate的一个实现。

RabbitTemplate支持消息的确认与返回，为了返回消息，RabbitTemplate 需要设置mandatory 属性为true,并且CachingConnectionFactory 的publisherReturns属性也需要设置为true。返回的消息会根据它注册的RabbitTemplate.ReturnCallback setReturnCallback 回调发送到给客户端，

一个RabbitTemplate仅能支持一个ReturnCallback 。

为了确认Confirms消息, CachingConnectionFactory 的publisherConfirms 属性也需要设置为true，确认的消息会根据它注册的RabbitTemplate.ConfirmCallback setConfirmCallback回调发送到给客户端。一个RabbitTemplate也仅能支持一个ConfirmCallback.

回到顶部

Sending messages

利用AmqpTemplate模板来发送消息。

AmqpTemplate提供了以下的几个方法来发送消息

void send(Message message) throws AmqpException;

void send(String routingKey, Message message) throws AmqpException;

void send(String exchange, String routingKey, Message message) throws AmqpException;

发送时，需要指定Exchange和routingKey. 如:

amqpTemplate.send("marketData.topic", "quotes.nasdaq.FOO",new Message("12.34".getBytes(), someProperties));

从Spring-Rabbit版本1.3开始，它提供了MessageBuilder 和 MessagePropertiesBuilder两个类，这可以方便我们以fluent流式的方式创建Message及MessageProperties对象。

Message message = MessageBuilder.withBody("foo".getBytes())

.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)

.setMessageId("123")

.setHeader("bar", "baz")

.build();

回到顶部

Receiving messages

接收消息可以有两种方式。比较简单的一种方式是，直接去查询获取消息，即调用receive方法,如果该方法没有获得消息，receive方法不阻塞，直接返回null。另一种方式是注册一个Listener监听器,一旦有消息到时来，异步接收。

注意: 接收消息，都需要从queue中获得,接收消息时，RabbitTemplate需要指定queue，或者设置默认的queue。

API提供的直接获得消息方法,如下

Message receive() throws AmqpException; 

Message receive(String queueName) throws AmqpException;

AmqpTemplate也提供了方法来直接接收POJOs对象(代替Message对象)，同时也提供了各种的MessageConverter用来处理返回的Object对象。

Object receiveAndConvert() throws AmqpException;

Object receiveAndConvert(String queueName) throws AmqpException;

AmqpTemplate 从Spring-Rabbit 1.3版本开始也提供了 receiveAndReply 方法来异步接收、处理及回复消息。

AmqpTemplate 要注意receive和reply阶段，大数时情况下，你需要提供仅仅一个ReceiveAndReplyCallback 的实现，用它来处理接收到消息和回复(对象)消息的业务逻辑。同时也要注意，ReceiveAndReplyCallback 可能返回null,在这种情况下，receiveAndReply 就相当于receive 方法。

boolean receiveAndReply(ReceiveAndReplyCallback callback) throws AmqpException;

boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback) throws AmqpException;

boolean receiveAndReply(ReceiveAndReplyCallback callback, String replyExchange, String replyRoutingKey) throws AmqpException;

boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback,String replyExchange, String replyRoutingKey) throws AmqpException;

回到顶部

Asynchronous Consumer异步接受消息

为了异步接收消息，Spring AMQP也提供了多种不同的实现方式。如:通过实现MessageListener的方式，或者通过注解@RabbitListener的方式等，来实现异步接受消息及处理。

MessageListener 异步接收消息，MessageListener 定义比较简单。

public interface MessageListener {

void onMessage(Message message);

}

如果你的程序需要依赖Channel,你需要用 ChannelAwareMessageListener这个接口。

public interface ChannelAwareMessageListener {

void onMessage(Message message, Channel channel) throws Exception;

}

如果你想将你的程序及Message API严格分开的话，可以用MessageListenerAdapter这个适配器类

MessageListenerAdapter listener = new MessageListenerAdapter(somePojo);

listener.setDefaultListenerMethod("myMethod");

看一下Container 容器。一般地，容器是处于“主动”的责任，以使侦听器回调可以保持被动触发处理。容器是生命周期的组件，它提供了开始和停止容器的方法。在配置容器时，我们需要配置AMQP Queue和MessageListener的对应连接,需要配置ConnectionFactory 的一个引用以及Listener、以及能够从该Queue中接收消息的Queue的引用,这样的Container才能通知到对应的Listener。

SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();

container.setConnectionFactory(rabbitConnectionFactory);

container.setQueueNames("some.queue");

container.setMessageListener(new MessageListenerAdapter(somePojo));

从RabbitMQ 的3.2版本开始， broker支持消费消息的priority 优先级。只需要在SimpleMessageListenerContainer 添加priority属性的设置。

<rabbit:listener-container connection-factory="rabbitConnectionFactory">

<rabbit:listener queues="some.queue" ref="somePojo" method="handle" priority="10" />

利用注解@RabbitListener实现异步接受消息

从spring-rabbit 1.4版本开始，可心利用注解的@RabbitListener来异步接收消息，它是更为简便的方式。

@Component

public class MyService {

@RabbitListener(queues = "myQueue")

public void processOrder(String data) { ... }

}

在容器中，需要配置@EnableRabbit，来支持@RabbitListener起作用。

@Configuration

@EnableRabbit

public class AppConfig { 

　　@Bean

　　public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {

　　　　SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();

　　　　factory.setConnectionFactory(connectionFactory());

　　　　factory.setConcurrentConsumers(3);

　　　　factory.setMaxConcurrentConsumers(10);

　　　　return factory;

　　}

}

通过上面的配置，就配置好Listener和Container了，就可以异步接收消息了。

回到顶部

Message Converters

AmqpTemplate 定义提供了各种发送和接收委拖给MessageConverter转化对象消息的方法。MessageConverter 本身比较简单，它提供了消息对象的转化,可将object转化成Message 对象，或者将Message 对象转化成Object对象。它提供了默认的SimpleMessageConverter实现，以及第三方的MessageConverter，如Jackson2JsonMessageConverter，MarshallingMessageConverter等，来处理消息与对象之间的转换。 

SimpleMessageConverter

SimpleMessageConverter是Spring AMQP中MessageConverter的一个默认实现。

public interface MessageConverter {

　　Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException; 

　　Object fromMessage(Message message) throws MessageConversionException; 

}

为了自定义转化对象，你也可以第三方的MessageConverter,如使用  Jackson2JsonMessageConverter 或者MarshallingMessageConverter,其中Jackson2JsonMessageConverter提供了Json对象的转化，MarshallingMessageConverter对象则提供了对象的Marshaller和 Unmarshaller转化。 

ContentTypeDelegatingMessageConverter

从Spring-Rabbit 1.4.2开始,它提供了ContentTypeDelegatingMessageConverter,它能根据不同的MessageProperties属性(contentType)决定来委托给具体的哪一个MessageConverter。

回到顶部

代码示例

import org.junit.Test;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

public class CodeTest {
    @Test
    public void testSendAndReceive() {
　　　　//创建ConnectionFactory //注意: guest的用户只能够在localhost 127.0.0.1进行测试
        String hostname = "localhost";
        String username = "mytest";
        String password = "mytest";
        String virtualHost = "/";
        CachingConnectionFactory cf = new CachingConnectionFactory(hostname);
        cf.setUsername(username);
        cf.setPassword(password);
        cf.setVirtualHost(virtualHost);

        RabbitAdmin admin = new RabbitAdmin(cf);

　　　　//创建Exchange
        String exchangeName = "direct.test.exchange";
        DirectExchange exchange = new DirectExchange(exchangeName);
        admin.declareExchange(exchange);

　　　　//创建Queue
        String queueName = "direct.test.queue";
        Queue queue = new Queue(queueName, true, false, false, null);
        admin.declareQueue(queue);

　　　　//创建Binding
        String routingKey = "direct.test.queue";
        admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(routingKey));

　　　　//创建RabbitTemplate
        RabbitTemplate rabbitTemplate = new RabbitTemplate(cf);
        rabbitTemplate.setExchange(exchangeName);
        rabbitTemplate.setQueue(queueName);

　　　　//创建Message
        String messageStr = "this is direct message";
        Message message = MessageBuilder.withBody(messageStr.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN).build();
　　　　//根据routingKey发送消息
        System.out.println("message=" + message);
        rabbitTemplate.send(routingKey, message);

　　　　//接收消息
        Message resultMessage = rabbitTemplate.receive();
        System.out.println("resultMessage=" + resultMessage);
        if (resultMessage != null) {
            System.out.println("receive massage=" + new String(resultMessage.getBody()));
        }
    }
}

 异步接收消息实例

<rabbit:template id="rabbitTemplate" connection-factory="connectionFactory" exchange="direct.test.exchange" queue="direct.test.queue"/> 
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:direct-exchange name="direct.test.exchange" durable="true">
　　<rabbit:bindings>
　　　　<rabbit:binding queue="direct.test.queue" key="direct.test.queue">rabbit:binding>
　　<rabbit:bindings>
<rabbit:direct-exchange/>
<rabbit:queue name="direct.test.queue" durable="true" auto-delete="false" exclusive="false"/>
<rabbit:connection-factory id="connectionFactory" username="mytest" password="mytest" host="127.0.0.1" port="5672"/>

<bean id="myMessageListener" class="com.company.rabbitmqpro.service.MyMessageListener">bean>
<bean id="simpleMessageListenerContainerFactory" class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
　　<property name="connectionFactory" ref="connectionFactory"/>
　　<property name="queueNames" >
　　　　<list>
　　　　　　<value>direct.test.queuevalue>
　　　　<list>
　　<property/>
　　<property name="messageListener" ref="myMessageListener">property>
　　<property name="concurrentConsumers" value="3"/>
　　<property name="maxConcurrentConsumers" value="10"/>
<bean/>

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
public class MyMessageListener implements MessageListener{
　　@Override
　　public void onMessage(Message message) {
　　　　System.out.println("received: " + message);
　　}
}

import javax.annotation.Resource;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "classpath*:applicationContext.xml" })
public class ListenerTest {
　　@Resource
　　private RabbitTemplate rabbitTemplate;
　　@Test
　　public void testSendAsynListener() {
　　　　String sendMsg = "this is direct message";
　　　　Message message = MessageBuilder.withBody(sendMsg.getBytes())
　　　　　　　　　　.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
　　　　　　　　　　.build();
　　　　String routingKey = "direct.test.queue";
　　　　rabbitTemplate.send(routingKey, message);
　　　　System.out.println("send ok");
　　}
}