精通 RabbitMQ 系列 02-阿里云开发者社区

阅读全文，约 15 分钟

这是江帅帅的第019篇原创

一、下载和安装 Erlang

1.1 安装依赖环境

先安装 GCC 等依赖环境

yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel

再安装 ncurses

yum -y install ncurses-devel

1.2 安装 Erlang

官网：https://www.erlang.org/downloads

或者使用 wget 下载，然后解压：

cd /usr/local/src/
wget http://erlang.org/download/otp_src_22.0.tar.gz
 
tar -zxvf otp_src_22.0.tar.gz

开始安装：

cd otp_src_22.0
./configure --prefix=/usr/local/soft
make
make install

安装位置：

whereis erlang

1.3 配置环境变量

修改 /etc/profile 配置文件

ERLANG_HOME=/usr/local/soft
export PATH=$PATH:$ERLANG_HOME/bin
export PATH ERLANG_HOME

执行

source /etc/profile

1.4 检验

erl

二、下载和安装 RabbitMQ

官网：http://www.rabbitmq.com/releases/rabbitmq-server/

下载

wget https://www.rabbitmq.com/releases/rabbitmq-server/v3.6.15/rabbitmq-server-generic-unix-3.6.15.tar.xz

解压

xz -d rabbitmq-server-generic-unix-3.6.15.tar.xz
 
tar -xvf rabbitmq-server-generic-unix-3.6.15.tar

编辑 /etc/profile 修改环境变量

三、基本介绍

3.1 工作模型

Broker 是 RabbitMQ 的服务器，消息队列的两大作用：存储消息和转发给消费者。Producter 和 Consumer 工作时，都必须跟服务器建立一个 TCP 的长链接，叫 Connection。但如果生产者和消费者都直接连接服务器的话，频繁的创建和释放一个 Connection 的话，则极其消耗性能，所以引入一个【消息通道】的虚拟连接，在保持一个 TCP 长连接中，创建和释放 Channel 来优化节省系统性能。

比如，很多使用 Client API 进行编程的接口都是基于 Channel 去完成的，包括 Spring 也针对 Channel 做了方法调用的封装。

在其他的消息中间件中，Producter（生成者）发送消息会直接就到 Queue（队列），但 RabbitMQ 不会，它先经过 Exchange（交换机）。交换机主要实现消息更加灵活的分发，通过它的类型我们就知道它为什么灵活了。

交换机没有自己运行的进程，只是一个地址的列表，消息发送过来，然后根据地址列表送到对应的队列。队列有自己的进程，用来存储消息的。交换机和队列之间，需要定义一个绑定的关系。

每一个业务系统，都需要来使用 RabbitMQ 服务器，然后创建了很多 Exchange 和 Queue，资源管理就有可能很混乱。如果某业务系统，希望单独使用一台硬件服务器，不想跟其他的混在一起，那这样是不是需要额外采购一台机器呢？

其实，我们可以创建多个 Virtual Host（虚拟机）来解决，虚拟机相当于小型的 RabbitMQ 服务器，在不同的虚拟机中就可以添加多个 Exchange 来绑定 Queue 解决问题了。虚拟机可以提高我们的资源利用率，也可以提供资源隔离，类似于变成中命名空间和包的概念。

也可以创建多个用户，对应分配一些不同虚拟机的权限。用户能够绑定虚拟机，也就能绑定到其中的交换机和队列，还有它们的绑定关系。不同的用户，不同的虚拟机，就实现了我们资源的隔离。

3.2 RabbitMQ 的几种交换机

RabbitMQ 中一共有 4 种 Exchange（交换机），分别为：

direct（默认）
topic
fanout
headers（几乎不用，性能差）

1. Direct Exchange 直连交换机

从图中可以看出，很多队列需要从直连交换机中取走消息。那么，这时需要使用 binding key 绑定关键字，将它们进行绑定。在发送消息的时候，也需要指定 routing key 路由关键字，用来对应匹配绑定关键字找到对应的队列。

例如，basicPulish("DIRECT_EXCHANGE", "key1", "cuihua");

2. Topic Exchange 主题交换机

星号 * 代表一个单词，井号 # 代表零个或多个单词。

例如：basicPulish("TOPIC_EXCHANGE", "key1", "dodo.abc"); ---- dodo.*

basicPulish("TOPIC_EXCHANGE", "key1", "dodo.aaa.bbb"); ---- dodo.#

3. Fanout Exchange 广播交换机

队列不需要和交换机指定任何的绑定关键字，发送消息的时候也不需要携带任何路由关键字，当发送一条消息到广播交换机的时候，所有相关的队列都能接收到相同的消息。

3.3 Java API 编程

RabbitMQ 官网 https://www.rabbitmq.com
生产者
消费者
参数详解

案例展示

1. 生产者

public class Producer {
 
