RabbitMQ 笔记

Message durability

确保消息在server 出现问题或者recovery能恢复： declare it as durable in the producer and consumer code.

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

Queue 指定

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

Fair dispatch

In order to defeat that we can use the basicQos method with the prefetchCount = 1 setting. This tells RabbitMQ not to give more than one message to a worker at a time.

---

exchange

rabbitmqctl list_exchanges
Listing exchanges for vhost / …
name type
amq.headers headers
amq.rabbitmq.trace topic
amq.topic topic
amq.match headers
direct
amq.direct direct
amq.fanout fanout

There are a few exchange types available: direct, topic, headers and fanout

默认exchange

default exchange, which we identify by the empty string (“”).

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

将queue与exchange 进行bind

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

The meaning of a binding key depends on the exchange type

Direct exchange

• We were using a fanout exchange, which doesn’t give us much flexibility - it’s only capable of mindless broadcasting.

• We will use a direct exchange instead. The routing algorithm behind a direct exchange is simple - a message goes to the queues whose binding key exactly matches the routing key of the message.

• bind multiple queues with the same binding key. 基于全量match，即string equal

• can’t do routing based on multiple criteria.

---

Topics

a message sent with a particular routing key will be delivered to all the queues that are bound with a matching binding key
基于表达式的分发

• star（*） can substitute for exactly one word.
• hash（ #） can substitute for zero or more words.

When a queue is bound with “#” (hash) binding key - it will receive all the messages, regardless of the routing key - like in fanout exchange.
When special characters, “*” (star) and “#” (hash), aren’t used in bindings, the topic exchange will behave just like a direct one.

---

Headers 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 specified upon 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 matching header value is sufficient. Alternatively, setting “x-match” to “all” mandates that all the values must match.
For “any” and “all”, headers beginning with the string x- will not be used to evaluate matches. Setting “x-match” to “any-with-x” or “all-with-x” will also use headers beginning with the string x- to evaluate matches.
Headers exchanges can be looked upon as “direct exchanges on steroids”. Because they route based on header values, they can be used as direct exchanges where the routing key does not have to be a string; it could be an integer or a hash (dictionary) 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不支持messsage priority，对应append-only的log 类，推荐使用stream -Stream Plugin — RabbitMQ

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

When Not to Use Quorum Queues

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

• Temporary nature of queues: transient or exclusive queues, high queue churn (declaration and deletion rates)
• Lowest possible latency: the underlying consensus algorithm has an inherently higher latency due to its data safety features
• When data safety is not a priority (e.g. applications do not use manual acknowledgements and publisher confirms are not used)
• Very long queue backlogs (streams are likely to be a better fit)

不同队列类型支持的属性

Durability

Transient queues will be deleted on node boot. They therefore will not survive a node restart, by design. Messages in transient 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.

Replicated and Distributed Queues

Quorum queues is replicated, data safety and consistency-oriented queue type. Classic queues historically supported replication but it is deprecated and should be avoided.
Queues can also be federated across loosely coupled nodes or clusters.
Note that intra-cluster replication and federation are orthogonal features and should not be considered direct 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 following factors need to be considered:

• There is some in-memory and on-disk 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

Maximum Number of Channels per Connection
On the server side, the limit is controlled 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 -确认

delivery tag -每个消息的唯一识别标志

When manual acknowledgements are used, any delivery (message) that was not acked is automatically requeued when the channel (or connection) on which the delivery happened is closed. This includes TCP connection loss by clients, consumer application (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 controlled 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 specified delivery tag. Alternatively, when this field is set to false, the message 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 method. Depending on whether no-wait was set or not, the broker may respond with a confirm.select-ok. Once the confirm.select method 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 message (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 following 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 infinite, 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 prefetch that will be applied 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 oldest messages in the queue). To modify this behaviour, use the overflow setting described 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 Notification

• 用于优雅推出消费端
• 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 shutdown.
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);

Disconnecting from RabbitMQ： To disconnect, simply close the channel and the connection:

channel.close();
conn.close();

RAM node

RAM node不落盘，可以选择RAM 对外提升性能，DISC node作为mirror和落盘保证。