AMQP 1.0 客户端库

此页面记录了 RabbitMQ 4.0 或更高版本 的 AMQP 1.0 客户端库的用法。

RabbitMQ 团队支持以下库

• RabbitMQ AMQP 1.0 Java 客户端
• RabbitMQ AMQP 1.0 .NET 客户端
• RabbitMQ AMQP 1.0 Go 客户端
• RabbitMQ AMQP 1.0 Python 客户端

应用程序开发者将在此处找到如何将这些库用于最常见的用例。有关许可、下载、依赖管理、高级和特定用法以及配置等其他信息，请参阅各个库的存储库中的 README 页面。

概述

RabbitMQ 团队维护了一组 设计和优化 用于 RabbitMQ 的 AMQP 1.0 客户端库。它们在 AMQP 1.0 之上提供了简单、安全且功能强大的 API。应用程序可以使用这些库发布和消费消息，并以跨编程语言的一致方式管理服务器拓扑。这些库还提供高级功能，例如自动连接和拓扑恢复，以及与队列的连接亲和性。

注意

RabbitMQ 与任何 AMQP-1.0 兼容的客户端库兼容。并非强制要求将 RabbitMQ AMQP 1.0 客户端库与 RabbitMQ 一起使用，但强烈建议应用程序这样做以获得最佳体验。

安全性

RabbitMQ AMQP 1.0 客户端库默认是安全的，它们始终创建持久实体，并且始终发布持久消息。

保证

RabbitMQ AMQP 1.0 客户端库提供至少一次的保证。

Broker 始终确认已正确处理已发布的消息。发布者通过在使用 unsettled 发送方结算模式 和 first 接收方结算模式 创建它们时实现这一点。

消费者必须始终指示消息处理结果给 Broker。消费者在创建时使用与发布者相同的设置（first 接收方结算模式 和 unsettled 发送方结算模式）。

客户端 API

本节介绍如何使用 RabbitMQ AMQP 1.0 客户端库连接到集群、发布和消费消息。

连接

库提供了一个节点或节点集群的入口点。它的名称是“环境”。环境允许创建连接。它可以包含在连接之间共享的基础设施相关配置设置（例如，Java 的线程池）。以下是如何创建环境

Java

创建环境

import com.rabbitmq.client.amqp.*;
import com.rabbitmq.client.amqp.impl.AmqpEnvironmentBuilder;

// ...

// create the environment instance
Environment environment = new AmqpEnvironmentBuilder()
    .build();
// ...
// close the environment when the application stops
environment.close();

C#

创建环境

using RabbitMQ.AMQP.Client;
using RabbitMQ.AMQP.Client.Impl;

// ...

// create the environment instance
IEnvironment environment = await AmqpEnvironment.CreateAsync(
    ConnectionSettingBuilder.Create().Build());
// ...
// close the environment when the application stops
await environment.CloseAsync();

Python

创建环境

from rabbitmq_amqp_python_client import Environment

# ...

# create the environment instance
environment = Environment("amqp://guest:guest@localhost:5672/")
# ...
# close the environment when the application stops
environment.close()

Go

创建环境

import (
    rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)
// ...
// create the environment instance for a single node
env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
// in case you have multiple endpoints you can use the following:
// clusterEnv := rmq.NewClusterEnvironment([]rmq.Endpoint{
//      {Address: "amqp://server1", Options: &rmq.AmqpConnOptions{}},
//      {Address: "amqp://server2", Options: &rmq.AmqpConnOptions{}},
//     })

// ...
// close the connections
env.CloseConnections(context.Background())

一个应用程序进程通常只有一个环境实例。应用程序必须在退出时关闭环境以释放其资源。

应用程序从环境中打开连接。它们必须指定适当的设置以连接到集群节点（URI、凭据）。

Java

打开连接

// open a connection from the environment
Connection connection = environment.connectionBuilder()
    .uri("amqp://admin:admin@localhost:5672/%2f")
    .build();
// ...
// close the connection when it is no longer necessary
connection.close();

C#

打开连接

// open a connection from the environment setting   
IConnection connection = await environment.CreateConnectionAsync();

//open a connection from the environment with different settings
ConnectionSettingBuilder otherSettingBuilder = ConnectionSettingBuilder.Create()
    .ContainerId("my_containerId")
    .Host("localhost");
IConnection connection = await environment.CreateConnectionAsync(otherSettingBuilder.Build());
// ...
// close the connection when it is no longer necessary
await connection.CloseAsync();

Python

打开连接

# open a connection from the environment setting   
connection = environment.connection()

# close the connection when it is no longer necessary
connection.close()

Go

打开连接

