RabbitMQ 笔记

本文介绍: 确保消息在server 出现问题或者recovery能恢复： declare i t a s durable in th e producer a nd consumer code.

Me ssa g e dur a bi li t y

确保消息在server 出现问题或者recovery能恢复： de cla re i t a s durable in th e producer a nd consumer code.

boolean durable = true;
channel.queueDeclare("hello", durable, false, false, null);

Que ue 指定

//使用指定的queue，以便协同,注意各方声明时候属性要一致
        channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
        channel.queuePurge(RPC_QUEUE_NAME);

Fai r dispatch

In order t o def eat that we can use th e basicQos method wi th the pre fetchCount = 1 set ting. Thi s tel ls Rab bitMQ not to gi v e more than on e message to a worker at a time.

exchange

rabbitmq ctl list _exchange s
Lis ting exchange s fo r v host / …
name type
amq.headers headers
amq.rabbitmq.trace topic
amq.top ic top ic
amq.match headers
direct
amq.direct direct
amq.fanout fanout

The re are a few exchange types available: direct, top ic, headers and fanout

默认 exchange

default exchange, whi ch we identify b y the empty string (“”).

channel.basicPublish("", "hello", null, message.getBytes());

将queue与exchange 进行bind

channel.queueBind(queueName, EXCHANGE_NAME, "black");

The meaning of a binding key de p ends on the exchange ty pe

Direct exchange

We were using a fanout exchange, whi ch doesn’t give us mu ch flexi bility – it’s only cap able of min d less b ro ad cast ing.
We will use a direct ex change in stead. The r outing algo rithm behi nd a direct ex change is simple – a message go es to the que ues whose binding key exa ctly matches the routing key of the message.
bind multiple que ues with the same binding key. 基于全量 match，即string equal
can’t do routing based on multiple criteria.

Top ics

a message sent with a partic ul ar routing key will be de live red to all the que ues that are bound with a matching binding key
基于表达式的分发

star（*） can su bstitute for exa ctly one word.
hash（ #） can su bstitute for zero or more words.

When a queue is bound with “#” (hash) binding key – it will receive all the messages, reg ardless of the routing key – like in fanout exc hange.
When sp ecial characte rs, “*” (star) and “#” (hash), aren’t used in bindings, the topic exc hange will behave just like a direct one.

Heade rs Exchange

A headers exchange is designed for routing on multiple attributes that are more easily expressed as message headers than a routing key. Headers exchanges ignore the routing key attribute. Instead, the attributes used for routing are taken from the headers attribute. A message is considered matching if the value of the header equals the value sp ecified up on binding.
It is possible to bind a queue to a headers exchange using more than one header for matching. In this case, the broker needs one more piece of information from the application developer, namely, should it consider messages with any of the headers matching, or all of them? This is what the “x-match” binding argument is for. When the “x-match” argument is set to “any”, just one mat c hing header value is su ffi cient. Alternatively, setting “x-mat ch” to “all” mandates that all the values must mat ch.
For “any” and “all”, headers beginning with the string x- will not be used to eva luate mat ches. Setting “x-match” to “any-with-x” or “all-with-x” will also use headers beginning with the string x- to eva luate matches.
Headers exchanges can be looked upon as “direct exchanges on ste roi ds”. Because they route based on header values, they can be used as di rect exchanges where the routing key does not have to be a string; it could be an integer or a hash (dict ionary) for example.

plugins

I have no name!@bca0cb31ba36:/$ rabbitmq-plugins list

Listing plugins with pattern "." ...Configured: E = explicitly enabled; e = implicitly enabled| Status: * = running on rabbit@queue-ram1|/[  ] rabbitmq_amqp1_0                  3.12.10[  ] rabbitmq_auth_backend_cache       3.12.10[  ] rabbitmq_auth_backend_http        3.12.10[  ] rabbitmq_auth_backend_ldap        3.12.10[  ] rabbitmq_auth_backend_oauth2      3.12.10[  ] rabbitmq_auth_mechanism_ssl       3.12.10[  ] rabbitmq_consistent_hash_exchange 3.12.10[  ] rabbitmq_event_exchange           3.12.10[  ] rabbitmq_federation               3.12.10[  ] rabbitmq_federation_management    3.12.10[  ] rabbitmq_jms_topic_exchange       3.12.10[  ] rabbitmq_management               3.12.10[E] rabbitmq_management_agent         3.12.10
[  ] rabbitmq_mqtt                     3.12.10
[  ] rabbitmq_peer_discovery_aws       3.12.10
[  ] rabbitmq_peer_discovery_common    3.12.10
[  ] rabbitmq_peer_discovery_consul    3.12.10
[  ] rabbitmq_peer_discovery_etcd      3.12.10
[  ] rabbitmq_peer_discovery_k8s       3.12.10
[E*] rabbitmq_prometheus               3.12.10
[  ] rabbitmq_random_exchange          3.12.10
[  ] rabbitmq_recent_history_exchange  3.12.10
[  ] rabbitmq_sharding                 3.12.10
[  ] rabbitmq_shovel                   3.12.10
[  ] rabbitmq_shovel_management        3.12.10
[  ] rabbitmq_stomp                    3.12.10
[  ] rabbitmq_stream                   3.12.10
[  ] rabbitmq_stream_management        3.12.10
[  ] rabbitmq_top                      3.12.10
[  ] rabbitmq_tracing                  3.12.10
[  ] rabbitmq_trust_store              3.12.10
[e*] rabbitmq_web_dispatch             3.12.10
[  ] rabbitmq_web_mqtt                 3.12.10
[  ] rabbitmq_web_mqtt_examples        3.12.10
[  ] rabbitmq_web_stomp                3.12.10
[  ] rabbitmq_web_stomp_examples       3.12.10

