使用介绍

Attention

出于安全考虑，我们不提供公网 IP 来连接 RabbitMQ，因此如果要在本地连接调试需要先参考文档如何使用网易云 OpenVPN ，根据自己的实例所在可用区使用或自行搭建 VPN 连接或直接通过与实例同可用区或同 VPC 的云服务器使用 RabbitMQ

名词说明

Virtual Host 包含若干个Exchange和Queue，表示一个节点
Exchange 接受客户端发送的消息，并根据Binding将消息路由给服务器中的队列，经常使用的Exchange包括direct, fanout, topic类型
Binding 连接Exchange和Queue，包含路由规则
Queue 消息队列，存储还未被消费的消息
Message Header+Body
Channel 通道，执行AMQP的命令, 一个连接可创建多个通道以节省资源
- RoutingKey 路由键，一个queue绑定一个路由键，每个消息发送时也会指定一个路由键，exchange只会将消息路由到RoutingKey匹配的queue中

使用说明

入门示例

首先根据管理页面提示创建一个消息队列实例。创建成功后，点击进入实例详情页，然后点击页面上的账号与密码，获取对应的RabbitMQ用户名密码。下面以RabbitMQ官方提供的java client sdk为例：

生产者

创建连接

//创建客户端配置
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");     //配置为Rabbitmq地址
factory.setPort(5672);
factory.setUsername(username);     //配置为上面获取的用户名密码
factory.setPassword(password);
//创建一个连接
Connection connection = factory.newConnection();
//打开一个通道
Channel channel = connection.createChannel();

创建Queue，并绑定交换机

//创建一个名为queue1的队列
channel.queueDeclare("queue1", false, false, false, null);

第二个参数durable false设置队列为非持久化，第三个参数exclusive false设置队列为非独占，第四个参数autoDelete设置为false表示队列上无连接时自动删除。队列默认会绑定到RabbitMQ的一个direct类型且名为""的exchange。direct交换器的特点是只会将收到的消息路由到RoutingKey匹配的队列中去，如果无法匹配，默认情况下发送到该exchange的消息消息将被丢弃，如果有多个Queue上均有匹配的RoutingKey，则RabbitMQ会自动选择一个queue发送消息。或者，我们也可以自己声明一个交换机并将队列绑定上去：

//创建交换机
channel.exchangeDeclare("exchangeName", "direct", false, false, null);

第三个和第四个参数设置exchange为非持久化，以及非自动删除。

//绑定队列到刚刚创建的exchange
channel.queueBind("queue1, "exchangeName", "");

发布消息

 channel.basicPublish("exchangeName", "key", null, "this is a message".getBytes("UTF-8"));

第一个参数表示exchange名称，第二个参数为消息的RoutingKey，第三个参数为消息选项(例如持久化选项等)，第四个参数表示消息内容。

消费者

创建连接

//创建客户端配置
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");     //配置为Rabbitmq地址
factory.setPort(5672);
factory.setUsername(username);     //配置为上面获取的用户名密码
factory.setPassword(password);
//创建一个连接
Connection connection = factory.newConnection();
//打开一个通道
Channel channel = connection.createChannel();

消费消息

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(" [x] Received '" + message + "'");
    }
};
//启动消费者，参数2表示使用autoAck自动确认机制
channel.basicConsume("queue1", true, consumer);

这里直接使用了DefaultConsumer类实现并覆盖了handleDelivery方法，当有消息到来的时候，client会回调handleDelivery方法，basicConsume表示启动消费者，并等待消息到来。

通常情况下，消费端也会调用exchangeDeclare方法和queueDeclare方法去声明交换器和队列，以防止生产者还未启动而导致消费者获取不到exchange和queue的相关信息。

保证消息可靠性

Attention

RabbitMQ提供了三种机制来保证消息的可靠性，他们分别为生产者confirm机制、消费者ack机制、persist持久化机制，注意开启这3个选项会对性能有较大的影响，所以应当根据实际情况选择。下面介绍如何使用这3种机制。

生产者 confirm 机制

一旦开启了生产者confirm机制，rabbitmq会在处理完消息后，在当前 channel 上通过发送 basic.ack 的方式对其进行 confirm 。

开启confirm

//开启confirm：发送select帧
channel.confirmSelect();

确认消息被发布

其中收到消息的delivery-tag 域的值标识了被 confirm 消息的序列号。RabbitMQ也可以通过设置 basic.ack 中的 multiple 域来表明到指定序列号为止的所有消息都已被正确的处理了。调用 channel.waitForConfirms()方法等待前面发送的消息被确认，返回true表示前面的消息都被正确的发布了。这里最好采用批量确认(发送一批消息后调用一次该方法)以获得较好的性能。

消费者ACK机制

