RabbitMQ简介，安装，Spring集成 demo，以及使用(spring集成在最下面)

一、 RabbitMQ简介

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。

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

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

下面将重点介绍RabbitMQ中的一些基础概念，了解了这些概念，是使用好RabbitMQ的基础。

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。ConnectionFactory为Connection的制造工厂。

Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

Queue

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



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



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

Message acknowledgment

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。

这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

另外pub message是没有ack的。

Message durability

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

Prefetch count

前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

Exchange

在上一节我们看到生产者将消息投递到Queue中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。



Exchange是按照什么逻辑将消息路由到Queue的？这个将在Binding一节介绍。

RabbitMQ中的Exchange有四种类型，不同的类型有着不同的路由策略，这将在Exchange Types一节介绍。

routing key

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。

在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。

RabbitMQ为routing key设定的长度限制为255 bytes。

Binding

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

Binding key

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。

在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。

binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

Exchange Types

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

fanout

fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。



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

direct（常用）

direct类型的Exchange路由规则也很简单，它会把消息路由到那些binding key与routing key完全匹配的Queue中。



以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

topic（常用）

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

• routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”

• binding key与routing key一样也是句点号“. ”分隔的字符串

• binding key中可以存在两种特殊字符“”与“#”，用于做模糊匹配，其中“”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）



以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，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不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。

在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。

该类型的Exchange没有用到过（不过也应该很有用武之地），所以不做介绍。

RPC

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。

但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。



RabbitMQ中实现RPC的机制是：

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

• 服务器端收到消息并处理

• 服务器端处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性

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

二、 RabbitMQ：安装、配置与使用初探

Windows 环境配置

rabbitMQ是一个在AMQP协议标准基础上完整的，可服用的企业消息系统。它遵循Mozilla Public License开源协议，采用 Erlang 实现的工业级的消息队列(MQ)服务器，Rabbit MQ 是建立在Erlang OTP平台上。

安装Erlang

RabbitMQ的运行需要erlang的支持，因此我们先安装erlang。

32位下载地址：http://www.erlang.org/download/otp_win32_18.2.1.exe

64位下载地址： http://www.erlang.org/download/otp_win64_18.2.1.exe

双击选择默认安装就好。

有的选择其他的安装方式，可能需要添加一下系统环境变量（正常安装的也要检查下）：

安装RabbitMQ

下载运行rabbitmq-server-3.6.5 ，需要其他版本或者32位系统的，可以去官网下载。

依旧可以不改变默认进行安装。

需要注意：默认安装的RabbitMQ 监听端口是5672

配置

激活 RabbitMQ’s Management Plugin

使用RabbitMQ 管理插件，可以更好的可视化方式查看Rabbit MQ 服务器实例的状态。

打开命令窗口：

输入命令：

"C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.5\sbin\rabbitmq-plugins.bat" enable rabbitmq_management

这样，就安装好插件了，是不是能使用了呢？别急，需要重启服务才行，使用命令：

（rabbitmq提供了更多插件，还有一个常用的插件trace关于日志的插件

http://blog.csdn.net/qq_22673717/article/details/78489601

）

net stop RabbitMQ && net start RabbitMQ

这时候的，也许会出现这种结果：



“发生错误：发生系统错误 5。 拒绝访问。”

这是什么鬼？查了下，原来，5代表的是：不是系统管理员权限。

问题解决方案：使用管理员打开cmd再执行此命令：



这样就结束了吗？当然没有。

创建用户，密码，绑定角色

使用rabbitmqctl控制台命令（位于C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.5\sbin>）来创建用户，密码，绑定权限等。

注意：安装路径不同的请看仔细啊。

rabbitmq的用户管理包括增加用户，删除用户，查看用户列表，修改用户密码。

查看已有用户及用户的角色：

rabbitmqctl.bat list_users

新增一个用户：

rabbitmqctl.bat add_user username password

此时来看下我们当前用户：



eric 后面没有“[administrator]”

这个administrator是干嘛用的呢？这就涉及到用户角色问题了：

按照我个人理解，rabbitmq用户角色可分为五类：超级管理员, 监控者, 策略制定者, 普通管理者以及其他。

(1) 超级管理员(administrator)

可登陆管理控制台(启用management plugin的情况下)，可查看所有的信息，并且可以对用户，策略(policy)进行操作。

(2) 监控者(monitoring)

