来源
RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。
支持消息的持久化、事务、拥塞控制、负载均衡等特性，使得RabbitMQ拥有更加广泛的应用场景。
RabbitMQ跟Erlang和AMQP有关。下面简单介绍一下Erlang和AMQP。
Erlang是一门动态类型的函数式编程语言，它也是一门解释型语言，由Erlang虚拟机解释执行。从语言模型上说，Erlang是基于Actor模型的实现。在Actor模型里面，万物皆Actor，每个Actor都封装着内部状态，Actor相互之间只能通过消息传递这一种方式来进行通信。对应到Erlang里，每个Actor对应着一个Erlang进程，进程之间通过消息传递进行通信。相比共享内存，进程间通过消息传递来通信带来的直接好处就是消除了直接的锁开销(不考虑Erlang虚拟机底层实现中的锁应用)。
AMQP(Advanced Message Queue Protocol)定义了一种消息系统规范。这个规范描述了在一个分布式的系统中各个子系统如何通过消息交互。而RabbitMQ则是AMQP的一种基于erlang的实现。AMQP将分布式系统中各个子系统隔离开来，子系统之间不再有依赖。子系统仅依赖于消息。子系统不关心消息的发送者，也不关心消息的接受者。
安装使用
1.rabbitmq安装
这里使用docker的方式
2.rabbitmq启动
docker run -d --name rabbitmq -p 5672:5672 rabbitmq
查看运行的实例
docker ps -a
开启web端口访问管理控制页面
上面的docker启动rabbitmq的时候并没有释放rabbitmq的管理控制台的端口，下面进行重新操作
为了方便，这里我们直接将原先启动的容器stop掉，然后删掉，重新运行释放端口,rabbitmq的管理控制台端口为15672
docker stop 2d5e5be34be7
2d5e5be34be7
docker rm 2d5e5be34be7
2d5e5be34be7
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq
8c015f9b6beb4e14e7726a7367c0e9571ae4987dca8bfb64343fd8f3eeaade03
启动容器后，执行docker命令，找到container的id
zw@zwtk:~$ docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq
7ecff31eb8155a6d9913362c1d745b40e962de92d5bc850113b68d6c777f0906
zw@zwtk:~$ docker ps -a
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
7ecff31eb815 rabbitmq "docker-entrypoint..." 7 seconds ago Up 7 seconds 4369/tcp, 0.0.0.0:5672->5672/tcp, 5671/tcp, 25672/tcp, 0.0.0.0:15672->15672/tcp rabbitmq
找到id为7ecff31eb815 执行下面命令进入docker的命令控制台下
zw@zwtk:~$ docker exec -it 7ecff31eb815 /bin/bash
root@7ecff31eb815:/#
然后在docker的控制台下执行下面命令来启动rabbitmq的管理控制台的插件
root@7ecff31eb815:/# rabbitmq-plugins enable rabbitmq_management
The following plugins have been enabled:
amqp_client
cowlib
cowboy
rabbitmq_web_dispatch
rabbitmq_management_agent
rabbitmq_management
Applying plugin configuration to rabbit@7ecff31eb815... started 6 plugins.
root@7ecff31eb815:/#
刷新页面http://localhost:15672/#/会看到rabbitmq的登录页面,通过guest/guest进行登录
编写客户端例子
HelloWorld
这部分我们将用java语言编写两个应用程序：一个发送单个消息的生产者和一个接受消息并打印他们的消费者。我们会忽视java的api的细节，专注于仅仅将例子跑通开始。一个hello world的消息例子。在下图中，P使我们的生产者，C是我们的消费者，中间的箱子是队列-一个rabbitmq为消费者存放消息的缓存。
在pom.xml文件中添加下面依赖，最新的版本可以通过连接查看
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.0.0</version>
</dependency>`
//发送端和接收端代码;
package TRabbitMQ;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.concurrent.TimeoutException;
public class TSender {
private final static String QUEUE_NAME = "hello";
public static void main(String[] argv)
throws java.io.IOException, TimeoutException {
//链接是对socket链接的抽象，并且负责版本协议的协商和认证。
// 这里我们链接了本地服务器的代理，也就是localhost。
// 如果我们想链接一个不同机器的代理，我们仅需要修改这里的名称或者ip地址即可。
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("localhost");
Connection connection = connectionFactory.newConnection();
//我们创建一个channel通道
Channel channel = connection.createChannel();
//为了发送，我们必须声明一个我们要发送信息的目标队列，然后我们往队列中发送消息。
//声明一个队列是等幂的，如果队列不存在会创建一个，消息内容是一个字节数组，所以你可以想随心所欲的编码你的内容。
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
String message = "Hello world!";
channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));
System.out.println(" [x] Sent '"+message+"'");
channel.close();
connection.close();
}
}
//=====================
package TRabbitMQ;
import com.rabbitmq.client.*;
import java.io.IOException;
public class TReceiver {
private static final String QUEUE_NAME="hello";
public static void main(String args[]) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
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");
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body,"UTF-8");
System.out.println("received:'"+message+"'");
}
};
channel.basicConsume(QUEUE_NAME,consumer);
}
}
RabbitMQ基本原理
首先我们肯定知道RabbitMQ就是消息队列的一种实现，那么围绕这个，我们就可以思考一个消息队列到底需要什么，当然是需要队列，那么这个队列就是Queue，那么其他的所有名词都是围绕这个Queue来拓展的。
首先，想要让队列不在本地运行，而在网络中运行，肯定会有连接这个概念，所以就会有Connection，我们发一条消息连接一次，这样很显然是浪费资源的，建立连接的过程也很耗时，所以我们就会做一个东西让他来管理连接，当我用的时候，直接从里边拿出来已经建立好的连接发信息，那么ConnectionFactory应运而生。
接下来，当程序开发时，可能不止用到一个队列，可能有订单的队列、消息的队列、任务的队列等等，那么就需要给不同的queue发信息，那么和每一个队列连接的这个概念，就叫Channel。
再往下来，当我们开发的时候还有时候会用到这样一种功能，就是当我发送一条消息，需要让几个queue都收到，那么怎么解决这个问题呢，难道我要给每一个queue发送一次消息？那岂不是浪费带宽又浪费资源，我们能想到什么办法呢，当然是我们发送给RabbitMQ服务器一次，然后让RabbitMQ服务器自己解析需要给哪个Queue发，那么Exchange就是干这件事的
但是我们给Exchange发消息，他怎么知道给哪个Queue发呢？这里就用到了RoutingKey和BindingKey
BindingKey是Exchange和Queue绑定的规则描述，这个描述用来解析当Exchange接收到消息时，Exchange接收到的消息会带有RoutingKey这个字段，Exchange就是根据这个RoutingKey和当前Exchange所有绑定的BindingKey做匹配，如果满足要求，就往BindingKey所绑定的Queue发送消息，这样我们就解决了我们向RabbitMQ发送一次消息，可以分发到不同的Queue的过程
至此，我们就把所有的名词贯通咯，接下来做个概要描述：
ConnectionFactory：与RabbitMQ服务器连接的管理器
Connection：与RabbitMQ服务器的TCP连接
Channel：与Exchange的连接，一个Connection可以包含多个Channel。之所以需要Channel，是因为TCP连接的建立和释放都是十分昂贵的，为了多路复用。RabbitMQ建议客户端线程之间不要共用Channel，但是建议尽量共用Connection。
Exchange：接受消息生产者的消息，并根据消息的RoutingKey和 Exchange绑定的BindingKey，以及Binding规则将消息路由给服务器中的队列。ExchangeType决定了Exchange路由消息的行为，例如，在RabbitMQ中，ExchangeType有direct、Fanout和Topic三种，不同类型的Exchange路由的行为是不一样的。
Message Queue：消息队列，用于存储还未被消费者消费的消息。
Message: 由Header和Body组成，Header是由生产者添加的各种属性的集合，包括Message是否被持久化、由哪个Message Queue接受、优先级是多少等。而Body是真正需要传输的APP数据。
RoutingKey：由Producer发送Message时指定，指定当前消息被谁接受
BindingKey：由Consumer在Binding Exchange与Message Queue时指定，指定当前Exchange下，什么样的RoutingKey会被下派到当前绑定的Queue中
Binding:联系了Exchange与Message Queue。Exchange在与多个Message Queue发生Binding后会生成一张路由表，路由表中存储着Message Queue所需消息的限制条件即Binding Key。当Exchange收到Message时会解析其Header得到Routing Key，Exchange根据Routing Key与Exchange Type将Message路由到Message Queue。Binding Key由Consumer在Binding Exchange与Message Queue时指定，而Routing Key由Producer发送Message时指定，两者的匹配方式由Exchange Type决定。
Server(broker): 接受客户端连接，实现AMQP消息队列和路由功能的进程。
Virtual Host:其实是一个虚拟概念，类似于权限控制组，可以通过命令分配给用户Virtual Host的权限，默认的guest用户是管理员权限，初始空间有/，一个Virtual Host里面可以有若干个Exchange和Queue，但是权限控制的最小粒度是Virtual Host
如下图：
名词阐述
基本流程
AMQP的三层协议
1.Module Layer，位于协议最高层，主要定义了一些供客户端调用的命令，客户端可以利用这些命令实现自己的业务逻辑，例如，客户端可以通过queue.declare声明一个队列，利用consume命令获取一个队列中的消息。
2.Session Layer，主要负责将客户端的命令发送给服务器，在将服务器端的应答返回给客户端，主要为客户端与服务器之间通信提供可靠性、同步机制和错误处理。
3.Transport Layer，主要传输二进制数据流，提供帧的处理、信道复用、错误检测和数据表示。
使用场景
- 单发送单接收
注1：queueDeclare第一个参数表示队列名称、第二个参数为是否持久化（true表示是，队列将在服务器重启时生存）、第三个参数为是否是独占队列（创建者可以使用的私有队列，断开后自动删除）、第四个参数为当所有消费者客户端长时间连接断开时是否自动删除队列、第五个参数为队列的其他参数
注2：basicPublish第一个参数为交换机名称、第二个参数为队列映射的路由key、第三个参数为消息的其他属性、第四个参数为发送信息的主体
- 单发送多接收
使用场景：一个发送端，多个接收端，如分布式的任务派发。为了保证消息发送的可靠性，不丢失消息，使消息持久化了。同时为了防止接收端在处理消息时down掉，只有在消息处理完成后才发送ack消息。
2、在使用channel.basicConsume接收消息时使autoAck为false，即不自动会发ack，由channel.basicAck()在消息处理完成后发送消息。
3、使用了channel.basicQos(1)保证在接收端一个消息没有处理完时不会接收另一个消息，即接收端发送了ack后才会接收下一个消息。在这种情况下发送端会尝试把消息发送给下一个not busy的接收端。
注意点：
1）It's a common mistake to miss the basicAck. It's an easy error, but the consequences are serious. Messages will be redelivered when your client quits (which may look like random redelivery), but RabbitMQ will eat more and more memory as it won't be able to release any unacked messages.
2）Note on message persistence
Marking messages as persistent doesn't fully guarantee that a message won't be lost. Although it tells RabbitMQ to save the message to disk, there is still a short time window when RabbitMQ has accepted a message and hasn't saved it yet. Also, RabbitMQ doesn't do fsync(2) for every message -- it may be just saved to cache and not really written to the disk. The persistence guarantees aren't strong, but it's more than enough for our simple task queue. If you need a stronger guarantee you can wrap the publishing code in atransaction.
3）Note about queue size
If all the workers are busy, your queue can fill up. You will want to keep an eye on that, and maybe add more workers, or have some other strategy.
4）RabbitMQ allows you to set Time To Live for both messages and queues.
- 场景3：Publish/Subscribe
使用场景：发布、订阅模式，发送端发送广播消息，多个接收端接收。
Producer：
mport com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
public class EmitLog {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String message = getMessage(argv);
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
channel.close();
connection.close();
}
发送端：
发送消息到一个名为“logs”的exchange上，使用“fanout”方式发送，即广播消息，不需要使用queue，发送端不需要关心谁接收。
Consumer：
ublic class ReceiveLogs {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
}
}
}
接收端：
1、声明名为“logs”的exchange的，方式为"fanout"，和发送端一样。
2、channel.queueDeclare().getQueue();该语句得到一个随机名称的Queue，该queue的类型为non-durable、exclusive、auto-delete的，将该queue绑定到上面的exchange上接收消息。
3、注意binding queue的时候，channel.queueBind()的第三个参数Routing key为空，即所有的消息都接收。如果这个值不为空，在exchange type为“fanout”方式下该值被忽略！
- Routing (按路线发送接收)
使用场景：发送端按routing key发送消息，不同的接收端按不同的routing key接收消息。
Producer：
public class EmitLogDirect {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String severity = getSeverity(argv);
String message = getMessage(argv);
channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
channel.close();
connection.close();
}
}
发送端和场景3的区别：
1、exchange的type为direct
2、发送消息的时候加入了routing key
Consumer：
public class ReceiveLogsDirect {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String queueName = channel.queueDeclare().getQueue();
if (argv.length < 1){
System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
System.exit(1);
}
for(String severity : argv){
channel.queueBind(queueName, EXCHANGE_NAME, severity);
}
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");
}
}
}
接收端和场景3的区别：
在绑定queue和exchange的时候使用了routing key，即从该exchange上只接收routing key指定的消息。
- 场景5：Topics (按topic发送接收)
使用场景：发送端不只按固定的routing key发送消息，而是按字符串“匹配”发送，接收端同样如此。
public class EmitLogTopic {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection();
channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String routingKey = getRouting(argv);
String message = getMessage(argv);
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (connection != null) {
try {
connection.close();
}
catch (Exception ignore) {}
}
}
}
发送端和场景4的区别：
1、exchange的type为topic
2、发送消息的routing key不是固定的单词，而是匹配字符串，如".lu.#"，匹配一个单词，#匹配0个或多个单词。
Consumer：
public class ReceiveLogsTopic {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection();
channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String queueName = channel.queueDeclare().getQueue();
if (argv.length < 1){
System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
System.exit(1);
}
for(String bindingKey : argv){
channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
}
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");
}
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (connection != null) {
try {
connection.close();
}
catch (Exception ignore) {}
}
}
}
}
接收端和场景4的区别：
1、exchange的type为topic
2、接收消息的routing key不是固定的单词，而是匹配字符串。
Exchange类型
RabbitMQ中，所有生产者提交的消息都由Exchange来接受，然后Exchange按照特定的策略转发到Queue进行存储
RabbitMQ提供了四种Exchange：fanout,direct,topic,header
header模式在实际使用中较少，本文只对前三种模式进行比较。
性能排序：fanout > direct >> topic。比例大约为11：10：6
- Direct Exchange
- 一般情况可以使用rabbitMQ自带的Exchange：”"(该Exchange的名字为空字符串，下文称其为default Exchange)。
- 这种模式下不需要将Exchange进行任何绑定(binding)操作
- 消息传递时需要一个“RouteKey”，可以简单的理解为要发送到的队列名字。
- 如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。
Channel channel = connection.createChannel();
channel.exchangeDeclare("exchangeName", "direct"); //direct fanout topic
channel.queueDeclare("queueName");
channel.queueBind("queueName", "exchangeName", "routingKey");
byte[] messageBodyBytes = "hello world".getBytes();
//需要绑定路由键
channel.basicPublish("exchangeName", "routingKey", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
- Fanout Exchange
Fanout Exchange – 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。
任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上。
1.可以理解为路由表的模式
2.这种模式不需要RouteKey
3.这种模式需要提前将Exchange与Queue进行绑定，一个Exchange可以绑定多个Queue，一个Queue可以同多个Exchange进行绑定。
4.如果接受到消息的Exchange没有与任何Queue绑定，则消息会被抛弃。
Channel channel = connection.createChannel();
channel.exchangeDeclare("exchangeName", "fanout"); //direct fanout topic
channel.queueDeclare("queueName");
channel.queueBind("queueName", "exchangeName", "routingKey");
channel.queueDeclare("queueName1");
channel.queueBind("queueName1", "exchangeName", "routingKey1");
byte[] messageBodyBytes = "hello world".getBytes();
//路由键需要设置为空
channel.basicPublish("exchangeName", "", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
- 三.Topic Exchange
Topic Exchange – 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“”匹配不多不少一个词。因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.” 只会匹配到“audit.irs”
任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上
1.这种模式较为复杂，简单来说，就是每个队列都有其关心的主题，所有的消息都带有一个“标题”(RouteKey)，Exchange会将消息转发到所有关注主题能与RouteKey模糊匹配的队列。
2.这种模式需要RouteKey，也许要提前绑定Exchange与Queue。
3.在进行绑定时，要提供一个该队列关心的主题，如“#.log.#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。
4.“#”表示0个或若干个关键字，“”表示一个关键字。如“log.”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.#”能与上述两者匹配。
5.同样，如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息。
分发策略
默认的情况下，rabbitmq将会按顺序分发消息给下一个消费者，平均而言，每个消费者会得到相同数量的消息，这种分发消息的策略成为轮询分发策略。
消息应答
执行一个任务可能花费几秒。如果一个消费者开始了一个很长的任务并且只执行了部分任务就死掉了，你可能会疑问到底发生了什么。用现在的代码来说，每次rabbitmq讲消息传输给消费者后，就立即将消息从内存中删除。这样的话，如果你杀掉了一个worker，你将会丢失掉他正在处理的消息。我们也会丢失掉所有的我们分发到这个特殊worker上但是没有处理的消息。
但是我们不想丢失掉任何的任务。如果一个worker死掉了，我们期望的是这个任务会被分发给另外一个worker。
为了确保一条消息永远不会被丢失，rabbitmq支持消息应答。当消息被消费者接受，处理之后，消费者会给rabbitmq发送应答，rabbitmq在接受到应答之后才会删除掉这条消息。
如果一个消费者死掉了，包括channel通道关闭了，链接关闭了或者tcp链接丢失了，这样就没有应答，rabbitmq会理解为这个消息没有被处理完毕，他将会重新将消息仍回到队列中。如果这时候有其他的消费者在线，那么他将会分发消息到其他的消费者中。这样的话，你就能确保没有消息会丢失，及时worker工作者偶然会死亡掉。
这里没有任何的消息超时，如果消费者死掉了，rabbitmq会重新传输消息。尽管处理一个消息耗费了非常非常长久的时间，那么也是ok的。
消息应答默认的情况下是开着的。可以通过autoAck=true标志来设置将其关闭。
消息持久化
上面讲解了如何确保一个消费者在死掉以后，任务不会丢失。但是我们的任务在rabbitmq服务停止掉后仍然会丢失。
当rabbitmq退出或者挂掉之后，他会遗忘掉队列和队列中的消息，除非你告诉他不要忘记。三件事情能够确保消息不会丢失：我们需要Exchange- 队列-消息都是持久的。
-设置Exchange和MessageQueue的durable属性为true：
只有durable为true的Exchange和durable为ture的Queues才能绑定，否则在绑定时，RabbitMQ都会抛错的。持久化会对RabbitMQ的性能造成比较大的影响，可能会下降10倍不止。
boolean durable = true;
channel.queueDeclare("hello", durable, false, false, null);
尽管这两条语句是对的，但是当前情况下是不生效的。因为我们之前已经声明hello队列是不持久化的。rabbitmq是不允许你用不同的参数重新定义一个已经存在的队列的，并且他会返回一个错误信息给任何想尝试这么做的程序。但是这里有一个快速生效的方法，就是声明一个不同的名字，比如task_queue：
boolean durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);
这个task_queue需要在producer和consumer的代码中都需要修改
- 消息进行持久化，通过设置MessageProperties(他实现了BasicProperties)为PERSISTENT_TEXT_PLAIN。
channel.basicPublish("", "task_queue",
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes());
将消息标记为持久化并不能完全保证消息不会丢失。尽管告诉rabbitmq将消息保存到磁盘中，但是这里仍然会有一个短时间的错失窗口-当rabbitmq已经接收到消息但是还没有对他进行保存。而且rabbitmq不会为每条消息都做fsync(2)-他可能仅仅保存到缓存中并不是真正的写到磁盘中。持久化保证还不强健，但是对于我们简单的任务队列已经足够了。如果你想要一个更强健的保证，你可以使用 publisher confirms
-
消息什么时候需要持久化？
RabbitMQ在两种情况下会将消息写入磁盘：- 消息本身在publish的时候就要求消息写入磁盘；
- 内存紧张，需要将部分内存中的消息转移到磁盘；
消息什么时候会刷到磁盘？
- 数据在写入文件时，首先会写入到这个Buffer，如果Buffer已满，则会将Buffer写入到文件（未必刷到磁盘）；
- 有个固定的刷盘时间：25ms，也就是不管Buffer满不满，每隔25ms，Buffer里的数据及未刷新到磁盘的文件内容必定会刷到磁盘；
- 每次消息写入后，如果没有后续写入请求，则会直接将已写入的消息刷到磁盘：使用Erlang的receive x after 0来实现，只要进程的信箱里没有消息，则产生一个timeout消息，而timeout会触发刷盘操作。
- 消息在磁盘文件中的格式
消息保存于$MNESIA/msg_store_persistent/x.rdq文件中，其中x为数字编号，从1开始，每个文件最大为16M（16777216），超过这个大小会生成新的文件，文件编号加1。消息以以下格式存在于文件中：
<<Size:64, MsgId:16/binary, MsgBody>>
MsgId为RabbitMQ通过rabbit_guid:gen()每一个消息生成的GUID，MsgBody会包含消息对应的exchange，routing_keys，消息的内容，消息对应的协议版本，消息内容格式（二进制还是其它）等等。
-
文件何时删除？
- 阈值合并文件：当所有文件中的垃圾消息（已经被删除的消息）比例大于阈值（GARBAGE_FRACTION = 0.5）时，会触发文件合并操作（至少有三个文件存在的情况下），以提高磁盘利用率。
- 删除文件：publish消息时写入内容，ack消息时删除内容（更新该文件的有用数据大小），当一个文件的有用数据等于0时，删除该文件。
-
消息索引什么时候需要持久化？
索引的持久化与消息的持久化类似，也是在两种情况下需要写入到磁盘中：- 要么本身需要持久化，
- 要么因为内存紧张，需要释放部分内存。
公平调度
你可能已经注意到，分发消息并没有像我们期望的那样工作，比如，有两个worker工人，当有的信息很重量级，有的即时消息会很轻量级，负责重量级的worker将会一直忙碌，而另外一个worker将会基本不工作。好吧，rabbitmq并不知道这个事情，因此就会一直这样分配任务消息。
这样的事情会发生的原因在于rabbitmq在一个消息进入到队列中之后仅仅调度消息。他并没有查看每个消费者没有做出应答的消息的数字，他仅仅盲目的每n个消息轮回给每n个消费者。
为了检测到这个，我们可以使用basicQos方法来设置为1.这会告诉rabbitmq每次不要给消费者超过一个的消息。换句话说，在消费者处理并应答上条消息之前不要为其分发新的任务消息。这样，他就会分配给下一个不在忙碌的worker。
注意queue size
如果所有的workers都在忙碌中，你的队列可能会增长。如果你想注意下他，你可能增加更多的workers或者你使用其他的策略。
无名交换区
channel.basicPublish("", "hello", null, message.getBytes());
第一个参数就是exchange的名字。空字符串标示默认的或者无名的exchange：消息被路由到routingKey所指定的队列中。
现在我们可以发布我们制定的交换区Exchange:
hannel.basicPublish( "logs", "", null, message.getBytes());
队列属性
a) Exclusive：排他队列，如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除。这里需要注意三点：
- 其一，排他队列是基于连接（Connection）可见的，同一连接的不同信道是可以同时访问同一个连接创建的排他队列的。
- 其二，“首次”，如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同。
- 其三，即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除的。这种队列适用于只限于一个客户端发送读取消息的应用场景。
b) Auto-delete:自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。
c) Durable:持久化，这个会在后面作为专门一个章节讨论。
d) 其他选项，例如如果用户仅仅想查询某一个队列是否已存在，如果不存在，不想建立该队列，仍然可以调用queue.declare，只不过需要将参数passive设为true，传给queue.declare，如果该队列已存在，则会返回true；如果不存在，则会返回Error，但是不会创建新的队列。
临时队列
首先，无论我们什么时候链接rabbit，我们需要一个新鲜的的，空的队列。为了达到这个目的，我们可能用一个随机的名字来创建一个队列，更好的情况下是让server为我们选择一个随机的队列。
其次，一旦我们不再链接消费者之后，队列应该立刻被删除。
在java的client中，我们使用无参数的queueDeclare()，我们创建一个不持久化的，单独的，自动删除的队列。
String queueName = channel.queueDeclare().getQueue();
Bindings
我们已经创建了一个fanout的交换区和一个队列，现在我们需要告诉交换区发送消息到我们的队列。在交换区和队列之间的关系叫做绑定
channel.queueBind(queueName, "logs", "");
这样之后，logs交换区会将消息发送到我们的队列
事务
对事务的支持是AMQP协议的一个重要特性。假设当生产者将一个持久化消息发送给服务器时，因为consume命令本身没有任何Response返回，所以即使服务器崩溃，没有持久化该消息，生产者也无法获知该消息已经丢失。如果此时使用事务，即通过txSelect()开启一个事务，然后发送消息给服务器，然后通过txCommit()提交该事务，即可以保证，如果txCommit()提交了，则该消息一定会持久化，如果txCommit()还未提交即服务器崩溃，则该消息不会服务器就收。当然Rabbit MQ也提供了txRollback()命令用于回滚某一个事务。
生产者消息投递可靠性-Confirm机制
使用事务固然可以保证只有提交的事务，才会被服务器执行。但是这样同时也将客户端与消息服务器同步起来，这背离了消息队列解耦的本质。
Rabbit MQ提供了一个更加轻量级的机制来保证生产者可以感知服务器消息是否已被路由到正确的队列中——Confirm。
如果设置channel为confirm状态，则通过该channel发送的消息都会被分配一个唯一的ID，然后一旦该消息被正确的路由到匹配的队列中后，服务器会返回给生产者一个Confirm，该Confirm包含该消息的ID，这样生产者就会知道该消息已被正确分发。对于持久化消息，只有该消息被持久化后，才会返回Confirm。
Confirm机制的最大优点在于异步，生产者在发送消息以后，即可继续执行其他任务。而服务器返回Confirm后，会触发生产者的回调函数，生产者在回调函数中处理Confirm信息。如果消息服务器发生异常，导致该消息丢失，会返回给生产者一个nack，表示消息已经丢失，这样生产者就可以通过重发消息，保证消息不丢失。Confirm机制在性能上要比事务优越很多。但是Confirm机制，无法进行回滚，就是一旦服务器崩溃，生产者无法得到Confirm信息，生产者其实本身也不知道该消息吃否已经被持久化，只有继续重发来保证消息不丢失，但是如果原先已经持久化的消息，并不会被回滚，这样队列中就会存在两条相同的消息，系统需要支持去重。
其他问题
当有多个消费者同时收取消息，且每个消费者在接收消息的同时，还要做其它的事情，且会消耗很长的时间，在此过程中可能会出现一些意外，比如消息接收到一半的时候，一个消费者宕掉了，这时候就要使用消息接收确认机制，可以让其它的消费者再次执行刚才宕掉的消费者没有完成的事情。
另外，在默认情况下，我们创建的消息队列以及存放在队列里面的消息，都是非持久化的，也就是说当RabbitMQ宕掉了或者是重启了，创建的消息队列以及消息都不会保存，为了解决这种情况，保证消息传输的可靠性，我们可以使用RabbitMQ提供的消息队列的持久化机制。
声明MessageQueue
在Rabbit MQ中，无论是生产者发送消息还是消费者接受消息，都首先需要声明一个MessageQueue。这就存在一个问题，是生产者声明还是消费者声明呢？要解决这个问题，首先需要明确：
a)消费者是无法订阅或者获取不存在的MessageQueue中信息。
b)消息被Exchange接受以后，如果没有匹配的Queue，则会被丢弃。
在明白了上述两点以后，就容易理解如果是消费者去声明Queue，就有可能会出现在声明Queue之前，生产者已发送的消息被丢弃的隐患。如果应用能够通过消息重发的机制允许消息丢失，则使用此方案没有任何问题。
但是如果不能接受该方案，这就需要无论是生产者还是消费者，在发送或者接受消息前，都需要去尝试建立消息队列。这里有一点需要明确，如果客户端尝试建立一个已经存在的消息队列，Rabbit MQ不会做任何事情，并返回客户端建立成功的。
声明Exchange
在Rabbit MQ中，声明一个Exchange需要三个参数：ExchangeName，ExchangeType和Durable。
ExchangeName是该Exchange的名字，该属性在创建Binding和生产者通过publish推送消息时需要指定。
ExchangeType，指Exchange的类型，在RabbitMQ中，有三种类型的Exchange：direct ，fanout和topic，不同的Exchange会表现出不同路由行为。
Durable是该Exchange的持久化属性消费者有2种方式获取队列中的消息:
a) 被动方式：通过basic.consume命令，订阅某一个队列中的消息,channel会自动在处理完上一条消息之后，接收下一条消息。（同一个channel消息处理是串行的）。除非关闭channel或者取消订阅，否则客户端将会一直接收队列的消息。
b) 主动方式：通过basic.get命令主动获取队列中的消息，但是绝对不可以通过循环调用basic.get来代替basic.consume，这是因为basic.get RabbitMQ在实际执行的时候，是首先consume某一个队列，然后检索第一条消息，然后再取消订阅。如果是高吞吐率的消费者，最好还是建议使用basic.consume。-
消息投递方式
如果有多个消费者同时订阅同一个队列的话，RabbitMQ是采用循环的方式分发消息的，每一条消息只能被一个订阅者接收。例如，有队列Queue，其中ClientA和ClientB都Consume了该队列，MessageA到达队列后，被分派到ClientA，ClientA回复服务器收到响应，服务器删除MessageA；再有一条消息MessageB抵达队列，服务器根据“循环推送”原则，将消息会发给ClientB，然后收到ClientB的确认后，删除MessageB；等到再下一条消息时，服务器会再将消息发送给ClientA。
可以看出，消费者再接到消息以后，都需要给服务器发送一条确认命令，这个即可以在handleDelivery里显示的调用basic.ack实现，也可以在Consume某个队列的时候，设置autoACK属性为true实现。这个ACK仅仅是通知服务器可以安全的删除该消息，而不是通知生产者，与RPC不同。 如果消费者在接到消息以后还没来得及返回ACK就断开了连接，消息服务器会重传该消息给下一个订阅者，如果没有订阅者就会存储该消息。
既然RabbitMQ提供了ACK某一个消息的命令，当然也提供了Reject某一个消息的命令。当客户端发生错误，调用basic.reject命令拒绝某一个消息时，可以设置一个requeue的属性，如果为true，则消息服务器会重传该消息给下一个订阅者；如果为false，则会直接删除该消息。当然，也可以通过ack，让消息服务器直接删除该消息并且不会重传。
RabbitMQ的transaction、confirm、ack三个概念的解释
RabbitMQ是采用的AMQP协议，AMQP协议定义了”确认”（acknowledgement）,它是从consumer到RabbitMQ的确认，表示一条消息已经被客户端正确处理。RabbitMQ扩展了AMQP协议，定义了从broker到publisher的”确认”，但将其称之为confirm。所以RabbitMQ的确认有2种，叫不同的名字，一个consumer acknowledgement，一个叫publisher confirm。
如果采用标准的 AMQP 协议，则唯一能够保证消息不会丢失的方式是利用事务机制 -- 令 channel 处于 transactional 模式、向其 publish 消息、执行 commit 动作。在这种方式下，事务机制会带来大量的多余开销，并会导致吞吐量下降 250% 。为了补救事务带来的问题，引入了 confirmation 机制（即 Publisher Confirm）。
根据AMQP协议规定，consumer acknowledgemenet是通过basic.ack方法实现的，consumer在收到一条消息后，可以向broker发送basic.ack方法，确认一条消息已经收到。在默认的情况下，consumer acknowledgement模式是开启的，如果不想发送basic.ack，可以在发送basic.consume方法时指定no-ack参数，关闭consumer acknowledgement模式。
Publisher confirm并没有在AMQ协议的基础上添加新的确认方法，而是复用了basic.ack方法。但是publisher confirm模式并不是默认打开的，需要调用confirm.select方法将channel设置成confirm模式。当开启了confirm模式之后，只有当一条消息被所有的mirrors接受之后，publisher才会收到这条消息的confirm，也就是一个basic.ack方法。
RabbitMQ支持事务(transaction)。事务模式也不是默认开启的，需要调用tx.select方法开启事务模式。当开启了事务模式后，只有当一个事务被所有的mirrors接受之后，tx.commit-ok才会返回给客户端。confirm模式和开启事务模式都可以保证”被所有的mirrors接受”，那么，开启confirm模式和开启事务模式有什么区别吗？不同点在于confirm是针对一条消息的，而事务是可以针对多条消息的（当然是针对同一个queue的多条消息）。另外就是，confirm模式只是针对publisher的设置，而事务模式即可以针对publisher，也可以针对consumer。如果针对publisher设置事务模式，则我们可以将多个basic.publish方法放在一个事务中，当所有的publish的消息被所有的mirrors接受后，publisher client会收到tx.commit-ok的方法。如果针对consumer设置事务模式，则我们可以将多个basic.ack方法放在一个事务中，收到tx.commit-ok时表示这些消息都被确认了。
深入持久化
在RabbitMQ中，MessageQueue主要由两部分组成，一个为AMQQueue，主要负责实现AMQP协议的逻辑功能。另外一个是用来存储消息的BackingQueue，本文重点关注的是BackingQueue的设计。
在RabbitMQ中BackingQueue又由5个子队列组成：Q1、Q2、Delta、Q3和Q4。RabbitMQ中的消息一旦进入队列，不是固定不变的，它会随着系统的负载在队列中不断流动，消息的状态不断发生变化。RabbitMQ中的消息一共有5种状态：
a)Alpha：消息的内容和消息索引都保存在内存中；
b)Beta：消息内容保存在磁盘上，消息索引保存在内存中；
c)Gamma：消息内容保存在磁盘上，消息索引在磁盘和内存都有；
d)Delta：消息内容和索引都在磁盘上；
注意：对于持久化的消息，消息内容和消息索引都必须先保存到磁盘上，才会处于上述状态中的一种，而Gamma状态的消息只有持久化的消息才会有该状态。
BackingQueue中的5个子队列中的消息状态，Q1和Q4对应的是Alpha状态，Q2和Q3是Beta状态，Delta对应的是Delta状态。上述就是RabbitMQ的多层队列结构的设计，我们可以看出从Q1到Q4，基本经历的是由RAM到DISK，再到RAM的设计。这样的设计的好处就是当队列负载很高的情况下，能够通过将一部分消息由磁盘保存来节省内存空间，当负载降低的时候，这部分消息又渐渐回到内存，被消费者获取，使得整个队列有很好的弹性。下面我们就来看一下，整个消息队列的工作流程。
引起消息流动主要有两方面的因素：其一是消费者获取消息；其二是由于内存不足，引起消息的换出到磁盘上（Q1-.>Q2、Q2->Delta、Q3->Delta、Q4->Q3）。RabbitMQ在系统运行时会根据消息传输的速度计算一个当前内存中能够保存的最大消息数量（Target_RAM_Count），当内存中的消息数量大于该值时，就会引起消息的流动。进入队列的消息，一般会按着Q1->Q2->Delta->Q3->Q4的顺序进行流动，但是并不是每条消息都一定会经历所有的状态，这个取决于当时系统的负载状况。
当消费者获取消息时，首先会从Q4队列中获取消息，如果Q4获取成功，则返回，如果Q4为空，则尝试从Q3获取消息；首先，系统会判断Q3队列是否为空，如果为空，则直接返回队列为空，即此时队列中无消息（后续会论证）。如果不为空，则取出Q3的消息，然后判断此时Q3和Delta队列的长度，如果都为空，则可认为Q2、Delta、Q3和Q4全部为空(后续说明)，此时将Q1中消息直接转移到Q4中，下次直接从Q4中获取消息。如果Q3为空，Delta不空，则将Delta中的消息转移到Q3中；如果Q3非空，则直接下次从Q3中获取消息。在将Delta转移到Q3的过程中，RabbitMQ是按照索引分段读取的，首先读取某一段，直到读到的消息非空为止，然后判断读取的消息个数与Delta中的消息个数是否相等，如果相等，则断定此时Delta中已无消息，则直接将Q2和刚读到的消息一并放入Q3中。如果不相等，则仅将此次读到的消息转移到Q3中。这就是消费者引起的消息流动过程。
下面我们分析一下由于内存不足引起的消息换出。消息换出的条件是内存中保存的消息数量+等待ACK的消息的数量>Target_RAM_Count。当条件触发时，系统首先会判断如果当前进入等待ACK的消息的速度大于进入队列的消息的速度时，会先处理等待ACK的消息。步骤基本上Q1->Q2或者Q3移动，取决于Delta队列是否为空。Q4->Q3移动，Q2和Q3向Delta移动。
最后，我们来分析一下前面遗留的两个问题，一个是为什么Q3队列为空即可认定整个队列为空。试想如果Q3为空，Delta不空，则在Q3取出最后一条消息时，Delta上的消息就会被转移到Q3上，与Q3空矛盾。如果Q2不空，则在Q3取出最后一条消息，如果Delta为空时，会将Q2的消息并入Q3，与Q3为空矛盾。如果Q1不空，则在Q3取出最后一条消息，如果Delta和Q3均为空时，则将Q1的消息转移到Q4中，与Q4为空矛盾。这也解释了另外一个问题，即为什么Q3和Delta为空，Q2就为空。
上述就是整个消息在RabbitMQ队列中流动过程。从上述流程可以看出，消息如果能够被尽早消费掉，就不需要经历持久化的过程，因为这样会加系统的开销。如果消息被消费的速度过慢，RabbitMQ通过换出内存的方式，防止内存溢出。
Ref:
http://m.blog.csdn.net/wanbf123/article/details/78052419
http://backend.blog.163.com/blog/static/202294126201322511327882/