入门示例中打开了autoAck自动确认机制，表示一旦消费者接收到了该消息，不管消费者是否处理完，立刻向RabbitMQ返回一个确认消息。有时候我们还需要保证消费者已经处理完了消息才能给RabbitMQ发送这条确认消息，通过以下方法设置：

//第二个参数autoAck参数为false，表示启动消费者时，关闭自动确认机制
channel.basicConsume("queue1", false, consumer);

我们可以通过调用channel.basicAck方法来确认一个消息已被正常的处理，或者调用我们可以通过调用channel.basicNack方法拒绝一个或一批消息。

persist持久化机制

前面的例子exchange, queue, message都是非持久化的，如果消息未被确认，一旦重启RabbitMQ，所有的信息都将丢失。可以通过以下字段进行持久化设置：

设置queue为持久化类型

// 设置第二个参数durable为true指定了queue为持久化类型，重启时该queue信息不会丢失
channel.queueDeclare("queue1", true, false, false, null);

设置exchange为持久化类型

// 设置第三个参数durable为true指定了exchange为持久化类型，重启时该exchange信息不会丢失
channel.exchangeDeclare("exchangeName", "direct", true, false, null);

设置message为持久化类型

// 第三个参数指定了消息属性为持久化的，发送的消息将被rabbitmq落盘
channel.basicPublish("exchangeName", "key", MessageProperties.PERSISTENT_TEXT_PLAIN, "this is a message".getBytes("UTF-8"));

消费端负载均衡

当一条消息到达了Queue中，如果该队列上同时有多个Consumer，默认情况下RabbitMQ采用 Round-robin dispatching循环分发的方法轮流给每个Consumer发送消息。当某个Consumer任务比较繁忙，就会导致有的Consumer几乎无事可做，有的Consumer却毫无休息的机会。此时可以打开消费者负载均衡选项，用来控制每个Consumer队列的最大待处理消息个数：

// 表示当前consumer同时最多处理prefetchCount个消息
channel.basicQos(prefetchCount);

除了前文提到的direct类型的exchange，RabbitMQ提供了另外2种类型的exchange：fanout exchange和topic exchange：

fanout exchange会将消息发送到所有绑定到该exchange的queue中而不关注RoutingKey。
topic exchange更为灵活，并且使用更为广泛，它会根据RoutingKey进行模式匹配，只要RoutingKey匹配就会发送到对应的queue中。

RoutingKey正则匹配：

每一条消息RoutingKey会包含若干个标识符(以点号分隔)，上图中的表示可以匹配一个标识符，#表示可以匹配0个或多个标识符。上图中Q1与.orange.绑定，Q2与.*.rabbit和lazy.#绑定。假设一条消息的routingKey为lazy.orange.pig，那么Q1和Q2都将与这条消息匹配。通过对RoutingKey和BindingKey的精心设计，可以满足点对点(私信)、组播、广播等业务场景的通信需求。

事务消息

事务消息对性能的影响较大，不推荐使用。简单介绍一下相关的接口：

在当前channel上打开TX事务模式

channel.txSelect();

在当前channel上提交一个TX事务

 channel.txCommit();

在当前channel上回滚一个TX事务

channel.txRollback();

进阶使用

TTL, 延迟队列

RabbitMQ允许对queue和message设置TTL值(过期时间)

Per-Queue Message TTL: 通过在声明queue的时候设置x-message-ttl，可以控制被 publish 到 queue 中的 message 被丢弃前能够存活的时间。

// 声明一个消息过期时间为60s的队列
Map<String, Object> args = new HashMap<String, Object>();  
args.put("x-message-ttl", 60000);  
channel.queueDeclare("queue1", false, false, false, args);

Per-Message TTL: 发送消息时设置expiration字段可以指定消息的过期时间

// 声明一条过期时间为60s的消息
AMQP.BasicProperties properties = new AMQP.BasicProperties();  
properties.setExpiration("60000");  
channel.basicPublish("myexchange", "routingkey", properties, "This is a message".getBytes());

利用TTL和DLX实现延迟队列:

dead-letter-exchange(DLX)是一种特殊的exchange，当消息在一个队列中变成死信后(消息被拒绝且requeue为false，或TTL过期，或队列达到最大长度)，它能被重新publish到DLX。

通过TTL和DLX的配合使用，可以实现高可靠的延迟队列，方法如下：

HashMap<String, Object> arguments = new HashMap<String, Object>();  
// 设置DLX为rabbitmq自带的amq.direct exchange
arguments.put("x-dead-letter-exchange", "amq.direct");  
// 设置死信队列中的消息RoutingKey为dlxKey
arguments.put("x-dead-letter-routing-key", "dlxKey");  
// 使用上面设置的参数，声明一个名为delayQueue的队列，默认绑定到rabbitmq自带的名为""的exchange
channel.queueDeclare("delayQueue", false, false, false, args);

