Vert.x RabbitMQ 客户端 (vertx-rabbitmq-client)

Vert.x RabbitMQ 客户端允许应用程序直接与 RabbitMQ 消息代理（AMQP 0.9.1 规范）进行全异步、非阻塞式交互。

快速上手

Maven 依赖引入

在 pom.xml 中添加以下依赖：

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-rabbitmq-client</artifactId>
  <version>5.0.12</version>
</dependency>

Gradle 依赖引入

在 build.gradle 中添加依赖：

dependencies {
  compile 'io.vertx:vertx-rabbitmq-client:5.0.12'
}

创建客户端

你可以通过完整的 AMQP URI 初始化客户端实例：

RabbitMQOptions config = new RabbitMQOptions();
config.setUri("amqp://xvjvsrrc:secret@localhost:5672/xvjvsrrc");
RabbitMQClient client = RabbitMQClient.create(vertx, config);

// 启动连接
client
  .start()
  .onComplete(asyncResult -> {
  if (asyncResult.succeeded()) {
    System.out.println("RabbitMQ 连接成功！");
  } else {
    System.out.println("连接 RabbitMQ 失败: " + asyncResult.cause().getMessage());
  }
});

也可以通过设置各离散属性进行精细化配置：

RabbitMQOptions config = new RabbitMQOptions();
config.setUser("user1");
config.setPassword("password1");
config.setHost("localhost");
config.setPort(5672);
config.setVirtualHost("vhost1");
config.setConnectionTimeout(6000); // 毫秒
config.setRequestedHeartbeat(60);  // 秒
config.setHandshakeTimeout(6000);  // 毫秒
config.setRequestedChannelMax(5);
config.setNetworkRecoveryInterval(500); // 毫秒
config.setAutomaticRecoveryEnabled(true);

RabbitMQClient client = RabbitMQClient.create(vertx, config);

支持配置多个节点地址以实现集群高可用故障转移：

config.setAddresses(Arrays.asList(Address.parseAddresses("firstHost,secondHost:5672")));

连接重连与拓扑恢复机制

RabbitMQClient 提供了两种机制应对断线重连：

官方原生 Java RabbitMQ 客户端驱动的 Auto Recovery 机制
Vert.x RabbitMQClient 自身基于底层 Socket 整体重启的 Reconnect 机制

若需使用官方驱动的 Auto Recovery，需将其启用并禁用 Vert.x 重试：

RabbitMQOptions options = new RabbitMQOptions();
options.setAutomaticRecoveryEnabled(true);
options.setReconnectAttempts(0);

或者使用 Vert.x 的连接重试策略，在连接故障时不断从头尝试重新建连：

RabbitMQOptions options = new RabbitMQOptions();
options.setAutomaticRecoveryEnabled(false);
options.setReconnectAttempts(Integer.MAX_VALUE);
options.setReconnectInterval(500);

为防范在连接重连后发生“消息已发送但服务器端交换机与队列尚未声明绑定”的竞态条件，可以通过 ConnectionEstablishedCallback 注册拓扑初始化钩子：

RabbitMQClient client = RabbitMQClient.create(vertx, config);
client.addConnectionEstablishedCallback(promise -> {
  client.exchangeDeclare("exchange", "fanout", true, false)
    .compose(v -> client.queueDeclare("queue", false, true, true))
    .compose(declareOk -> client.queueBind(declareOk.getQueue(), "exchange", ""))
    .onComplete(promise);
});

开启 SSL/TLS 安全加密

RabbitMQOptions options = new RabbitMQOptions()
  .setSsl(true);

若启用 trustAll(true)，客户端将信任所有服务端证书，但在生产环境中易受中间人攻击，请勿在生产中开启。生产环境推荐配置 JKS、PKCS12 或 PEM 证书信任库：

RabbitMQOptions options = new RabbitMQOptions()
  .setSsl(true)
  .setPemTrustOptions(new PemTrustOptions().addCertPath("/path/ca-cert.pem"));

核心消息操作

消息发布 (BasicPublish)

Buffer message = Buffer.buffer("Hello RabbitMQ, from Vert.x !");
client
  .basicPublish("", "my.queue", message)
  .onComplete(pubResult -> {
    if (pubResult.succeeded()) {
      System.out.println("消息发布成功！");
    } else {
      pubResult.cause().printStackTrace();
    }
  });

带确认的消息发布 (Confirm Mode)

开启 Confirm 模式并等待 Broker 确认接收：

Buffer message = Buffer.buffer("Hello RabbitMQ, from Vert.x !");

// 开启确认模式（仅需在初始化时执行一次）
client
  .confirmSelect()
  .compose(v -> client.basicPublish("", "my.queue", message))
  .compose(v -> client.waitForConfirms()) // 等待 Broker 发回确认
  .onComplete(waitResult -> {
    if (waitResult.succeeded()) {
      System.out.println("消息已安全送达 Broker！");
    } else {
      waitResult.cause().printStackTrace();
    }
  });

高吞吐高可靠消息发布器 (RabbitMQPublisher)

为了在兼顾极高吞吐的同时确保零丢失，RabbitMQ 提供了异步批量确认机制。RabbitMQPublisher 封装了标准的内部重试缓冲与批处理流水线：

RabbitMQPublisher publisher = RabbitMQPublisher.create(vertx, client, options);

messages.forEach((k, v) -> {
  AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
    .messageId(k)
    .build();
  publisher.publish("exchange", "routingKey", properties, v.toBuffer());
});

// 监听异步消息确认流
publisher.getConfirmationStream().handler(conf -> {
  if (conf.isSucceeded()) {
    messages.remove(conf.getMessageId()); // 成功确认，移出待确认缓存
  }
});

消息消费 (BasicConsumer)

将队列绑定为响应式流进行流式消费：

client
  .basicConsumer("my.queue")
  .onComplete(res -> {
    if (res.succeeded()) {
      RabbitMQConsumer mqConsumer = res.result();
      mqConsumer.handler(message -> {
        System.out.println("收到消息正文: " + message.body().toString());
      });
    }
  });

消费流支持随时调用 pause() 和 resume() 进行背压流量控制。

手动确认消费 (Manual Ack)

client
  .basicConsumer("my.queue", new QueueOptions().setAutoAck(false))
  .onComplete(consumeResult -> {
    if (consumeResult.succeeded()) {
      RabbitMQConsumer consumer = consumeResult.result();
      consumer.handler(msg -> {
        JsonObject json = (JsonObject) msg.body();
        // 业务处理完毕后手动回传 ack 确认：
        client.basicAck(json.getLong("deliveryTag"), false);
      });
    }
  });

AMQP

MQTT