Vert.x AMQP 客户端 (vertx-amqp-client)

Vert.x AMQP 客户端允许 Vert.x 应用程序直接与支持 AMQP 1.0 标准的消息代理（Broker）与路由中间件进行交互。其核心能力包括：

连接至 AMQP 1.0 消息代理或路由器（全面支持 SASL 身份认证与 TLS 安全连接）
从队列（Queue）或主题（Topic）中消费消息
向队列或主题发送消息
监听与确认发送消息的接收回执（Acknowledgement）

AMQP 1.0 协议原生支持持久化订阅、会话状态保持、强安全机制及精细化消息路由。更多协议细节请参阅 AMQP 官方主页。

Vert.x AMQP 客户端底层基于 Vert.x Proton 构建。若业务需要对底层 AMQP 传输执行极度精细的微调控制，推荐直接使用 Vert.x Proton。

引入依赖

要在项目中使用 Vert.x AMQP 客户端，请在构建描述文件中添加以下依赖：

Maven（在 pom.xml 中）：

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-amqp-client</artifactId>
  <version>5.1.10</version>
</dependency>

Gradle（在 build.gradle 文件中）：

compile 'io.vertx:vertx-amqp-client:5.1.10'

创建 AMQP 客户端实例

在将依赖引入类路径后，可以按如下方式初始化 AmqpClient：

AmqpClientOptions options = new AmqpClientOptions()
  .setHost("localhost")
  .setPort(5672)
  .setUsername("user")
  .setPassword("secret");

// 使用客户端内部独立的 Vert.x 实例进行创建
AmqpClient client1 = AmqpClient.create(options);

// 传入外部显式的 Vert.x 实例（推荐在 Vert.x 应用中采用此方式）
AmqpClient client2 = AmqpClient.create(vertx, options);

初始化时支持传入显式的 Vert.x 实例；若省略，客户端内部会自动创建一套专有 Vert.x 实例并在客户端关闭时一并销毁。

AmqpClientOptions 用于配置远程服务器地址、认证凭据等参数，亦可直接用于定制底层 Proton 驱动配置。

此外，主机、端口、用户名与密码还支持通过操作系统环境变量或 JVM 系统属性无侵入注入：

主机（Host）：系统属性 amqp-client-host，环境变量 AMQP_CLIENT_HOST（必填）
端口（Port）：系统属性 amqp-client-port，环境变量 AMQP_CLIENT_PORT（默认 5672）
用户名：系统属性 amqp-client-username，环境变量 AMQP_CLIENT_USERNAME
密码：系统属性 amqp-client-password，环境变量 AMQP_CLIENT_PASSWORD

建立网络连接

客户端创建后，需显式调用 connect 方法连接至远程服务器：

client
  .connect()
  .onComplete(ar -> {
  if (ar.failed()) {
    System.out.println("无法连接至 AMQP 代理服务器: " + ar.cause().getMessage());
  } else {
    System.out.println("成功建立 AMQP 连接！");
    AmqpConnection connection = ar.result();
  }
});

连接成功后返回的 AmqpConnection 对象是后续创建消息接收器（Receiver）与发送器（Sender）的核心句柄。

创建消息接收器 (Receiver)

接收器专门用于消费队列中的消息，可以通过以下两种方式获取：

connection
  .createReceiver("my-queue")
  .onComplete(done -> {
    if (done.failed()) {
      System.out.println("无法创建接收器");
    } else {
      AmqpReceiver receiver = done.result();
      receiver.handler(msg -> {
        // 每当接收到消息时触发回调
        System.out.println("收到消息: " + msg.bodyAsString());
      });
    }
  });

connection
  .createReceiver("my-queue")
  .onComplete(done -> {
    if (done.failed()) {
      System.out.println("无法创建接收器");
    } else {
      AmqpReceiver receiver = done.result();
      receiver
        .exceptionHandler(t -> {
          // 异常处理逻辑
        })
        .handler(msg -> {
          // 挂载消息监听处理器
        });
    }
  });

接收器实现了 Vert.x 响应式读流（ReadStream）规范，支持使用 pause 与 resume 动态暂停/恢复接收；背压流控机制由底层通过 AMQP Credits 信用额度全自动协同控制。

接收到的消息为不可变的 AmqpMessage 实例，包含了 AMQP 规范中定义的所有元数据。

默认情况下消息由客户端自动应答（Auto-Ack）。你可以通过 setAutoAcknowledgement(false) 禁用自动应答，并使用以下方法显式应答：

accepted()：确认接收并消费成功
rejected()：拒绝并丢弃消息
released()：释放消息以便重新派发给其他消费者

创建消息发送器 (Sender)

发送器用于向队列或主题发布消息：

connection
  .createSender("my-queue")
  .onComplete(done -> {
  if (done.failed()) {
    System.out.println("创建发送器失败");
  } else {
    AmqpSender sender = done.result();
  }
});

构建消息使用 AmqpMessageBuilder：

AmqpMessageBuilder builder = AmqpMessage.create();

// 简单的纯文本消息
AmqpMessage m1 = builder.withBody("hello").build();

// 覆盖默认目标地址的消息
AmqpMessage m2 = builder.withBody("hello").address("another-queue").build();

// 包含 JSON 载荷、TTL 过期时间与自定义属性的丰富消息
AmqpMessage m3 = builder
  .withJsonObjectAsBody(new JsonObject().put("message", "hello"))
  .subject("subject")
  .ttl(10000)
  .applicationProperties(new JsonObject().put("prop1", "value1"))
  .build();

发送消息支持即发即弃（Send）与带确认回执（SendWithAck）：

// 即发即弃方式
sender.send(AmqpMessage.create().withBody("hello").build());

// 监听服务端确认回执
sender
  .sendWithAck(AmqpMessage.create().withBody("hello").build())
  .onComplete(acked -> {
  if (acked.succeeded()) {
    System.out.println("消息已被远程 Broker 确认接收！");
  } else {
    System.out.println("消息未被确认: " + acked.cause().getMessage());
  }
});

请求-响应模式实现 (Request-Reply)

可以通过动态接收器（Dynamic Receiver）配合匿名发送器（Anonymous Sender）实现经典的 RPC 请求-响应语义。动态接收器的响应地址由 Broker 临时分配生成，并通过 replyTo 传给对端：

// 发送请求并等待应答端：
connection
  .createDynamicReceiver()
  .onComplete(replyReceiver -> {
  // Broker 动态生成的私有应答队列地址：
  String replyToAddress = replyReceiver.result().address();

  // 监听应答回包：
  replyReceiver.result().handler(msg -> {
    System.out.println("收到应答: " + msg.bodyAsString());
  });

  // 发送请求消息并携带 replyTo 与 correlationId：
  connection
    .createSender("my-queue")
    .onComplete(sender -> {
    sender.result().send(AmqpMessage.create()
      .replyTo(replyToAddress)
      .id("my-message-id")
      .withBody("这是请求正文").build());
  });
});

关闭与资源释放

在业务处理完毕后，应显式调用 close 释放资源。关闭 AmqpConnection 会级联关闭其下创建的所有 Receiver 和 Sender；关闭 AmqpClient 则会关闭全部活跃的网络连接。

Kafka

RabbitMQ