// open a connection from the environment setting   
connection, err := env.NewConnection(context.Background())

// close the connection when it is no longer necessary
connection.close()

库默认使用 ANONYMOUS SASL 身份验证机制。连接应为长期存在的对象，应用程序应避免连接抖动。当不再需要它们时，必须关闭它们。

发布

必须创建一个发布者才能发布消息。发布者将发布消息的目标通常在创建时设置，但也可以在每条消息的基础上设置。

以下是如何在创建时设置目标的情况下声明发布者

Java

创建发布者

Publisher publisher = connection.publisherBuilder()
    .exchange("foo").key("bar")
    .build();
// ...
// close the publisher when it is no longer necessary 
publisher.close();

C#

创建发布者

// The publisher can use exchange (optionally with a key) or queue to publish messages. 
IPublisher publisher = await connection.PublisherBuilder().Exchange("foo").Key("bar")
    .BuildAsync();

// ...
// close the publisher when it is no longer necessary 
await publisher.CloseAsync();
publisher.Dispose();

Python

创建发布者

# The publisher can use exchange (optionally with a key) or queue to publish messages. 
# You can use the AddressHelper utility class to get the addr from the exchange name and the key
exchange_address = AddressHelper.exchange_address("foo", "bar")
publisher = connection.publisher(addr)
# close the publisher when it is no longer necessary 
publisher.close()

Go

创建发布者

// you can use ExchangeAddress and QueueAddress to publish to the queue
publisher, err := amqpConnection.NewPublisher(context.Background(), &rmq.ExchangeAddress{
        Exchange: "foo",
        Key:      "bar",
    }, nil)

// close the publisher when it is no longer necessary 
publisher.close()

在前面的示例中，使用发布者发布的每条消息都将发送到具有 bar 路由键的 foo 交换机。

信息

RabbitMQ 使用包含交换机、队列和绑定的 AMQ 0.9.1 模型。

消息从发布者实例创建。它们遵循 AMQP 1.0 消息格式。可以定义正文（作为字节数组）、标准属性和应用程序属性。

当消息发布时，Broker 会在异步回调中指示如何处理该消息。客户端应用程序根据 Broker 为消息返回的状态（AMQP 术语中的“结果”）采取适当的措施（例如，如果消息未被 accepted，则将消息存储在另一个位置）。

以下代码片段显示了如何创建消息、发布消息以及处理来自 Broker 的响应

Java

发布消息

// create the message
Message message = publisher
    .message("hello".getBytes(StandardCharsets.UTF_8))
    .messageId(1L);

// publish the message and deal with broker feedback
publisher.publish(message, context -> {
    // asynchronous feedback from the broker
    if (context.status() == Publisher.Status.ACCEPTED) {
        // the broker accepted (confirmed) the message
    } else {
        // deal with possible failure
    }
});

C#

发布消息

// create the message
var message = new AmqpMessage("Hello");
// publish the message and deal with broker feedback
// The result is synchronous, use a `List<Task<PublishResult>>` to increase the performances 
PublishResult pr = await publisher.PublishAsync(message);
  switch (pr.Outcome.State)
    {
        case OutcomeState.Accepted:
            // the broker accepted (confirmed) the message
            break;
        case OutcomeState.Released:
            // the broker could not route the message anywhere 
            break;
        case OutcomeState.Rejected:
            // at least one queue rejected the message
            break;
    }

Python

发布消息

