RabbitMQ学习总结

共计 35827 个字符，预计需要花费 90 分钟才能阅读完成。

这段时间主要是学习设计模式，一直想写一系列的设计模式相关博文，但是设计模式理解起来说难也不难，说简单也不简单，但要想能深入理解并在项目架构中运用自如的话，还是得花点时间好好积累的。扯得有点远了~

今天咱们来聊一聊消息队列（MQ），以 RabbitMQ 为例。

消息队列：在消息的传输过程中保存消息的的容器。

这是一个较为经典的消费 - 生产者模型，说起来比较抽象，打个比方：A 线程需要给 B 线程发送消息（A、B 线程不一定是在同一台机器上的），A 线程先把消息发送到消息队列服务器上，然后 B 线程去读取或是订阅消息服务器上消息队列中的消息，线程 A 和 B 之间并没有进行直接通信。MQ 服务器在中间起到中继的作用。

比较适合异步传输，这里解释一下什么是异步和同步。

异步：发送方不关心消息有没有发送成功，只发送消息，不去获取消息是否发送成功。

同步：发送方关心消息是否发送成功，发送消息后，会等待接收方返回状态码，根据状态码来判断是否发送成功，然后执行相对于的动作。

下边以 Http 中的同步和异步为例：

如：普通的 B / S 架构客户端和服务器端之间的通信就是同步的，即提交请求 —> 等待服务器处理完毕返回消息 —> 拿到服务器返回的消息，处理完毕。

如：Ajax 技术就是异步的，请求通过事件触发 —> 服务器处理（浏览器不用等待，仍可以做其他的事情）—> 处理完毕。

有人可能会好奇说应用场景怎么说到了同步和异步，那说明你还不是很理解技术和应用场景之间的紧密联系。

在消息队列中有很多类似的产品，以后我会独立发一篇博文来经行说明。我们这里以 RabbitMQ 为例。

RabbitMQ 是 AMQP（高级消息队列协议）的一个标准实现，关于它的快速入门，可以在这里查看：http://www.rabbitmq.com/getstarted.html。

每个开源项目都有自己的设计方法以及模块角色，RabbitMQ 也不例外。结构图如下：

RabbitMQ 学习总结

Broker：即消息队列服务器实体

Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。

Queue：消息队列载体，每个消息都会被投入到一个或多个队列。

Binding：绑定，它的作用就是把 exchange 和 queue 按照路由规则绑定起来。

Routing Key：路由关键字，exchange 根据这个关键字进行消息投递。

vhost：虚拟主机，一个 broker 里可以开设多个 vhost，用作不同用户的权限分离。

producer：消息生产者，就是投递消息的程序。

consumer：消息消费者，就是接受消息的程序。

channel：消息通道，在客户端的每个连接里，可建立多个 channel，每个 channel 代表一个会话任务。

生产者客户端：

客户端连接到 RabbitMQ 服务器上，打开一个消息通道（channel）；
客户端声明一个消息交换机（exchange），并设置相关属性。
客户端声明一个消息队列（queue），并设置相关属性。
客户端使用 routing key 在消息交换机（exchange）和消息队列（queue）中建立好绑定关系。
客户端投递消息都消息交换机（exchange）上
客户端关闭消息通道（channel）以及和服务器的连接。

服务器端：

exchange 接收到消息后，根据消息的 key（这个 key 的产生规则暂时没研究，有知道的小伙伴可以留言告诉我）和以及设置的 binding，进行消息路由，将消息投递到一个或多个消息队列中。

关于 exchange 也有几个类型：

(1). Direct 交换机：完全根据 key 进行投递。例如，绑定时设置了 routing key 为 abc，客户端提交信息提交信息时只有设置了 key 为 abc 的才会投递到队列；

(2).Topic 交换机：在 key 进行模式匹配后进行投递。例如：符号”#”匹配一个或多个字符，符号”*”匹配一串连续的字母字符，例如”abc.#”可以匹配”abc.def.ghi”，而”abc.*”只可以匹配”abc.def”。

(3).Fanout 交换机：它采取广播模式，消息进来时，将会被投递到与改交换机绑定的所有队列中。

消费者客户端：

（暂时未研究，后续更新）

CentOS 5.6 安装 RabbitMQ http://www.linuxidc.com/Linux/2013-02/79508.htm

RabbitMQ 客户端 C ++ 安装详细记录 http://www.linuxidc.com/Linux/2012-02/53521.htm