参考 URL
Firehose Tracer Firehose Tracer — RabbitMQ

Plugin Development Basics — RabbitMQ

Queue Properties

Queues have properties that define how they behave. There is a set of mandatory properties and a map of optional ones:

Name
Durable (the queue will survive a broker restart)
Exclusive (used by only one connection and the queue will be deleted when that connection closes)
Auto-delete (queue that has had at least one consumer is deleted when last consumer unsubscribes)
Arguments (optional; used by plugins and broker-specific features such as message TTL, queue length limit, etc)

Queue type (e.g. quorum or classic) –推荐quorum，单quorum不支持 mes ssage pr iority，对应 append-only的log 类，推荐使用 stream -Stream Plugin — RabbitMQ

在这里插入图片描述

Consumer pr iorities allow you to ensure that high pr iority consumers receive messages while they are active, with messages only going to lower pr iority consumers when the high pr iority consumers block.

When Not to Use Quorum Queues

In some cases quorum queues should not be used. They typi cally involve:

Temp ora ry nature of queues: trans ient or exclusive queues, high queue churn (declaration and delet ion rates)
Lowest possible latency: the underl ying consensus algorithm has an inherently higher latency due to its data safety features
When data safety is not a pr iority (e.g. applications do not use manual acknowledgements and publisher confirms are not used)
Very long queue backlogs (streams are lik ely to be a better fit)

不同队列类型支持的属性

在这里插入图片描述

Durability

Trans ient queues will be deleted on node boot. They therefore will not survive a node restart, by design. Messages in trans ient queues will also be discarded.
Durable queues will be recovered on node boot, including messages in them published as persistent

Exclusive Queues

An exclusive queue can only be used (consumed from, purged, deleted, etc) by its declaring connection.

Repli cated and Distributed Queues

Quorum queues is replicated, data safety and consi ste ncy-oriented queue type. Clas sic queues historically supported replication but it is deprecated and should be avoided.
Queues can also be federated across lo os ely coup led nodes or clusters.
Note that intra-cluster replication and federation are orthogonal features and should not be considered di rect alternatives.
Streams is another replicated data structure supported by RabbitMQ, with a different set of supported operations and features.

Time-to-Live and Length Limit

Queues can have their length limited. Queues and messages can have a TTL.

RabbitMQ中的消息 优先级是如何 实现的？

注意只有classic的队列支持优先级。

By default, RabbitMQ classic queues do not support priorities. When creating priority queues, a maximum priority can be chosen as you see fit. When choosing a priority value, the fol lo wing factors need to be considered:

There is some in-memory and on-di sk cost per priority level per queue. There is also an additional CPU cost, especially when consuming, so you may not wish to create huge numbers of levels.
The message priority field is defined as an unsigned byte, so in practice priorities should be between 0 and 255.
Messages without priority property are treated as if their priority were 0. Messages with a priority which is higher than the queue’s maximum are treated as if they were published with the maximum priority.

channels

Maxi mum Number of Chann els per Connection
On the server side, the limit is control led using the channel_max:

# no more 100 channels can be opened on a connection at the same time
channel_max = 100

Clients can be configured to allow fewer channels per connection. With RabbitMQ Java client, ConnectionFactory#setRequestedChannelMax is the method that controls the limit:

ConnectionFactory cf = new ConnectionFactory();
// Ask for up to 32 channels per connection. Will have an effect as long as the server is configured
// to use a higher limit, otherwise the server's limit will be used.
cf.setRequestedChannelMax(32);

Acknowledgements –确认

