当前位置：
  1. 首页
  2. 互联网
  3. 正文
本文介绍: 确保消息在server 出现问题或者recovery能恢复： declare it as durable in the producer and consumer code.

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

Topics

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

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:

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:

不同队列类型支持的属性

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

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:

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

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 –生产者确认

Channel Prefetch Setting (QoS)

获取数量

 // 每次处理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和落盘保证。

原文地址:https://blog.csdn.net/weixin_40455124/article/details/134747139

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。

如若转载，请注明出处：http://www.7code.cn/show_29844.html

如若内容造成侵权/违法违规/事实不符，请联系代码007邮箱：suwngjj01@126.com进行投诉反馈，一经查实，立即删除！

上一篇 房产中介管理信息系统的设计与实现
下一篇 opencv学习三：保存图片

相关文章

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

插入图片
浏览器会保存昵称、邮箱和网站cookies信息，下次评论时使用。