    // 交换机类型
    private static final String EXCHANGE_NAME = "excName";
    // 路由 key
    private static final String ROUTING_KEY = "rkey";
    // 队列名
    private static final String QUEUE_NAME = "qName";
    // IP 地址
    private static final String IP_ADDRESS = "192.168.207.133";
    // 端口号
    private static final int PORT = 5672;
 
    public static void main(String[] args) throws Exception {
 
        // 1. 连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(IP_ADDRESS);
        factory.setPort(PORT);
        factory.setUsername("root");
        factory.setPassword("1234");
 
        // 2. 配置信息
        // 连接对象
        Connection conn = factory.newConnection();
        // 信道
        Channel channel = conn.createChannel();
        // 交换器
        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
        // 队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        // 绑定
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
 
        // 3. 发送消息
        String msg = "cuihua is a good girl...";
        channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
 
        // 4. 关闭资源
        channel.close();
        conn.close();
    }
}

2. 消费者

public class Consumer {
 
    // 队列名
    private static final String QUEUE_NAME = "qName";
    // IP 地址
    private static final String IP_ADDRESS = "192.168.207.133";
    // 端口号
    private static final int PORT = 5672;
 
    public static void main(String[] args) throws Exception {
        // 指定地址和端口
        Address[] addresses = new Address[]{
                new Address(IP_ADDRESS, PORT)
        };
 
        // 连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername("root");
        factory.setPassword("1234");
 
        // 连接方式
        // 创建连接方式
        Connection conn = factory.newConnection(addresses);
        // 创建信道
        final Channel channel = conn.createChannel();
        // 设置客户端最多接收未被 ack 的消息个数
        channel.basicQos(64);
 
        com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 
                System.out.println("msg : " + new String(body));
 
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e){
                    e.printStackTrace();
                }
 
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
 
        channel.basicConsume(QUEUE_NAME, consumer);
 
        TimeUnit.SECONDS.sleep(5);
        channel.close();
        conn.close();
    }
}

3.4 进阶知识

怎么自动删除没人消费的消息？
无法路由的消息，去了哪里？
可以让消息优先得到消费吗？
如何实现延迟发送消息？
MQ 怎么实现 RPC？
RabbitMQ 流量控制怎么做？设置队列大小有用吗？

1. 怎么自动删除没人消费的消息？

TTL：Time To Live

a. 设置一个队列的消息

b. 指定消息的 TTL

死信队列 Dead Letter Queue

2. 无法路由的消息，去了哪里？

死信交换机 DLX：Dead Letter Exchange

哪些情况消息会变成死信？

a. 消息过期 
b. Reject Nack 并且 requeue == false
c. 队列达到最大长度 —— 先入队的消息会被删除

3. 可以让消息优先得到消费吗？

设置 x-max-priority 属性，指定优先数字。

4. 如果实现延迟发送消息？

1）延迟队列的插件。

2）实现的思路：TTL + DLX

5. MQ 怎么实现 RPC？

为什么要实现 RPC？我们原来在一个项目中，或者一个代码工程中，需要去调用一个其他业务模块的方法是很简单的。但现在是在不同的网络节点上、不同语言实现的系统上，还要实现与方法调用一样的方式去工作。

6. RabbitMQ 流量控制怎么做？设置队列大小有用吗？

设置 x-max-length 是没有太大作用的，是因为正常情况下，消费者是不会停止的，其中最大长度是堆积消息的长度，而不是说把消费者停掉，如果消费者区消息的速度比发消息的速度还快的话，那永远达不到最大长度。

1）服务端

没有特别明确的方式能设置只接收指定量的消息。

我们的消息要么存放在内存中，要么存放在磁盘中，或者同时存放在内存和磁盘中。流量的控制，需要从硬件资源的角度去控制，有一个 Flow Control 策略，从内存使用率控制来考虑。比如，当前机器内存是 2G 的话，当内存的可用空间达到 0.4 限度的时候，就不再接收消息，所有消息生产者就会被阻塞。

或者是，我们磁盘的可用空间用到只剩 1G 的时候，就不再接收消息了。

2）消费端

在官网的 RabbitMQ Tutorials 第二个 Work queues 中可以看到，两个消费者去轮询接收消息。但这种平均分发的方式，不一定会公平，平均 ≠ 不公平。是因为，第一个消费者收到是奇数消息，第二个消费者能够收到是偶数消息。那如果我们每个奇数消息都需要很长时间来处理时，第一个消费者第一个消息都还没处理完的时候，第二个消费者有可能已经处理了好多个消息。

可以针对我们每个消息通道或者消费者，去设置 prefetch count 预取数量，对应的方法是 channel.basicQos()。

如果设置 channel.basicQos(1)，也就是说，当前还有 1 条消息在处理，还没有给队列发送应答的时候，这个队列就不会再发送消息了。

精通 RabbitMQ 系列 02