用 Python 尝试 RabbitMQ http://www.linuxidc.com/Linux/2011-12/50653.htm

RabbitMQ 集群环境生产实例部署 http://www.linuxidc.com/Linux/2012-10/72720.htm

Ubuntu 下 PHP + RabbitMQ 使用 http://www.linuxidc.com/Linux/2010-07/27309.htm

在 CentOS 上安装 RabbitMQ 流程 http://www.linuxidc.com/Linux/2011-12/49610.htm

RabbitMQ 概念及环境搭建 http://www.linuxidc.com/Linux/2014-12/110449.htm

RabbitMQ 入门教程 http://www.linuxidc.com/Linux/2015-02/113983.htm

RabbitMQ 支持数据持久化，也就是把数据写在磁盘上，可以增加数据的安全性。消息队列持久化包括三个部分：

消息交换机（exchange）持久化，在声明时指定 durable 为 1
消息队列（queue）持久化，在声明时指定 durable 为 1
消息持久化，在投递时指定 delivery_mode 为 2（1 是非持久化）

如果消息交换机（exchange）和消息队列（queue）都是持久化的话，那么他们之间的绑定（Binding）也是持久化的。如果消息交换机和消息队列之间一个持久化、一个非持久化，那么就不允许绑定。

更多详情见请继续阅读下一页的精彩内容：http://www.linuxidc.com/Linux/2016-07/133178p2.htm

RabbitMQ 是一个消息代理，事实上，它接收生产者产生的消息，然后将消息传递给消费者。在这个过程中，它可以路由，可以缓冲，或者更具你设定的规则来将消息持久化。RabbitMQ 和消息传输过程中一般会用一些术语：

生产者（Producing ）：

意思无非是指发送消息的那一端，如果一个程序发送消息，那么它就将被称为 生产者，这里用大写的 P 来表示。

RabbitMQ 学习总结

队列（queue ）：

相当于邮箱的名字，它活动在 RabbitMQ 服务器里边。虽然消息流会通过 RabbitMQ 和你的应用程序，但是只会被存储在队列中。队列是不受任何限制的，它可以尽可能多的去存储你需要存储的消息（本质上来说它是个无限缓冲）。可以多个生产者向同一个消息队列发送消息，也可以多个消费者同时从一个消息队列中来接收消息。消息队列可以如下图模型。

RabbitMQ 学习总结

消费者（Consuming ）：

即接收端，消费者主要是等待接收消息的程序，用下图表示:

RabbitMQ 学习总结

注意：在大多数应用场景中，生产者、消费者以及 RabbitMQ 服务是不会同时运行在一台机器上的。

下边将会实现两个 Java 程序，一个只发送生产者一条消息的生产者，一个接收消息、并打印消息的消费者。

在下边的对话中，”P”是我们的生产者，”C”是我们的消费者，中间的是矩形是队列（BabbitMQ 维护的消息缓冲）

RabbitMQ 学习总结

RabbitMQ 遵守 AMQP 协议，AMQP 协议是一个开放、通用的消息协议。关于 AMQP 协议的客户端，有多种语言的实现版本。本文使用的是 RabbitMQ 提供的 Java 客户端。

可在 http://www.rabbitmq.com/java-client.html 这里下载，然后将相对应的 Jar 包拷贝你的工作目录下。RabbitMQ 的 Java 客户端在 Maven 库中也有，groupId 是 com.rabbitmq，artifactId 是 amqp-client。

RabbitMQ 学习总结

我们将消息发送器起名为 Send，消息接收器起名为 Recv。发送器将会连接 RabbitMQ，发送一条消息，然后退出。

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

public class Send {private final static String QUEUE_NAME = "hello";

  public static void main(String[] argv) throws java.io.IOException {/*连接可以抽象为 socket 连接，为我们维护协议版本信息和协议证书等。这里我们连接
    上了本机的消息服务器实体（localhost）。如果我们想连接其它主机上的 RabbitMQ 服务，只需要修改一下主机名或是 IP 就可以了*/
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    /*接下创建 channel（信道），这是绝大多数 API 都能用到的。为了发送消息，你必须要声明一个消息消息队列，然后向该队列里推送消息*/
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    String message = "Hello World!";
    channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
    System.out.println("[x] Sent'" + message + "'");
    
    /*声明一个幂等的队列（只有在该队列不存在时，才会被创建）。消息的上下文是一个
      字节数组，你可以指定它的编码。*/
    channel.close();
    connection.close();}
}

