使用HiveMQ Async Client无法连接本地MQTT Broker求助

问题背景 #

我编写了publish方法用于连接MQTT Broker、发送消息后断开连接，代码如下：

def publish(mqttCfg: MqttConfig, topic: String, mqttQos: MqttQos): Future[Unit] = {
    val client = asyncMqttClient(mqttCfg)

    // 定义发布结果的包装类型
    sealed trait PublishResult
    case class SuccessfulPublish(mqttPublishResult: Mqtt5PublishResult) extends PublishResult
    case class FailedPublish(error: Throwable) extends PublishResult

    asyncMqttClient(mqttCfg).connect()
      .thenCompose(_ => client.publishWith().topic(topic).qos(mqttQos).payload("HELLO WORLD!".getBytes()).send())
      .thenAccept(result => {
        val publishResult = Try(result)
        publishResult match {
          case Success(message) =>
            println(s"publishedResult: $message") // TODO: 替换为日志
          case Failure(error) =>
            println(s"Failed to publish: ${error.getMessage}") // TODO: 替换为日志
        }
      })
      .thenCompose(_ => client.disconnect())
      .thenAccept(_ => println("disconnected"))
      .asScala.map(_ => ())
  }

配套的Scala测试用例：

"MqttClientFactory#publish" should "connect to a local MQTT broker and publish" in {
    val mqttConfig = MqttConfig("cpo-platform-test", "test.mosquitto.org", 1883)
    val published = MqttClientFactory.publish(
      mqttConfig,
      "cpo-test-topic",
      MqttQos.EXACTLY_ONCE
    )
    whenReady(published, timeout(Span(100, Seconds))) { _ => {
      val client = MqttClientFactory.asyncMqttClient(mqttConfig)
      println("In here ****************** ")
      client
        .connect()
        .thenCompose(_ => client.subscribeWith().topicFilter("cpo-test-topic").qos(MqttQos.EXACTLY_ONCE).callback(println).send())
      }
    }
  }

运行测试时，whenReady等待Future完成阶段抛出错误：

The future returned an exception of type: java.util.concurrent.CompletionException, with message: com.hivemq.client.mqtt.exceptions.MqttClientStateException: MQTT client is not connected..
ScalaTestFailureLocation: com.openelectrons.cpo.mqtt.MqttClientFactoryTest at (MqttClientFactoryTest.scala:29)

测试过本地多个MQTT Broker（Eclipse Mosquitto、Cedalo Broker），均出现相同错误。

补充信息 #

asyncMqttClient方法实现 #

def asyncMqttClient(mqttCfg: MqttConfig): Mqtt5AsyncClient = {
    Mqtt5Client.builder()
      .identifier(mqttCfg.appName)
      .serverHost(mqttCfg.serverHost)
      .serverPort(mqttCfg.serverPort)
      .simpleAuth()
        .username(mqttCfg.user.getOrElse(""))
        .password(mqttCfg.pass.getOrElse("").getBytes("UTF-8"))
      .applySimpleAuth()
      .buildAsync()
  }

本地Mosquitto启动配置（Docker Compose） #

version: "3.7"
services:
  mqtt5:
    image: eclipse-mosquitto
    container_name: mqtt5
    ports:
      - 1883:1883 # 默认MQTT端口
      - 9001:9001 # Websocket端口
    volumes:
      - /opt/softwares/mosquitto/mqtt5/config:/mosquitto/config

mosquitto.conf配置 #

listener 1883
allow_anonymous true
persistence true
persistence_location /mosquitto/data/
log_dest file /mosquitto/log/mosquitto.log

Broker日志输出 #

1696310903: New connection from 192.168.208.1:57752 on port 1883.
1696310903: New client connected from 192.168.208.1:57752 as cpo-platform-test (p5, c1, k60).
1696310903: Client cpo-platform-test closed its connection.

已通过mosquitto_pub和mosquitto_sub验证Broker正常工作，但测试代码运行时连接后立即断开，请求排查问题。

问题排查与解决方案 #

1. 核心错误：创建了两个独立的MQTT客户端实例 #

在publish方法中，你先初始化了val client = asyncMqttClient(mqttCfg)，但后续调用connect()时，又通过asyncMqttClient(mqttCfg)创建了另一个全新的客户端实例。这导致：

• 第一个客户端实例执行了connect()，但没有被用来发送消息
• 第二个client实例从未执行connect()，直接调用publish自然会报“未连接”错误

修改publish方法，统一使用同一个客户端实例：

def publish(mqttCfg: MqttConfig, topic: String, mqttQos: MqttQos): Future[Unit] = {
    val client = asyncMqttClient(mqttCfg)

    client.connect()
      .thenCompose(_ => client.publishWith()
        .topic(topic)
        .qos(mqttQos)
        .payload("HELLO WORLD!".getBytes())
        .send())
      .thenAccept(result => {
        println(s"publishedResult: $result") // TODO: 替换为日志组件
      })
      .exceptionally(error => {
        println(s"Failed to publish: ${error.getMessage}") // TODO: 替换为日志组件
        null // 适配CompletableFuture返回要求
      })
      .thenCompose(_ => client.disconnect())
      .thenAccept(_ => println("disconnected"))
      .asScala.map(_ => ())
  }

2. 测试用例的优化点 #

原测试用例存在两个问题：

• 回调中创建的订阅客户端与发布客户端使用相同的Client ID，MQTT Broker不允许同一ID同时存在两个连接，会直接踢掉旧连接
• 订阅逻辑没有等待完成就结束，可能导致消息接收不及时

修改后的测试示例：

"MqttClientFactory#publish" should "connect to a local MQTT broker and publish" in {
    val publishConfig = MqttConfig("cpo-platform-test", "test.mosquitto.org", 1883)
    // 使用不同的Client ID创建订阅客户端
    val subscribeConfig = publishConfig.copy(appName = "cpo-platform-test-sub")
    
    // 先等待订阅客户端连接并完成订阅
    val subscribeFuture = MqttClientFactory.asyncMqttClient(subscribeConfig)
      .connect()
      .thenCompose(_ => subscribeConfig.client.subscribeWith()
        .topicFilter("cpo-test-topic")
        .qos(MqttQos.EXACTLY_ONCE)
        .callback(msg => println(s"Received message: ${new String(msg.getPayloadAsBytes)}"))
        .send())
      .asScala
    
    whenReady(subscribeFuture, timeout(Span(10, Seconds))) { _ =>
      // 订阅就绪后执行发布操作
      val publishFuture = MqttClientFactory.publish(
        publishConfig,
        "cpo-test-topic",
        MqttQos.EXACTLY_ONCE
      )
      whenReady(publishFuture, timeout(Span(10, Seconds))) { _ =>
        println("Publish operation completed successfully")
        // 可在此添加消息接收验证逻辑
      }
    }
  }

3. 额外注意事项 #

• 由于Broker配置了allow_anonymous true，确保MqttConfig中的user和pass为None，避免不必要的认证逻辑干扰
• 原代码中定义的PublishResult类型未实际使用，可直接移除以简化代码

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