如何用新版本Vert.x实现含QManager与Channel的MQ Series队列消息收发

你之前的代码误用了Vert.x的标准MQTT客户端，而QManager、Channel是IBM MQ（MQ Series）原生协议的专属配置项，标准MQTT协议并不包含这些参数，因此新版Vert.x MQTT客户端移除了相关API。正确的做法是使用Vertx官方提供的vertx-ibm-mq客户端来对接IBM MQ原生队列。

步骤1：引入依赖 #

先在项目中添加vertx-ibm-mq的依赖（版本需与你的Vert.x核心版本保持一致）：

Maven #

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-ibm-mq</artifactId>
  <version>4.5.1</version>
</dependency>

Gradle #

implementation 'io.vertx:vertx-ibm-mq:4.5.1'

步骤2：配置连接参数 #

创建IBM MQ连接配置，直接设置QManager、Channel等核心参数：

IBMMQConnectionOptions connectionOptions = new IBMMQConnectionOptions()
  .setHost("localhost")
  .setPort(1414) // IBM MQ原生协议默认端口
  .setQueueManager("my-qmanager")
  .setChannel("my-channel")
  .setUsername("my-client-id")
  .setPassword("my-password");

步骤3：实现消息收发完整示例 #

import io.vertx.core.Vertx;
import io.vertx.ibm.client.IBMMQClient;
import io.vertx.ibm.client.IBMMQConnectionOptions;
import io.vertx.ibm.client.IBMMQMessage;
import io.vertx.ibm.client.IBMMQProducer;
import io.vertx.ibm.client.IBMMQConsumer;

public class IBMMQVertxExample {
  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();

    // 创建IBM MQ客户端实例
    IBMMQClient client = IBMMQClient.create(vertx);

    // 连接到IBM MQ
    client.connect(connectionOptions, connectResult -> {
      if (connectResult.succeeded()) {
        System.out.println("成功连接到IBM MQ");

        // --------------------------
        // 监听输入队列 "entrada"
        // --------------------------
        IBMMQConsumer consumer = client.createConsumer("entrada");
        consumer.handler(message -> {
          System.out.println("收到消息: " + message.bodyAsString());
          // 手动确认消息（保证可靠性）
          message.acknowledge();
        });
        consumer.exceptionHandler(err -> 
          System.err.println("消费者异常: " + err.getMessage())
        );

        // --------------------------
        // 发送消息到输出队列 "salida"
        // --------------------------
        IBMMQProducer producer = client.createProducer("salida");
        IBMMQMessage sendMsg = new IBMMQMessage()
          .setBody("Hola Mundo".getBytes());

        producer.send(sendMsg, sendResult -> {
          if (sendResult.succeeded()) {
            System.out.println("消息发送成功");
          } else {
            System.err.println("消息发送失败: " + sendResult.cause().getMessage());
          }
        });
        producer.exceptionHandler(err -> 
          System.err.println("生产者异常: " + err.getMessage())
        );

      } else {
        System.err.println("连接IBM MQ失败: " + connectResult.cause().getMessage());
      }
    });

    // 应用退出时关闭连接
    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
      client.close(closeResult -> {
        if (closeResult.succeeded()) {
          System.out.println("IBM MQ连接已关闭");
        }
      });
      vertx.close();
    }));
  }
}

关键注意事项 #

• 端口区分：IBM MQ原生协议默认端口是1414，不要和MQTT的1883端口混淆。
• 队列权限：确保你的账号对目标队列拥有PUT（发送）和GET（接收）权限。
• 版本兼容：vertx-ibm-mq版本必须与Vert.x核心版本一致，避免兼容性问题。
• 消息确认：如果需要保证消息不丢失，务必调用message.acknowledge()完成消息确认。

内容的提问来源于stack exchange，提问作者Victor Marrerp