# create the message
message = Message(body="Hello")
# publish the message and deal with broker feedback
# The result is synchronous
status = publisher.publish(Message(message)
match status.remote_state:
    case OutcomeState.ACCEPTED:
        # the broker accepted (confirmed) the message
    case OutcomeState.RELEASED:
        # the broker could not route the message anywhere 
    case OutcomeState.REJECTED:
        # at least one queue rejected the message

Go

发布消息

// create the message
message =  rmq.NewMessage([]byte("Hello")
// publish the message and deal with broker feedback
publishResult, err := publisher.Publish(context.Background(),message)
if err != nil {
    // there is an error 
}
    switch publishResult.Outcome.(type) {
        case *rmq.StateAccepted:
            // the broker accepted (confirmed) the message
            break
        case *rmq.StateReleased:
           // the broker could not route the message anywhere 
            break
        case *rmq.StateRejected:
            // at least one queue rejected the message
            stateType := publishResult.Outcome.(*rmq.StateRejected)
            if stateType.Error != nil {
                // in case there is error 
            }
            break
        default:
            // these status are not supported. Leave it for AMQP 1.0 compatibility
            // see: https://rabbitmq.cn/docs/amqp#outcomes
        }

上面的发布者示例将消息发送到给定的交换机以及给定的路由键，但这并非发布者唯一支持的目标。以下是发布者支持的非空目标

Java

创建具有不同目标的发布者

// publish to an exchange with a routing key
Publisher publisher1 = connection.publisherBuilder()
    .exchange("foo").key("bar") // /exchanges/foo/bar
    .build();

// publish to an exchange without a routing key
Publisher publisher2 = connection.publisherBuilder()
    .exchange("foo") // /exchanges/foo
    .build();

// publish to a queue
Publisher publisher3 = connection.publisherBuilder()
    .queue("some-queue") // /queues/some-queue
    .build();

C#

创建具有不同目标的发布者

// publish to an exchange with a routing key
Publisher publisher = await connection.PublisherBuilder()
    .Exchange("foo")
    .Key("bar")
    .BuildAsync();


// publish to an exchange without a routing key
Publisher publisher = await connection.PublisherBuilder()
    .Exchange("foo") // /exchanges/foo
    .BuildAsync();

// publish to a queue
  IPublisher publisher = await _connection.PublisherBuilder()
  .Queue("some-queue")// /queues/some-queue
  .BuildAsync();

Python

创建具有不同目标的发布者

# publish to an exchange with a routing key
# You can use the AddressHelper utility class to get the addr from the exchange name and the key
exchange_address = AddressHelper.exchange_address("foo", "bar")
publisher = connection.publisher(addr)


# publish to an exchange without a routing key
exchange_address = AddressHelper.exchange_address("foo")
publisher = connection.publisher(addr)

# publish to a queue
queue_address = AddressHelper.queue_address("some-queue")
publisher = connection.publisher(queue_address)

Go

创建具有不同目标的发布者

// publish to an exchange with a routing key
publisher, err := connection.NewPublisher(context.Background(), &ExchangeAddress{
            Exchange: "foo",
            Key:      "bar",
        }, nil)


// publish to an exchange without a routing key
publisher, err := connection.NewPublisher(context.Background(), &ExchangeAddress{
            Exchange: "foo",
        }, nil)
    

// publish to a queue
publisher, err := connection.NewPublisher(context.Background(), &QueueAddress{
    Queue: "some-queue"}, 
    nil)
        

信息

库将 API 调用转换为 地址格式 v2。

也可以在每条消息的基础上定义目标。必须定义没有任何目标的发布者，并且每条消息都在属性部分的 to 字段中定义其目标。库在消息创建 API 中提供帮助程序来定义消息目标，从而避免处理 地址格式。

以下代码片段显示了如何创建没有目标的发布者并定义具有不同目标类型的消息

Java

在消息中设置目标

// no target defined on publisher creation
Publisher publisher = connection.publisherBuilder()
    .build();

// publish to an exchange with a routing key
Message message1 = publisher.message()
    .toAddress().exchange("foo").key("bar")
    .message();

// publish to an exchange without a routing key
Message message2 = publisher.message()
    .toAddress().exchange("foo")
    .message();

// publish to a queue
Message message3 = publisher.message()
    .toAddress().queue("my-queue")
    .message();

C#

在消息中设置目标

// no target defined on publisher creation
IPublisher publisher = await connection.PublisherBuilder()
                       .BuildAsync();

// publish to an exchange with a routing key
IMessage message = new AmqpMessage("Hello!").ToAddress()
    .Exchange("foo")
    .Key("bar")
    .Build()

await aPublisher.PublishAsync(message);

// publish to a queue
IMessage message = new AmqpMessage("Hello!").ToAddress()
    .Queue("foo")
    .Build()

await aPublisher.PublishAsync(message);

Python

在消息中设置目标

# no target defined on publisher creation
publisher = connection.publisher()

# publish to an exchange with a routing key
# You can use the AddressHelper.message_to_address_helper 
# utility class to set the destination in the message
message = Message(body="Hello!")
exchange_address = AddressHelper.exchange_address("foo", "bar")
message = AddressHelper.message_to_address_helper(message, exchange_address)
publisher.publish(message)

# publish to a queue
message = Message(body="Hello!")
queue_address = AddressHelper.queue_address("some-queue")
message = AddressHelper.message_to_address_helper(message, queue_address)
publisher = publisher.publish(message)

Go

在消息中设置目标

// // no target defined on publisher creation (nil)
publisher, err := connection.NewPublisher(context.Background(), nil,
                 nil)

// create message with an exchange with a routing key
msg, err = NewMessageWithAddress([]byte("hello"), &ExchangeAddress{
            Exchange: "foo",
            Key:      "bar",
        })

// create a message with a queue 
msg, err = NewMessageWithAddress([]byte("hello"), &QueueAddress{
    Queue: "some-queue"})


// use the publish
publishResult, err = publisher.Publish(context.Background(), msg)
// ...
        

消费

消费者创建

创建消费者包括指定要从中消费的队列和处理消息的回调

Java

创建消费者

Consumer consumer = connection.consumerBuilder()
    .queue("some-queue")
    .messageHandler((context, message) -> {
        byte[] body = message.body();
        // ...
        context.accept(); // settle the message
    })
    .build(); // do not forget to build the instance!

C#

创建消费者

IConsumer consumer = await connection.ConsumerBuilder()
    .Queue("some-queue")
    .MessageHandler(async (context, message) =>
    {
        // deal with the message
        await context.AcceptAsync();// settle the message
    }
).BuildAndStartAsync();

Python

创建消费者

class MyMessageHandler(AMQPMessagingHandler):

    def __init__(self):
        super().__init__()

    def on_message(self, event: Event):
        # ...
        
        self.delivery_context.accept(event)# settle the message
        
queue_address = AddressHelper.queue_address("some-queue")
consumer = connection.consumer(queue_address, message_handler=MyMessageHandler())
consumer.run()

Go

创建消费者

// create the consumer
consumer, err := connection.NewConsumer(context.Background(), "some-queue", nil)
// receive the message
deliveryContext, err := consumer.Receive(context.Background())
// ....
deliveryContext.Accept(context.Background()) // settle the message

一旦应用程序完成消息处理，它必须结算它。这向 Broker 指示处理结果以及应该如何处理消息（例如，删除消息）。应用程序必须结算消息，否则它们的 信用 将耗尽，Broker 将停止向它们分发消息。

下一节介绍消息结算的语义。

消息处理结果（结果）

库允许应用程序以不同的方式结算消息。它们在消息应用程序的上下文中使用了尽可能明确的术语。每个术语都映射到 AMQP 规范中的给定结果。

• accept：应用程序成功处理了消息，可以从队列中删除它（accepted 结果）
• discard：应用程序无法处理消息，因为它无效，Broker 可以丢弃它或在配置的情况下死信它（rejected 结果）
• requeue：应用程序未处理消息，Broker 可以重新排队并将其传递给相同或不同的消费者（released 结果）

discard 和 requeue 具有可选的消息注释参数，可以与消息标头部分中已有的注释参数组合使用。此类消息注释可用于提供有关 discard 或 requeue 原因的详细信息。应用程序特定的注释键必须以 x-opt- 前缀开头，而 Broker 理解的注释键仅以 x- 开头。discard 和 requeue 都使用带有消息注释参数的 modified 结果。

只有仲裁队列支持使用 modified 结果修改消息注释。

消费者优雅关闭

消费者通过接受、丢弃或重新排队消息来结算消息。

当消费者关闭时，未结算的消息将被重新排队。这可能导致消息的重复处理。

这是一个例子

• 消费者为给定消息执行数据库操作。
• 消费者在接受（结算）消息之前被关闭。
• 消息被重新排队。
• 另一个消费者获取消息并再次执行数据库操作。

完全避免重复消息很困难，这就是为什么处理应该是幂等的。当消费者关闭时，消费者 API 提供了一种避免重复消息的方法。它包括暂停消息的传递，获取未结算消息的数量以确保它在某个时候达到 0，然后关闭消费者。这确保了消费者最终静止，并且所有接收到的消息都已处理。

这是一个消费者优雅关闭的示例

Java

优雅地关闭消费者

// pause the delivery of messages
consumer.pause();
// ensure the number of unsettled messages reaches 0
long unsettledMessageCount = consumer.unsettledMessageCount();
// close the consumer
consumer.close();

C#

优雅地关闭消费者

// pause the delivery of messages
consumer.pause();
// ensure the number of unsettled messages reaches 0
long unsettledMessageCount = consumer.UnsettledMessageCount();
// close the consumer
consumer.close();

Python

优雅地关闭消费者

# CURRENTLY NOT IMPLEMENTED

Go

优雅地关闭消费者

// CURRENTLY NOT IMPLEMENTED

应用程序仍然可以在不暂停消费者的情况下关闭消费者，但存在多次处理同一消息的风险。

对流的支持

库对消费者配置中的流提供开箱即用的支持。

可以设置在从流消费时附加到哪里

Java

附加到流的开头

Consumer consumer = connection.consumerBuilder()
    .queue("some-stream")
    .stream()
        .offset(ConsumerBuilder.StreamOffsetSpecification.FIRST)
    .builder()
    .messageHandler((context, message) -> {
        // message processing
    })
    .build(); 

C#

附加到流的开头

IConsumer consumer = await connection.ConsumerBuilder()
    .Queue("some-stream")
    .Stream()
    .Offset(StreamOffsetSpecification.First)
    .Builder()
    .MessageHandler( async (context, message) => {
            // message processing
    })
    .BuildAndStartAsync();

Python

附加到流的开头

class MyMessageHandler(AMQPMessagingHandler):

    def __init__(self):
        super().__init__()

    def on_message(self, event: Event):
        # accepting
        self.delivery_context.accept(event)
        # deal with the message

stream_address = AddressHelper.queue_address("some-stream")
consumer = consumer_connection.consumer(
    stream_address,
    message_handler=MyMessageHandler(),
    # can be first, last, next or an offset long
    # you can also specify stream filters with methods: apply_filters and filter_match_unfiltered
    stream_filter_options=StreamOptions(offset_specification=OffsetSpecification.first),
)

Go

附加到流的开头

consumer, err := connection.NewConsumer(context.Background(), qName, 
    &StreamConsumerOptions{
            Offset: &OffsetFirst{},
        })

还支持流过滤配置

Java

配置流过滤

Consumer consumer = connection.consumerBuilder()
    .queue("some-stream")
    .stream() 
        .filterValues("invoices", "orders") 
        .filterMatchUnfiltered(true) 
    .builder() 
    .messageHandler((context, message) -> {
        // message processing
    })
    .build();

C#

配置流过滤

IConsumer consumer = await connection.ConsumerBuilder()
    .Queue("some-stream")
    .Stream()
    .FilterValues(["invoices", "order"]) 
    .FilterMatchUnfiltered(true) 
    .Builder()
    .MessageHandler(async (context, message) => {
            // message processing
        }
).BuildAndStartAsync();

Python

class MyMessageHandler(AMQPMessagingHandler):

    def __init__(self):
        super().__init__()

    def on_message(self, event: Event):
        # accepting
        self.delivery_context.accept(event)
        # deal with the message

stream_address = AddressHelper.queue_address("some-stream")
consumer = consumer_connection.consumer(
    stream_address,
    message_handler=MyMessageHandler(),
    stream_filter_options=StreamOptions(stream_filters=["invoices", "order"], match_unfiltered=True),
)

Go

配置流过滤

consumer, err := connection.NewConsumer(context.Background(), qName, &
        StreamConsumerOptions{
            Offset:           &OffsetFirst{},
            Filters:          []string{"invoices", "order"},
        })

在处理流时，还可以考虑将原生流协议与您首选编程语言的流客户端库一起使用。

拓扑管理

应用程序可以管理 RabbitMQ 的 AMQ 0.9.1 模型：声明和删除交换机、队列和绑定。

为此，他们需要从连接中获取管理 API

Java

从环境中获取管理对象

Management management = connection.management();
// ...
// close the management instance when it is no longer needed
management.close();

C#

从环境中获取管理对象

IManagement management = connection.Management();
// ...
// close the management instance when it is no longer needed
await management.CloseAsync()

Python

从环境中获取管理对象

management = connection.management()
 # ...
 # close the management instance when it is no longer needed
management.close()

Go

从连接中获取管理对象

management = connection.management()
  // ...
  // close the management instance when it is no longer needed
management.close()

一旦不再需要管理 API，就应立即关闭它。应用程序通常在启动时创建它需要的拓扑，因此管理对象可以在此步骤之后关闭。

交换机

以下是如何创建内置类型的交换机

Java

创建内置类型的交换机

management.exchange()
    .name("my-exchange")
    .type(Management.ExchangeType.FANOUT) // enum for built-in type
    .declare();

C#

创建内置类型的交换机

IExchangeSpecification exchangeSpec = management
        .Exchange(exchangeName)
        .Type(ExchangeType.TOPIC);
await exchangeSpec.DeclareAsync();

Python

创建内置类型的交换机

exchange_name = "my-exchange"
management.declare_exchange(ExchangeSpecification(name=exchange_name, exchange_type=ExchangeType.topic))

Go

创建内置类型的交换机

// there are structs for Fanout, Direct, etc..
exchange_name = "my-exchange"
exchangeInfo, err := management.DeclareExchange(context.TODO(), &TopicExchangeSpecification{
            Name: exchangeName,
        })

也可以将交换机类型指定为字符串（对于非内置类型交换机）

Java

创建非内置类型的交换机

management.exchange()
    .name("my-exchange")
    .type("x-delayed-message") // non-built-in type
    .autoDelete(false)
    .argument("x-delayed-type", "direct")
    .declare();

C#

创建非内置类型的交换机

await _management.Exchange("myExchange")
    .Type("x-delayed-message")
    .Argument("x-delayed-type", "direct")
    .DeclareAsync();

Python

创建非内置类型的交换机

    exchange_arguments = {}
    exchange_arguments["x-delayed-type"] = "direct"

    exchange_info = management.declare_exchange(
        ExchangeCustomSpecification(
            name="myExchange",
            exchange_type="x-delayed-message",
            arguments=exchange_arguments,
        )
    )

Go

创建非内置类型的交换机

_, err := management.DeclareExchange(context.TODO(), &CustomExchangeSpecification{
        Name:             "myExchange",
        ExchangeTypeName: "x-delayed-message",
            Arguments: map[string]any{
                "x-delayed-type": "direct",
            },
        })

以下是如何删除交换机

Java

删除交换机

management.exchangeDelete("my-exchange");

C#

删除交换机

await management.Exchange("my-exchange").DeleteAsync();

Python

删除交换机

exchange_name = "my-exchange"
management.delete_exchange(exchange_name)

Go

创建内置类型的交换机

exchange_name = "my-exchange"
management.DeleteExchange(context.TODO(),exchange_name)

队列

以下是如何使用默认队列类型创建队列

Java

创建经典队列

management.queue()
    .name("my-queue")
    .exclusive(true)
    .autoDelete(false)
    .declare();

C#

创建经典队列

IQueueSpecification queueSpec = management
    .Queue("myqueue")
    .Exclusive(true)
    .AutoDelete(false)
await queueSpec.DeclareAsync();

Python

创建经典队列

queue_name = "myqueue"
management.declare_queue(ClassicQueueSpecification(name=queue_name))

Go

创建经典队列

queue_name = "myqueue"
queueInfo, err := management.DeclareQueue(context.TODO(), &ClassicQueueSpecification{
            Name: queueName})

管理 API 显式支持队列参数

Java

创建带有参数的队列

management.queue()
    .name("my-queue")
    .type(Management.QueueType.CLASSIC)
    .messageTtl(Duration.ofMinutes(10))
    .maxLengthBytes(ByteCapacity.MB(100))
    .declare();

C#

创建带有参数的队列

IQueueSpecification queueSpec = management
    .Queue("my-queue")
    .Type(QueueType.CLASSIC)
    .MessageTtl(TimeSpan.FromMinutes(10))
    .MaxLengthBytes(ByteCapacity.Mb(100));
await queueSpec.DeclareAsync();

Python

创建带有参数的队列

management.declare_queue(ClassicQueueSpecification(name="my-queue", message_ttl=timedelta(minutes=10), max_len_bytes=100000000))

Go

创建带有参数的队列

queueInfo, err := management.DeclareQueue(context.TODO(), &ClassicQueueSpecification{
            Name:           "my-queue",
            MaxPriority:    32,
            MaxLengthBytes: CapacityGB(1),
            IsAutoDelete:   false,
    })

管理 API 还区分了所有队列类型共享的参数和仅对给定类型有效的参数。以下是创建仲裁队列的示例

Java

创建仲裁队列

management
    .queue()
    .name("my-quorum-queue")
    .quorum() // set queue type to 'quorum'
        .quorumInitialGroupSize(3) // specific to quorum queues
        .deliveryLimit(3) // specific to quorum queues
    .queue()
    .declare();

C#

创建仲裁队列

IQueueSpecification queueSpec = management
    .Queue("my-quorum-queue")
    .Quorum() // set queue type to 'quorum'
        .QuorumInitialGroupSize(3) // specific to quorum queues
        .DeliveryLimit(3) // specific to quorum queues
    .Queue();
await queueSpec.DeclareAsync();

Python

创建仲裁队列

queue_name = "my-quorum-queue"
management.declare_queue(QuorumQueueSpecification(name=queue_name, quorum_initial_group_size=3, deliver_limit=3))

Go

创建仲裁队列

queueInfo, err := management.DeclareQueue(context.TODO(), &QuorumQueueSpecification{
            Name: "my-quorum-queue",
            QuorumInitialGroupSize = 3,
            DeliveryLimit: 3,
        })

可以查询有关队列的信息

Java

获取队列信息

Management.QueueInfo info = management.queueInfo("my-queue");
long messageCount = info.messageCount();
int consumerCount = info.consumerCount();
String leaderNode = info.leader();

C#

获取队列信息

IQueueInfo queueInfo = await management.GetQueueInfoAsync("my-queue");
ulong messageCount = queueInfo.MessageCount();
uint consumerCount = queueInfo.ConsumerCount();
string leader = queueInfo.Leader();

Python

获取队列信息

queue_name = "my-queue"
queue_info = management.queue_info(name=queue_name)
message_count = queue_info.message_count
consumer_count = queue_info.consumer_count
leader = queue_info.leader

Go

获取队列信息

queueInfo, err := management.QueueInfo(context.TODO(), "my-queue")

messageCount := queueInfo.MessageCount();
consumerCount := queueInfo.ConsumerCount();
leader := queueInfo.Leader();

此 API 也可用于检查队列是否存在。

以下是如何删除队列

Java

删除队列

management.queueDelete("my-queue");

C#

删除队列

await management.Queue("myqueue").DeleteAsync();

Python

删除队列

management.delete_queue(name="myqueue")

Go

删除队列

management.DeleteExchange(context.TODO(),"myqueue")

绑定

管理 API 支持将队列绑定到交换机

Java

将队列绑定到交换机

management.binding()
    .sourceExchange("my-exchange")
    .destinationQueue("my-queue")
    .key("foo")
    .bind();

C#

将队列绑定到交换机

IBindingSpecification bindingSpec = management.Binding()
    .SourceExchange("my-exchange")
    .DestinationQueue("my-queue")
    .Key("foo");
await bindingSpec.BindAsync();

Python

将队列绑定到交换机

bind_name = management.bind(
    BindingSpecification(
        source_exchange="my-exchange",
        destination_queue="my-queue",
        binding_key="foo",
    )
)

Go

将队列绑定到交换机

// ExchangeToQueueBindingSpecification implements BindingSpecification interface 
bindingPath, err := management.Bind(context.TODO(), &rmq.ExchangeToQueueBindingSpecification{
        SourceExchange:   "my-exchange",
        DestinationQueue: "my-queue",
        BindingKey:       "foo",
    })

还支持交换机到交换机的绑定

Java

将交换机绑定到另一个交换机

management.binding()
    .sourceExchange("my-exchange")
    .destinationExchange("my-other-exchange")
    .key("foo")
    .bind();

C#

将交换机绑定到另一个交换机

IBindingSpecification bindingSpec = management.Binding()
    .SourceExchange("my-exchange")
    .DestinationExchange("my-other-exchange")
    .Key("foo");
await bindingSpec.BindAsync();

Python

将交换机绑定到另一个交换机

    
binding_exchange_queue_path = management.bind(
    ExchangeToExchangeBindingSpecification(
        source_exchange="my-exchange",
        destination_exchange="my-other-exchange",
     )
)

Go

绑定一个交换机到另一个交换机

// ExchangeToExchangeBindingSpecification implements BindingSpecification interface 
bindingPath, err := management.Bind(context.TODO(), &rmq.ExchangeToExchangeBindingSpecification{
        SourceExchange:   "my-exchange",
        DestinationExchange: "my-other-exchange",
    })

也可以取消绑定实体

Java

删除交换机和队列之间的绑定

management.unbind()
    .sourceExchange("my-exchange")
    .destinationQueue("my-queue")
    .key("foo")
    .unbind();

C#

删除交换机和队列之间的绑定

IBindingSpecification bindingSpec = management.Binding()
    .SourceExchange("my-exchange")
    .DestinationQueue("my-queue")
    .Key("foo");
await bindingSpec.UnbindAsync();

Python

删除交换机和队列之间的绑定

management.unbind(BindingSpecification(
    source_exchange="my-exchange",
    destination_queue="my-queue",
))

Go

删除交换机和队列之间的绑定

// bindingPath is the bind result
err = management.Unbind(context.TODO(), bindingPath)

高级用法

生命周期监听器

应用程序可以通过添加监听器来响应某些 API 组件的状态更改。应用程序可以向连接添加监听器，以便在连接在连接后恢复时停止发布消息。然后，应用程序可以在连接恢复并再次打开时恢复发布。

以下是如何在连接上设置监听器

Java

在连接上设置监听器

Connection connection = environment.connectionBuilder()
    .listeners(context -> { // set one or several listeners
        context.previousState(); // the previous state
        context.currentState(); // the current (new) state
        context.failureCause(); // the cause of the failure (in case of failure)
        context.resource(); // the connection
    }).build();

C#

将事件附加到 ChangeState

connection.ChangeState += ( 
     sender, // the sender instance ( in this case the connection)
     fromState, // the previous state
     toState, // the current (new) state
     e // the cause of the failure (in case of failure)
     ) =>
{
  
};

Python

将事件附加到 ChangeState

# CURRENTLY NOT IMPLEMENTED 

Go

将通道注册到 StateChanged

stateChanged := make(chan *rmq.StateChanged, 1)
go func(ch chan *rmq.StateChanged) {
        for statusChanged := range ch {
            // statusChanged.From from status
            // statusChanged.To to status 
            // StateClosed has the func GetError() in case of error 

        }
    }(stateChanged)

connection.NotifyStatusChange(stateChanged)

也可以在发布者实例上设置监听器

Java

在发布者上设置监听器

Publisher publisher = connection.publisherBuilder()
    .listeners(context -> {
        // ...
    })
    .exchange("foo").key("bar")
    .build();

C#

将事件附加到 ChangeState

publisher.ChangeState += (sender, fromState, toState, e) =>
{
}

Python

将事件附加到 ChangeState

# CURRENTLY NOT IMPLEMENTED FOR PYTHON

Go

# CURRENTLY NOT IMPLEMENTED Due Of https://github.com/Azure/go-amqp/issues/99

以及在消费者实例上

Java

在消费者上设置监听器

Consumer consumer = connection.consumerBuilder()
    .listeners(context -> {
        // ...
    })
    .queue("my-queue")
    .build();

C#

将事件附加到 ChangeState

consumer.ChangeState += (sender, fromState, toState, e) =>
{
}

Python

将事件附加到 ChangeState

# CURRENTLY NOT IMPLEMENTED FOR PYTHON

Go

# CURRENTLY NOT IMPLEMENTED Due Of https://github.com/Azure/go-amqp/issues/99

自动连接恢复

默认情况下激活自动连接恢复：客户端库将在意外关闭后自动恢复连接（例如，网络故障、节点重启等）。一旦连接恢复，自动拓扑恢复也会激活：客户端库将为恢复的连接重新创建 AMQP 实体以及发布者和消费者。开发人员不必太担心网络稳定性和节点重启，因为客户端库会处理它。

客户端每 5 秒尝试重新连接，直到成功为止。可以通过自定义退避延迟策略来更改此行为

Java

为连接恢复设置退避策略

Connection connection = environment.connectionBuilder()
    .recovery()
    .backOffDelayPolicy(BackOffDelayPolicy.fixed(Duration.ofSeconds(2)))
    .connectionBuilder().build();

C#

为连接恢复设置退避策略

class MyBackOffDelay : IBackOffDelayPolicy {
    ...
}

await AmqpConnection.CreateAsync(
    ConnectionSettingBuilder.Create().RecoveryConfiguration(
    RecoveryConfiguration.Create()
    .BackOffDelayPolicy(new MyBackOffDelay())).Build());

Python

    environment = Environment(
        "amqp://guest:guest@localhost:5672/",
        recovery_configuration=RecoveryConfiguration(back_off_reconnect_interval=timedelta(seconds=2)),
    )

Go

    // to the BackOffReconnectInterval the client adds a random 500 ms 
    env := NewEnvironment("amqp://", &AmqpConnOptions{
            RecoveryConfiguration: &RecoveryConfiguration{
                ActiveRecovery:           true,
                BackOffReconnectInterval: 5 * time.Second,
                MaxReconnectAttempts:     5,
            },
        })

如果拓扑恢复不适用于给定的应用程序，也可以停用拓扑恢复。应用程序通常会注册连接生命周期监听器，以了解连接何时恢复并相应地恢复其自身状态。

Java

停用拓扑恢复

Connection connection = environment.connectionBuilder()
    .recovery()
    .topology(false) // deactivate topology recovery
    .connectionBuilder()
    .listeners(context -> {
        // set listener that restores application state when connection is recovered
    })
    .build();

C#

停用拓扑恢复

await AmqpConnection.CreateAsync(
    ConnectionSettingBuilder.Create().RecoveryConfiguration(
    RecoveryConfiguration.Create()
    .Topology(false)) // deactivate topology recovery
    .Build());

Python

     # CURRENTLY NOT IMPLEMENTED

Go

     // CURRENTLY NOT IMPLEMENTED

也可以完全停用恢复

Java

停用恢复

Connection connection = environment.connectionBuilder()
    .recovery()
    .activated(false) // deactivate recovery
    .connectionBuilder().build();

C#

停用恢复

await AmqpConnection.CreateAsync(
        ConnectionSettingBuilder.Create().RecoveryConfiguration(
        RecoveryConfiguration.Create().
        Activated(false)).// deactivate recovery
        Build());

Python

    environment = Environment(
        "amqp://guest:guest@localhost:5672/",
        recovery_configuration=RecoveryConfiguration(active_recovery=False),
    )

Go

停用恢复

     env := NewEnvironment("amqp://", &AmqpConnOptions{
            RecoveryConfiguration: &RecoveryConfiguration{
                ActiveRecovery:           false,
            },
        })