delive ry tag –每个消息的唯一识别标志

When manual acknowledgements are used, any delive ry (mes sage) that was not acked is auto matically requeued when the channel (or connection) on which the delivery happened is closed. This includes TCP connection loss by clients, consumer ap pli cation (process) failures, and channel-level protocol exceptions (covered below).

Consumer Acknowledgement Modes and Data Safety Considerations

basic.ack is used for positive acknowledgements
basic.nack is used for negative acknowledgements (note: this is a RabbitMQ extension to AMQP 0-9-1)
basic.reject is used for negative acknowledgements but has one limitation compared to basic.nack

channel.basicAck 支持批量确认
Manual acknowledgements can be batched to reduce network traffic. This is done by setting the multiple field of acknowledgement methods (see above) to true.

    // AMQImpl.Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
    channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
// 每次处理1 条消息
    channel.basicQos(1);

    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
        String message = new String(delivery.getBody(), "UTF-8");

        System.out.println(" [x] Received '" + message + "'");
        try {
            doWork(message);
        } finally {
            System.out.println(" [x] Done");
            // false 待办单个
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    };
    //    public String basicConsume(String queue, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback) throws IOException {
    // 第二次参数标志是否自动ack
    channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { });

Negative Acknowledgement and Requeuing of Deliveries

This behaviour is control led by the requeue field. When the field is set to true, the broker will requeue the delivery (or multiple deliveries, as will be explained shortly) with the speci fied delivery tag. Alternatively, when this field is set to fa lse, the mes sage will be routed to a Dead Letter Exchange if it is configured, otherwise it will be discarded.

 // negatively acknowledge, the message will
             // be discarded
             channel.basicReject(deliveryTag, false);
交给其它consumer 处理
 // requeue the delivery
             channel.basicReject(deliveryTag, true);
 basic.nack 和Basic.Reject 的差异是支持批量

 // requeue all unacknowledged deliveries up to
             // this delivery tag
             channel.basicNack(deliveryTag, true, true);

三者的实现代码

public void basicAck(long deliveryTag, boolean multiple) throws IOException {
    this.transmit(new AMQImpl.Basic.Ack(deliveryTag, multiple));
    this.metricsCollector.basicAck(this, deliveryTag, multiple);
}

public void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException {
    this.transmit(new AMQImpl.Basic.Nack(deliveryTag, multiple, requeue));
    this.metricsCollector.basicNack(this, deliveryTag);
}

public void basicReject(long deliveryTag, boolean requeue) throws IOException {
    this.transmit(new AMQImpl.Basic.Reject(deliveryTag, requeue));
    this.metricsCollector.basicReject(this, deliveryTag);
}

Publisher Confirms –生产者 确认

To enable confirms, a client sends the confirm.select meth od. Depending on whether no-wait was set or not, the broker may respond with a confirm.select-ok. Once the confirm.select meth od is used on a channel, it is said to be in confirm mode.
通过计数来确认
Once a channel is in confirm mode, both the broker and the client count messages (counting starts at 1 on the first confirm.select). The broker then confirms messages as it handles them by sending a basic.ack on the same channel.
如要百分百确认消息落盘了，可以用confirm 模式
In order to guarantee persistence, a client should use confirms. If the publisher’s channel had been in confirm mode, the publisher would not have received an ack for the lost mes sage (since the message hadn’t been written to disk yet).
可以在发送后同步获取（ channel.waitForConfirmsOrDie(5_000);），也可以异步获取是否发送成功（channel.addConfirmListener((sequenceNumber, multiple) -> {
）参考PublisherConfirms类

Channel Prefetch Setting (QoS)

获取数量

A value of 0 means “no limit”, allowing any number of unacknowledged messages.
The QoS setting can be configured for a specific channel or a specific consumer. The Consumer Prefetch guide explains the effects of this scoping.

 // 每次处理1 条消息
        channel.basicQos(1);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");

            System.out.println(" [x] Received '" + message + "'");
            try {
                doWork(message);
            } finally {
                System.out.println(" [x] Done");
                // false 待办单个
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

            }
        };
        //    public String basicConsume(String queue, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback) throws IOException {
        // 第二次参数标志是否自动ack
        channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { });
        channel.consum
        //channel.getDefaultConsumer().handleCancel();

Single Consumer

The fol lo wing basic example in Java will receive a maximum of 10 unacknowledged messages at once:

Channel channel = ...;
Consumer consumer = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue", false, consumer);

A value of 0 is treated as inf inite, allowing any number of unacknowledged messages.

Channel channel = ...;
Consumer consumer = ...;
channel.basicQos(0); // No limit for this consumer
channel.basicConsume("my-queue", false, consumer);