注：若 RabbitMQ 服务磁盘空间不足的话，运行会出错，参数设置为：disk_free_limit，更多的参数配置可在这里找到 http://www.rabbitmq.com/configure.html#config-items

RabbitMQ 会往接收器上推消息，与只发送一条消息的发送端不同，这里我们将监听消息并将消息打印出来。

RabbitMQ 学习总结

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;

public class Recv {private final static String QUEUE_NAME = "hello";

  public static void main(String[] argv) throws java.io.IOException, java.lang.InterruptedException {/*这里怎么打开连接和信道，以及声明用于接收消息的队列，这些步骤与发送端基本上是一样的*/
    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");
    
    /*这里用到了额外的类 QueueingConsumer 来缓存服务器将要推过来的消息。我们通知服务器向接收端推送消息，
   然后服务器将会向客户端异步推送消息，这里提供了一个可以回调的对象来缓存消息，直到我们做好准备来使用
    它，这个类就是 QueueingConsumer*/
    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(QUEUE_NAME, true, consumer);

    while (true) {QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      String message = new String(delivery.getBody());
      System.out.println("[x] Received'" + message + "'");
    }
  }
}

消息接收完后 QueueingConsumer.nextDelivery()将会发生阻塞，暂停运行，直到有其他的消息推过来。

如果你需要检查和验证队列的话，需要使用 rabbitmqctl list_queues。

参考链接：http://www.rabbitmq.com/tutorials/tutorial-one-java.html

今天咱们来聊一聊消息队列（MQ），以 RabbitMQ 为例。

消息队列：在消息的传输过程中保存消息的的容器。

比较适合异步传输，这里解释一下什么是异步和同步。

异步：发送方不关心消息有没有发送成功，只发送消息，不去获取消息是否发送成功。

同步：发送方关心消息是否发送成功，发送消息后，会等待接收方返回状态码，根据状态码来判断是否发送成功，然后执行相对于的动作。

下边以 Http 中的同步和异步为例：

如：普通的 B / S 架构客户端和服务器端之间的通信就是同步的，即提交请求 —> 等待服务器处理完毕返回消息 —> 拿到服务器返回的消息，处理完毕。

如：Ajax 技术就是异步的，请求通过事件触发 —> 服务器处理（浏览器不用等待，仍可以做其他的事情）—> 处理完毕。

有人可能会好奇说应用场景怎么说到了同步和异步，那说明你还不是很理解技术和应用场景之间的紧密联系。

在消息队列中有很多类似的产品，以后我会独立发一篇博文来经行说明。我们这里以 RabbitMQ 为例。

RabbitMQ 是 AMQP（高级消息队列协议）的一个标准实现，关于它的快速入门，可以在这里查看：http://www.rabbitmq.com/getstarted.html。

每个开源项目都有自己的设计方法以及模块角色，RabbitMQ 也不例外。结构图如下：

RabbitMQ 学习总结

Broker：即消息队列服务器实体

Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。

Queue：消息队列载体，每个消息都会被投入到一个或多个队列。

Binding：绑定，它的作用就是把 exchange 和 queue 按照路由规则绑定起来。

Routing Key：路由关键字，exchange 根据这个关键字进行消息投递。

vhost：虚拟主机，一个 broker 里可以开设多个 vhost，用作不同用户的权限分离。

producer：消息生产者，就是投递消息的程序。

consumer：消息消费者，就是接受消息的程序。

channel：消息通道，在客户端的每个连接里，可建立多个 channel，每个 channel 代表一个会话任务。

生产者客户端：

客户端连接到 RabbitMQ 服务器上，打开一个消息通道（channel）；
客户端声明一个消息交换机（exchange），并设置相关属性。
客户端声明一个消息队列（queue），并设置相关属性。
客户端使用 routing key 在消息交换机（exchange）和消息队列（queue）中建立好绑定关系。
客户端投递消息都消息交换机（exchange）上
客户端关闭消息通道（channel）以及和服务器的连接。

服务器端：

关于 exchange 也有几个类型：

(1). Direct 交换机：完全根据 key 进行投递。例如，绑定时设置了 routing key 为 abc，客户端提交信息提交信息时只有设置了 key 为 abc 的才会投递到队列；

(3).Fanout 交换机：它采取广播模式，消息进来时，将会被投递到与改交换机绑定的所有队列中。

消费者客户端：