再声明一个接收延迟消息的队列：

// 声明一个名为RECV_QUEUE的queue
channel.queueDeclare("RECV_QUEUE", true, false, false, null);  
// 绑定到amq.direct exchange，且绑定的RoutingKey为dlxKey
channel.queueBind("RECV_QUEUE", "amq.direct", "dlxKey");

投递消息：

// 声明一条过期时间为60s的消息
AMQP.BasicProperties properties = new AMQP.BasicProperties();  
properties.setExpiration("60000");  
//投递到默认的""exchange，消息过期后会被重新投递到amq.direct，然后路由到RECV_QUEUE中
channel.basicPublish("", "routingkey", properties, "This is a message".getBytes());

接收消息：

//消费者在RECV_QUEUE队列中等待消息到来
channel.basicConsume("RECV_QUEUE", true, consumer);

队列自动删除

queue.declare 命令中的 x-expires 参数控制 queue 被自动删除前可以处于未使用状态的时间，即当前queue上没有任何Consumer的队列被自动删除的时间。

// queue未被使用30分钟后自动删除
Map<String, Object> args = new HashMap<String, Object>();  
args.put("x-expires", 1800000);  
channel.queueDeclare("myqueue", false, false, false, args);

队列最大长度

通过x-max-length参数可以设置单个队列的最大长度：

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-max-length", 10);
channel.queueDeclare("myqueue", false, false, false, args);

Attention

当队列达到最大长度限制，此时新消息到达时，队列头部的消息将被丢弃。

优先级队列

RabbitMQ提供了x-max-priority参数以支持优先级队列。

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-max-priority", 10);
channel.queueDeclare("my-priority-queue", true, false, false, args);

x-max-priority参数的值表示该队列支持的最大优先级。消息优先级可以通过下列方法设置：

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.priority(5);
channel.basicPublish("exchange_priority", "rk_priority", builder.build(), ("messages-" + i).getBytes());

消息优先级默认为0，其范围在0~x-max-priority之间，对于大于x-max-priority的值将作为x-max-priority处理。

使用消息优先级特性将带来一些性能损耗，并且如果一个空闲的消费者连接到空队列，此时新消息进入的话，消息不会花费任何时间在队列中等待，消息将直接被消费，此时的消息优先级将没有意义。

policy策略

如果我们使用客户端声明了一个queue并设置TTL之后想去更新这个TTL，除了重新声明队列，并重新设置TTL的办法之外，RabbitMQ提供了一种称为policy的更为灵活的办法去更新相关的参数：

policy通过正则表达式匹配指定的对象(queue或者exchange)，然后可以在这些queue或者exchange上应用我们的策略设置。管理平台相关policy的设置页面如下：

镜像队列

在集群环境中,队列只有元数据会在集群的所有节点同步,但队列中的数据只会存在于一个节点，数据没有冗余且容易丢，甚至在durable的情况下,如果所在的服务器节点宕机，就要等待节点恢复才能继续提供消息服务。

NQS利用RabbiMQ提供的镜像队列，高可用模式下队列数据将在所有节点同步并落盘，因此任一一个节点宕机将不影响服务的可用性。镜像队列也有主从之分，正常情况下只有master才对外服务，slave节点只提供备份服务。同样的，我们也可以通过policy来进行镜像队列的设置(上图的HA选项)。

Alternate Exchange(AE)

当一条消息无法被一个exchange路由到一个queue时，默认情况下消息将被丢弃。某些情况，需要知道被rabbitmq丢弃的消息有哪些，如果定义了AE, 消息将被路由到AE绑定的queue当中。

注册流控事件

当消费者的消费速度小于生产者的生成速度时，会出现消息堆积的情况，此时的性能将会出现大幅度下降，甚至还可能会出现流控，发生流控时的RabbitMQ将会停止生产。很多时候我们需要知道何时发生了流控事件，除了NQS提供的事件报警机制外，利用客户端本身提供的功能也能获得该通知：

ConnectionFactory factory = new ConnectionFactory();
        Connection connection = factory.newConnection();
        connection.addBlockedListener(new BlockedListener() {
            public void handleBlocked(String reason) throws IOException {
                // Connection is now blocked
            }

            public void handleUnblocked() throws IOException {
                // Connection is now unblocked
            }
        });

热门搜索

数据分析及可视化

数据研发

数据治理

数据服务化

计算服务

网络服务

数据库

消息

存储与 CDN

管理与监控

其他

使用介绍

Attention

名词说明

使用说明

入门示例

生产者

消费者

保证消息可靠性

Attention

生产者 confirm 机制

消费者ACK机制

persist持久化机制

消费端负载均衡

事务消息

进阶使用

TTL, 延迟队列

队列自动删除

队列最大长度

Attention

优先级队列

policy策略

镜像队列

Alternate Exchange(AE)

注册流控事件