可登陆管理控制台(启用management plugin的情况下)，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)

(3) 策略制定者(policymaker)

可登陆管理控制台(启用management plugin的情况下), 同时可以对policy进行管理。

(4) 普通管理者(management)

仅可登陆管理控制台(启用management plugin的情况下)，无法看到节点信息，也无法对策略进行管理。

(5) 其他的

无法登陆管理控制台，通常就是普通的生产者和消费者。

给 eric 变成 “超级管理员” 角色：

rabbitmqctl.bat set_user_tags username administrator

再看下结果：



当然，除了上面的administrator 还有 monitoring、policymaker、management、自定义名称 ，对应上面介绍到的不同的角色。

像我们人一样，我们角色除了是公司的员工，还是父母的孩子、子女的爸妈等，用户也可以同时具有多个角色,设置方式:

rabbitmqctl.bat  set_user_tags  username tag1 tag2 ...

如果觉得guest 这个不安全（它的默认密码是guest）,想更改密码，好办：

rabbitmqctl change_password userName newPassword

有的人也许会说，我就是看guest不爽，老子新增了administrator用户了，就是想干掉它，可以：

rabbitmqctl.bat delete_user username

这时，可能有人就要问了：命令框是只有一个用户了，你怎么操作啊？怎么查看执情况啊/

当然有路子啦：

使用浏览器打开 http://localhost:15672 访问Rabbit Mq的管理控制台，使用刚才创建的账号登陆系统：





其实，除了可查看所有的信息 ，上面的命令 增删改查、权限设置，都可以在这个页面完成，还可以依据业务需求设置策略(policy)，具体的就不在这里啰嗦了，大家可以看看网上的帖子。

权限设置

用户有了角色，那也需要权限设置啊，别急，慢慢来：

按照官方文档，用户权限指的是用户对exchange，queue的操作权限，包括配置权限，读写权限。

我们配置权限会影响到exchange、queue的声明和删除。

读写权限影响到从queue里取消息、向exchange发送消息以及queue和exchange的绑定(binding)操作。

例如： 将queue绑定到某exchange上，需要具有queue的可写权限，以及exchange的可读权限；向exchange发送消息需要具有exchange的可写权限；从queue里取数据需要具有queue的可读权限

权限相关命令为：

(1) 设置用户权限

rabbitmqctl  set_permissions  -p  VHostPath  User  ConfP  WriteP  ReadP

(2) 查看(指定hostpath)所有用户的权限信息

rabbitmqctl  list_permissions  [-p  VHostPath]

(3) 查看指定用户的权限信息

rabbitmqctl  list_user_permissions  User

(4) 清除用户的权限信息

rabbitmqctl  clear_permissions  [-p VHostPath]  User

linux环境安装

###安装erlang环境

这里我要把erlang安装在我的servers文件夹下（根据个人情况自己环境情况，需要装哪个地方都行）
1.切换到servers文件夹下
2.下载 otp_src_18.2.1.tar.gz
3.在servers下新建erlang文件夹
4.解压 otp_src_18.2.1.tar.gz 到/servers/erlang下
5.切换到erlang文件夹下
6.执行./configure -prefix=/servers/erlang  就会开始编译安装  会编译到 /servers/erlang 下
7.执行make和 make install
8.修改/etc/profile文件，增加下面的环境变量：
set erlang environment   export PATH=$PATH:/servers/erlang/bin
9.查看/etc/profile文件，确认环境变量是否添加 vi etc/profile

[root@localhost ~]# cd /servers
[root@localhost servers]#  wget http://erlang.org/download/otp_src_18.2.1.tar.gz [root@localhost servers]#  mkdir /serves/erlang
[root@localhost servers]#  tar xvfz otp_src_18.2.1.tar.gz /servers/erlang
[root@localhost servers]#  cd /erlang
[root@localhost servers]#  ./configure -prefix=/servers/erlang
[root@localhost servers]#  make
[root@localhost servers]#  make install
[root@localhost servers]#  set erlang environment   export PATH=$PATH:/servers/erlang/bin
[root@localhost servers]#  vi /etc/profile

10.若没有添加成功直接修改/etc/profile文件：在该文件中最下面新增export PATH=$PATH:/servers/erlang/bin  然后保存就好
修改完成，退到编辑命令：Ctrl + q
保存修改文件： w
退出到系统命令操作界面：q

[root@localhost servers]# vim /etc/profile

