使用Eclipse Paho MQTT Golang连接AWS IoT分进程运行报EOF错误

在Linux系统上使用Go v1.17.2，通过paho.mqtt.golang v1.4.1库连接AWS MQTT，代码基于EMQX示例，采用AWS IoT Core提供的证书实现TLS连接，遇到以下异常：

正常场景 #

• 同一Go程序中同时运行发布和订阅代码时，一切正常，输出如下：

2022/08/11 19:47:42 Connected
Subscribed to topic: topic_1
2022/08/12 13:47:42 Received message: Message 0 from topic: topic_1
2022/08/11 19:47:43 Received message: Message 1 from topic: topic_1
...
2022/08/11 19:47:51 Received message: Message 9 from topic: topic_1

• 仅运行订阅代码，通过AWS IoT控制台发布消息时也正常，输出如下：

2022/08/11 20:25:27 Connected
2022/08/11 20:25:27 Subscribed to topic: topic_1
2022/08/11 20:25:29 Received message: {
  "message": "Hello from AWS IoT console"
} from topic: topic_1
2022/08/11 20:25:30 Received message: {
  "message": "Hello from AWS IoT console"
} from topic: topic_1

异常场景 #

将发布代码作为独立Go程序（go run运行），订阅代码作为另一个独立程序在同一机器同时运行时，订阅端频繁出现EOF错误，输出如下：

2022/08/11 19:54:50 Connected
2022/08/11 19:54:54 Connect lost: EOF
...
2022/08/11 19:54:59 Connected
2022/08/11 19:54:59 Connect lost: EOF

代码片段 #

发布代码 #

client := mqtt.NewClient(opts)
if token := client.Connect(); token.Wait() && token.Error() != nil {
    log.Fatalln(token.Error())
}

publish(client, *topic)
client.Disconnect(250)
...

func publish(client mqtt.Client, topic string) {
    num := 10
    for i := 0; i < num; i++ {
       text := fmt.Sprintf("Message %d", i)
       token := client.Publish(topic, 0, false, text)
       token.Wait()
       time.Sleep(time.Second)
    }
}

订阅代码 #

client := mqtt.NewClient(opts)
if token := client.Connect(); token.Wait() && token.Error() != nil {
   log.Fatalln(token.Error())
}

sub(client, *topic)
time.Sleep(11 * time.Second)
client.Disconnect(250)

...

func sub(client mqtt.Client, topic string) {
    token := client.Subscribe(topic, 1, nil)
    token.Wait()
    log.Printf("Subscribed to topic: %s", topic)
}

MQTT客户端配置 #

opts := mqtt.NewClientOptions()
opts.AddBroker(fmt.Sprintf("tls://%s:%d", *host, 8883))
opts.SetClientID("basicPubSub")
opts.SetTLSConfig(tlsConfig)
opts.SetDefaultPublishHandler(messagePubHandler)
opts.OnConnect = connectHandler
opts.OnConnectionLost = connectLostHandler

推测问题与同时存在两个AWS IoT/MQTT连接有关。

问题根源是两个客户端使用了相同的ClientID连接AWS IoT Core。

AWS IoT Core遵循MQTT 3.1.1规范：当同一个ClientID的新连接建立时，服务端会强制断开旧连接。这就是订阅端频繁出现EOF（连接被强制关闭）的原因——发布程序用相同的ClientID连接后，订阅端的连接被踢下线，触发ConnectionLost事件。

解决步骤：

1. 为每个客户端生成唯一的ClientID
   • 可以在ClientID后添加随机字符串、进程ID或时间戳，示例：

     import "github.com/google/uuid"
     // 生成带UUID的唯一ClientID
     opts.SetClientID("basicPubSub-" + uuid.NewString())
     

     或基于进程ID生成：

     import "os"
     // 生成带进程ID的唯一ClientID
     opts.SetClientID(fmt.Sprintf("basicPubSub-%d", os.Getpid()))
     

2. 验证ClientID唯一性
   • 确保发布和订阅程序使用的ClientID完全不同，避免重复冲突

修改后重新运行两个独立程序，订阅端的连接就不会被强制断开，EOF错误会消失。

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