Independent Consumers

This example starts two consumers on the same channel, each of which will independently receive a maximum of 10 unacknowledged messages at once:

Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10); // Per consumer limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

Multiple Consumers Sharing the Limit

Channel channel = ...;
Consumer consumer1 = ...;
Consumer consumer2 = ...;
channel.basicQos(10, false); // Per consumer limit
channel.basicQos(15, true);  // Per channel limit
channel.basicConsume("my-queue1", false, consumer1);
channel.basicConsume("my-queue2", false, consumer2);

Configurable Default Prefetch

RabbitMQ can use a default pre fetch that will be ap plied if the consumer doesn’t specify one. The value can be configured as rabbit.default_consumer_prefetch in the advanced configuration file:
%% advanced.config file
[
{rabbit, [
{default_consumer_prefetch, {false,250}}
]
}
].

Queue Length Limit

The default behaviour for RabbitMQ when a maximum queue length or size is set and the maximum is reached is to drop or dead–letter messages from the front of the queue (i.e. the ol dest messages in the queue). To modify this behaviour, use the overflow setting des cribed below.

Queue Overflow Behaviour

Use the overflow setting to configure queue overflow behaviour. If overflow is set to reject-publish or reject-publish-dlx, the most recently published messages will be discarded. In addition, if publisher confirms are enabled, the publisher will be informed of the reject via a basic.nack message. If a message is routed to multiple queues and rejected by at least one of them, the channel will inform the publisher via basic.nack. The message will still be published to all other queues which can enqueue it. The difference between reject-publish and reject-publish-dlx is that reject-publish-dlx also dead–letters rejected messages.

What is a Lazy Queue

A “lazy queue” is a classic queue which is running in lazy mode. When the “lazy” queue mode is set, messages in classic queues are moved to disk as early as practically possible. These messages are loaded into RAM only when they are requested by consumers.

Exchange to Exchange Bindings

Java Client Example
Use the Channel#exchangeBind method. The following example binds an exchange “destination” to “source” with routing key “routingKey”.

Channel ch = conn.createChannel();
ch.exchangeBind("destination", "source", "routingKey");
---

Consumer Cancel Notifi cation

用于优雅推出消费端
if the client issues a basic.cancel on the same channel, which will cause the consumer to be cancelled and the server replies with a basic.cancel-ok.

public void basicCancel(final String consumerTag) throws IOException {
处理异常退出
channel.queueDeclare(queue, false, true, false, null);
Consumer consumer = new DefaultConsumer(channel) {
    @Override
    public void handleCancel(String consumerTag) throws IOException {
        // consumer has been cancelled unexpectedly
    }
};
channel.basicConsume(queue, consumer);

boolean autoAck = false;
channel.basicConsume(queueName, autoAck, "myConsumerTag",
     new DefaultConsumer(channel) {
         @Override
         public void handleDelivery(String consumerTag,
                                    Envelope envelope,
                                    AMQP.BasicProperties properties,
                                    byte[] body)
             throws IOException
         {
             String routingKey = envelope.getRoutingKey();
             String contentType = properties.getContentType();
             long deliveryTag = envelope.getDeliveryTag();
             // (process the message components here ...)
             channel.basicAck(deliveryTag, false);
         }
     });

channel.basicCancel(consumerTag);

rabbitmq cluster

默认通过 cookie 识别为同一cluster

In case of a node failure, clients should be able to reconnect to a different node, recover their topology and continue operation. For this reason, most client libraries accept a list of endpoints (hostnames or IP addresses) as a connection option. The list of hosts will be used during initial connection as well as connection recovery, if the client supports it. See documentation guides for individual clients to learn more.

Node Failure Handling

RabbitMQ brokers tolerate the failure of individual nodes. Nodes can be started and stopped at will, as long as they can contact a cluster member node known at the time of shut down.
Quorum queue allows queue contents to be replicated across multiple cluster nodes with parallel replication and a predictable leader election and data safety behavior as long as a majority of replicas are online.

Non-replicated classic queues can also be used in clusters. Non-mirrored queue behaviour in case of node failure depends on queue durability.

client 端

Using Lists of Endpoints
It is possible to specify a list of endpoints to use when connecting. The first reachable endpoint will be used. In case of connection failures, using a list of endpoints makes it possible for the application to connect to a different node if the original one is down.
To use multiple of endpoint, provide a list of Addresses to ConnectionFactory#newConnection. An Address represents a hostname and port pair.

Address[] addrArr = new Address[]{ new Address(hostname1, portnumber1)
                                 , new Address(hostname2, portnumber2)};
Connection conn = factory.newConnection(addrArr);