11. source profile使环境变量生效

[root@localhost servers]#  source profile

12.测Erlang环境是否安装成功，输入erl命令，erl语言的退出命令是halt().

rabbitmq安装

同样我这里把rabbitmq安装到我的/servers文件夹下

1.下载rabbitmq-server-3.6.9.tar.xz

[root@localhost servers]# wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.9/rabbitmq-server-generic-unix-3.6.9.tar.xz

2.将rabbitmq包进行解压到 /servers/rabbitmqServer中：

[root@localhost servers]# mkdir rabbitmqServer
[root@localhost servers]# tar xvfz rabbitmq-server-generic-unix-3.6.9.tar.xz /servers/rabbitmqServer

3.配置环境变量：修改/etc/profile
export PATH=/servers/rabbitmqServer/sbin:$PATH
执行source /etc/profile使得PATH路径更新，rabbitMQ安装成功。

4. 启用MQ管理方式（执行下面命令后MQ即已经启动）：
rabbitmq-plugins enable rabbitmq_management   #启动后台管理
rabbitmq-server -detached   #后台运行rabbitmq

5.server默认访问端口15672 监听端口5672

6.用户，权限，角色管理参考上面windows版本的

三、 简单实例

依赖引入

工程采用eclipse/idea + maven，maven依赖

<!-- rabbitmq相关依赖 -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.0.4</version>
</dependency>
<!-- 序列化相关依赖 -->
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.6</version>
</dependency>

因为后续例子里面有用到序列化的，因此加上序列化工具包相关依赖

代码示例1（简单实现）

新建发送者Send.java，代码如下：

package com.luo.rabbit.test.one;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Send {
//队列名称
private final static String QUEUE_NAME = "queue";
public static void main(String[] argv) throws java.io.IOException
{
/**
* 创建连接连接到MabbitMQ
*/
ConnectionFactory factory = new ConnectionFactory();
//设置MabbitMQ所在主机ip或者主机名
factory.setHost("127.0.0.1");
//创建一个连接
Connection connection = factory.newConnection();
//创建一个频道
Channel channel = connection.createChannel();
//指定一个队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//发送的消息
String message = "hello world!";
//往队列中发出一条消息
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("Sent '" + message + "'");
//关闭频道和连接
channel.close();
connection.close();
}
}

新建接收者Recv.java，代码如下：

package com.luo.rabbit.test.one;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class Recv  {
//队列名称
private final static String QUEUE_NAME = "queue";
public static void main(String[] argv) throws java.io.IOException,
java.lang.InterruptedException
{
//打开连接和创建频道，与发送端一样
ConnectionFactory factory = new ConnectionFactory();
//设置MabbitMQ所在主机ip或者主机名
factory.setHost("127.0.0.1");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println("Waiting for messages. To exit press CTRL+C");
//创建队列消费者
QueueingConsumer consumer = new QueueingConsumer(channel);
//指定消费队列
channel.basicConsume(QUEUE_NAME, true, consumer);
while (true)
{
//nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println("Received '" + message + "'");
}

}
}

分别运行这两个类，先后顺序没有关系，先运行发送者再运行接收者，效果如下：

代码示例2（简单实现）

新建一个连接类：

因为发送者和接收者的连接代码都是一样的，之后让二者继承这个连接类即可。连接类代码

BaseConnector.java：
package com.luo.rabbit.test.two;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class BaseConnector {
protected Channel channel;
protected Connection connection;
protected String queueName;
public BaseConnector(String queueName) throws IOException{
this.queueName = queueName;
//打开连接和创建频道
ConnectionFactory factory = new ConnectionFactory();
//设置MabbitMQ所在主机ip或者主机名  127.0.0.1即localhost
factory.setHost("127.0.0.1");
//创建连接
connection = factory.newConnection();
//创建频道
channel = connection.createChannel();
//声明创建队列
channel.queueDeclare(queueName, false, false, false, null);
}
}

发送者Sender.java：

package com.luo.rabbit.test.two;

import java.io.IOException;
import java.io.Serializable;
import org.apache.commons.lang.SerializationUtils;

public class Sender extends BaseConnector {
public Sender(String queueName) throws IOException {
super(queueName);
}

public void sendMessage(Serializable object) throws IOException {
channel.basicPublish("",queueName, null, SerializationUtils.serialize(object));
}
}

新建一个对象

前面讲过，我们想发送一个对象给接受者，因此，我们先新建一个对象，因为发送过程需要序列化，因此这里需要实现java.io.Serializable接口：

package com.luo.rabbit.test.two;
import java.io.Serializable;
public class MessageInfo implements Serializable {
private static final long serialVersionUID = 1L;
//渠道
private String channel;
//来源
private String content;
public String getChannel() {
return channel;
}
public void setChannel(String channel) {
this.channel = channel;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
}

序列化就是将一个对象的状态（各个属性量）保存起来，然后在适当的时候再获得。序列化分为两大部分：序列化和反序列化。序列化是这个过程的第一部分，将数据分解成字节流，以便存储在文件中或在网络上传输。反序列化就是打开字节流并重构对象。对象序列化不仅要将基本数据类型转换成字节表示，有时还要恢复数据。恢复数据要求有恢复数据的对象实例。

接收者代码Receiver.java：

package com.luo.rabbit.test.two;
import java.io.IOException;
import org.apache.commons.lang.SerializationUtils;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.ShutdownSignalException;
public class Receiver extends BaseConnector implements Runnable, Consumer {
public Receiver(String queueName) throws IOException {
super(queueName);
}
//实现Runnable的run方法
public void run() {
try {
channel.basicConsume(queueName, true,this);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 下面这些方法都是实现Consumer接口的
**/
//当消费者注册完成自动调用
public void handleConsumeOk(String consumerTag) {
System.out.println("Consumer "+consumerTag +" registered");
}
//当消费者接收到消息会自动调用
public void handleDelivery(String consumerTag, Envelope env,
BasicProperties props, byte[] body) throws IOException {
MessageInfo messageInfo = (MessageInfo)SerializationUtils.deserialize(body);
System.out.println("Message ( "
+ "channel : " + messageInfo.getChannel()
+ " , content : " + messageInfo.getContent()
+ " ) received.");
}
//下面这些方法可以暂时不用理会
public void handleCancelOk(String consumerTag) {
}
public void handleCancel(String consumerTag) throws IOException {
}
public void handleShutdownSignal(String consumerTag,
ShutdownSignalException sig) {
}
public void handleRecoverOk(String consumerTag) {
}
}

这里，接收者实现了，Runnable接口和com.rabbitmq.client.Consumer接口。

实现Runnable接口的目的是为了实现多线程，java实现多线程的方式有两种：一种是继承Thread类，一种是实现Runnable接口。详情请看这篇文章：http://developer.51cto.com/art/201203/321042.htm。

//指定消费队列

channel.basicConsume(QUEUE_NAME, true, consumer);

最后一个参数是需要传递com.rabbitmq.client.Consumer参数的，实现了Consumer接口之后我们只需要传递this就好了。另外，Consumer有很多方法，上面代码除了构造方法和run方法（run是实现Runnable接口的），其他都是实现Consumer接口的，这些方法的具体含义，大家可以直接看com.rabbitmq.client.Consumer源码。

测试类 Test.java：

package com.luo.rabbit.test.two;
public class Test {
public static void main(String[] args) throws Exception{
Receiver receiver = new Receiver("testQueue");
Thread receiverThread = new Thread(receiver);
receiverThread.start();
Sender sender = new Sender("testQueue");
for (int i = 0; i < 5; i++) {
MessageInfo messageInfo = new MessageInfo();
messageInfo.setChannel("test");
messageInfo.setContent("msg" + i);
sender.sendMessage(messageInfo);
}
}
}

运行效果：

四、spring整合RabbitMQ

依赖引入

工程采用eclipse/idea + maven，maven依赖

pom.xml文件

!-- rabbitmq相关依赖 -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.5.1</version>
</dependency>
<!-- 序列化相关依赖 -->
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.6</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.5.5.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>4.2.4.RELEASE</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>

rabbmitmq配置文件 rabbitmq-config.properties

mq.host=127.0.0.1
mq.username=admin
mq.password=111111
mq.port=5672
mq.vhost=testvhost
mq.queue=testqueue

Spring配置 applicationContext.xml

<!--消息队列集成 -->
<import resource="application-mq.xml"/>

RabbitMQ初始化 application-mq.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:util="http://www.springframework.org/schema/util"
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.5.xsd http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd"> <!-- 属性文件 -->
<util:properties id="appConfig" location="classpath:rabbitmq-config.properties"/>

<!-- 连接配置 connection-cache-size="10"-->
<rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}"
password="${mq.password}" port="${mq.port}" channel-cache-size="10" connection-timeout="200"  />
<rabbit:admin connection-factory="connectionFactory"/>

<!-- 消息队列客户端 -->
<rabbit:template id="amqpTemplate" exchange="${mq.queue}_exchange" connection-factory="connectionFactory" reply-timeout="5000" retry-template="retryTemplate" />

<!-- queue 队列声明 -->
<!--durable 是否持久化
exclusive 仅创建者可以使用的私有队列，断开后自动删除
auto-delete 当所有消费端连接断开后，是否自动删除队列 -->
<rabbit:queue id="test_queue" name="${mq.queue}_testQueue" durable="true" auto-delete="false" exclusive="false" />

<!-- 交换机定义 -->
<!-- 交换机：一个交换机可以绑定多个队列，一个队列也可以绑定到多个交换机上。如果没有队列绑定到交换机上，则发送到该交换机上的信息则会丢失。
**交换机的四种模式：
direct：转发消息到 routigKey 指定的队列。
topic：按规则转发消息（最灵活）。
headers：这个用得少
fanout：转发消息到所有绑定队列
-->
<rabbit:topic-exchange name="${mq.queue}_exchange" durable="true" auto-delete="false">
<rabbit:bindings>
<!-- 设置消息Queue匹配的pattern (direct模式为key) -->
<rabbit:binding queue="test_queue" pattern="${mq.queue}_pattern"/>
</rabbit:bindings>
</rabbit:topic-exchange>

<!--消息重试-->
<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate">
<property name="backOffPolicy">
<bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy">
<property name="initialInterval" value="100" /><!--初始间隔时间-->
<property name="multiplier" value="3.0"/><!--递增倍数-->
<property name="maxInterval" value="200" /><!--最大间隔-->
</bean>
</property>
<property name="retryPolicy">
<bean class="org.springframework.retry.policy.SimpleRetryPolicy">
<property name="maxAttempts" value="5"/>
</bean>
</property>
</bean>
<!-- 拒绝请求消息，并回复该请求者的请求被服务端拒绝-->
<bean id="messageRecoverer" class="cn.com.gzqinghui.sysmgr.common.rabbitMQ.MessageConverter">
<property name="replyToTemplate" ref="amqpTemplate"/>
</bean>

<!-- 配置监听 消费者 -->
<bean name="rabbitmqConsumer" class="cn.com.gzqinghui.sysmgr.common.rabbitMQ.Consumer"></bean>

<!-- 使用spring进行配置 监听  通过设置Messagelistener 的containerconcurrency属性来设置消费者数量
acknowledge:消息确认机制  auto：自动确认  manual：手动确认（一定要在消息处理完手动处理消息，
否则消息对列里面的数据会越来越多 ，逻辑也会重复处理）
concurrency：通道数量，根据对数据量的预估去设置消费者数量
prefetch：每个通道一次获取请求数量-->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual"  concurrency= "15" prefetch="3"  >
<!--queues 监听队列，多个用逗号分隔ref 监听器 -->
<rabbit:listener queues="test_queue" ref="rabbitmqConsumer"/>
</rabbit:listener-container>

</beans>

----------

Demo

1、创建生产者 Producer.java

package cn.com.gzqinghui.sysmgr.common.rabbitMQ;

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

/**
* ClassName:Producer
* Date:     2017年11月1日
* @author   zhangk
* @since    JDK 1.8
*/
@Service
public class Producer {
@Autowired
private AmqpTemplate amqpTemplate;

/**
* 发送消息到队列中
* @param exchange_key 交换机名称，默认是“/” 交换机新增需要去Erlang 中添加
* @param queue_key 队列名称，队列标识
* @param object 消息内容
*/
public void sendQueue(String exchange_key, String queue_key, Object object) {
// convertAndSend 将Java对象转换为消息发送至匹配key的交换机中Exchange
amqpTemplate.convertAndSend(exchange_key, queue_key, object);
}
}

convertAndSend：将Java对象转换为消息发送到匹配Key的交换机中Exchange，由于配置了JSON转换，这里是将Java对象转换成JSON字符串的形式。原文：Convert a Java object to an Amqp Message and send it to a default exchange with a specific routing key.

2、异步接收消息Consumer.java

监听器配置见 application-mq.xml

package cn.com.gzqinghui.sysmgr.common.rabbitMQ; /**
* Project Name:qyk_testSpring
* File Name:Consumer.java
* Package Name:org.qiyongkang.spring.rabbitmq
* Date:2017年3月6日上午10:22:58
* Copyright (c) 2017, Thinkive(http://www.thinkive.com/) All Rights Reserved.
*
*/
import cn.com.gzqinghui.sysmgr.common.rabbitMQ.util.SerializeUtil;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import java.io.IOException;
import java.util.Map;
/**
* ClassName:Producer
* Date:     2017年11月1日
* @author   zhangk
* @since    JDK 1.8
*/
public class Consumer implements ChannelAwareMessageListener {
/**
*
* @param message 消息载体
* @param channel 当前链接通道
* @throws IOException
*/
@SuppressWarnings("unchecked")
@Override
public void onMessage(Message message, Channel channel) throws IOException {
Map<String, Object> content = (Map<String, Object>) SerializeUtil.toObject(message.getBody());
//业务处理
System.out.println("消费内容：" + content+"-------------------------------------------------------------");

//消息确认机制为手动确认的时候需要以下方法手动处理消息
//消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//ack返回false，并重新回到队列
//channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
//拒绝消息      //channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
}

}

3、序列化工具 SerializeUtil.java

package cn.com.gzqinghui.sysmgr.common.rabbitMQ.util;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;

/**
* ClassName:SerializeUtil <br/>
* Function: TODO ADD FUNCTION. <br/>
* Reason: TODO ADD REASON. <br/>
* Date: 2017年11月1日 下午3:06:09 <br/>
*
* @author zhnagk
* @version
* @since JDK 1.8
* @see
*/
public class SerializeUtil {
/**
* 对象转数组
*
* @param obj
* @return
*/
public static byte[] toByteArray(Object obj) {
byte[] bytes = null;
ByteArrayOutputStream bos = new ByteArrayOutputStream();
try {
ObjectOutputStream oos = new ObjectOutputStream(bos);
oos.writeObject(obj);
oos.flush();
bytes = bos.toByteArray();
oos.close();
bos.close();
} catch (IOException ex) {
ex.printStackTrace();
}
return bytes;
}

/**
* 数组转对象
*
* @param bytes
* @return
*/
public static Object toObject(byte[] bytes) {
Object obj = null;
try {
ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
ObjectInputStream ois = new ObjectInputStream(bis);
obj = ois.readObject();
ois.close();
bis.close();
} catch (IOException ex) {
ex.printStackTrace();
} catch (ClassNotFoundException ex) {
ex.printStackTrace();
}
return obj;
}
}

4、JUnit测试

package cn.com.gzqinghui.sysmgr.common.rabbitMQ;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

/**
* ClassName:ProducerTest <br/>
* Function: TODO ADD FUNCTION. <br/>
* Reason: TODO ADD REASON. <br/>
* Date:     2017年11月1日 上午10:47:36 <br/>
* @author   zhangk
* @version
* @since    JDK 1.8
* @see
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {"classpath:applicationContext.xml"})
public class ProducerTest extends AbstractJUnit4SpringContextTests {
@Autowired
private Producer producer;
final String queueId = "testqueue";
@Before
public void init(){
BeanFactory factory = new ClassPathXmlApplicationContext(
"classpath:applicationContext.xml");
producer = factory.getBean(Producer.class);
}

@Test
public void testSendQueue() throws InterruptedException {

Thread.sleep(100000);
try {
/*for (int i = 0; i < 100000; i++) {
Map<String, Object> map = new HashMap<String, Object>();
map.put("name", "testQueue"+i);
//指定转换器以及路由key，就会被绑定相应路由key的消费者进行消费处理
producer.sendQueue(queueId + "_exchange", queueId + "_pattern", map);
System.out.println("生产者内容：" + "testQueue"+i+"-------------------------------------------------------------");
}*/
System.out.println("------------------------------------生产结束-------------------------------------------------------------");

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

}

注：

服务启动请注意tomcat日志问题

测试期间，测试10万条数据，结果全部打印到tomcat日志中，瞬间爆炸，catalina.out文件直接飙升之3.56G，而且就算不在测试，监听器运行rabbitmq日志一样不停输出，catalina.out 文件基本上是以每分钟1MB的速度增加

如果发现类似问题可以参考：http://blog.csdn.net/qq_22673717/article/details/78488038