共享订阅无法正常持续接受数据

[smalleel-qiks]

前景：Bifromq-3.3.3
租户配置：

Setting.OutBoundBandWidth=1024 * 1024 * 1024L
Setting.OutBoundBandWidth= 512 * 1024L
Setting.InBoundBandWidth= 512 * 1024L
Setting.MaxSharedGroupMembers= 1000
Setting.SessionInboxSize= 1000

共享订阅客户端采用Vertx 4.5.11，

MqttClientOptions options = new MqttClientOptions()
.setClientId(info.clientId())
.setUsername(info.username())
.setPassword(info.password())
.setAckTimeout(30)
.setCleanSession(cleanSession)
.setKeepAliveInterval(60)
.setReceiveBufferSize(1024* 1024 * 1024)
.setMaxMessageSize(512*1024)
.setMaxInflightQueue(100_000)
;

订阅topic，2两个客户端同时订阅一下两个topic

$share/g2/g5rmdv6wMCU/+/user/update
$oshare/g1/as/mqtt/status/g5rmdv6wMCU/+

消息体：

{"deviceName": "856000001315", "msg": {"action": "report", "iccid": "110", "phase": 0, "status": "on", "interval": "1", "poweryg": 6, "poweryga": 1, "powerygb": 2, "powerygc": 3, "powerwg": 6, "powerwga": 1, "powerwgb": 2, "powerwgc": 3, "powerrate": 1, "powerratea": 1, "powerrateb": 1, "powerratec": 1, "frequency": 1, "zxyg": 10, "zxwg": "0", "fxyg": 0, "fxwg": "0", "zhyg": 10, "dl": 20, "dlb": 20, "dlc": 20, "dy": 200, "dyb": 200, "dyc": 200, "signal": 15}}

[popduke]

请按ISSUE模板要求把MQTT连接、发布、订阅相关参数以及ISSUE上下文补充完整。

[smalleel-qiks]

Environment



• Version: 3.3.3
• JVM Version: 17.0.8
• Hardware Spec: 32C32G,2.5GbE
• OS: Windows11

Reproducible Steps

• PUB Client Parameters
  • MQTT Connection:
    • version: 5.0
    • Keep Alive: 6
    • Clean Session: true
  • MQTT Pub:
    • Topic: g5rmdv6wMCU/{deviceName}/user/update
    • QoS: 0
    • Retain: false
• SUB Client Parameters:
  • MQTT Connection:
    • Clean Session: true
    • KeepAliveInterval: 60
    • ReceiveBufferSize: 256 * 1024 * 1024
    • MaxMessageSize: 256 * 1024
    • MaxInflightQueue: 100_000
    • AckTimeout: 30
  • MQTT Sub:
    • TopicFilter: $share/g1/g5rmdv6wMCU/+/user/update,$share/g1/as/mqtt/status/g5rmdv6wMCU/+
    • QoS: 1
  • Connection Count:
    • PUB Client count: 2000
    • SUB Client count: 2
  • Load Generated:
    • PUB QPS per connection: 36000/s
    • SUB QPS per connection: Max 2400
    • Payload size: 512byte

[smalleel-qiks]

如果我这样做可以保持执行订阅数据，但是我不知道会发生什么
```
vertx.setPeriodic(1000, v -> {
if(client.isConnected()){
client.subscribe(topicFilterMap);
}
});

[popduke]

MQTT Sub:
TopicFilter: $share/g1/g5rmdv6wMCU/+/user/update,$share/g1/as/mqtt/status/g5rmdv6wMCU/+
QoS: 1
Connection Count:
PUB Client count: 2000
SUB Client count: 2
Load Generated:
PUB QPS per connection: 36000/s
SUB QPS per connection: Max 2400
Payload size: 512K

单订阅端每秒收2400条 512kb 大小的qos1消息？

[smalleel-qiks]

是的，qos1单订阅只能达到1200每秒

[popduke]

评估下你的资源条件能不能支撑单订阅连接2400 * 512KB ~= 1200MB/s的接收能力

[smalleel-qiks]

抱歉 每条消息大约512byte

[smalleel-qiks]

参数已调整还是会出现：共享订阅无法正常持续接受数据

[smalleel-qiks]

不使用共享订阅功能可以持续收到消息

[smalleel-qiks]

共享订阅搭配通配符是否存在一定问题嘞？

[popduke]

能在其他测试客户端复现吗？或可以提供下你完整的复现程序

[smalleel-qiks]

代码地址：https://pan.baidu.com/s/1bpCC4rKMOHJvn2n3k3FAKw?pwd=qj4q
问题点：com.smalleel.iot.core.sub.integrator.impl.MqttIntegrator第90行、如果注释掉会导致无法持续接收订阅数据
启动类：src/test/ com.smalleel.iot.core.sub.BifromqDataStreamTest

等候您的回复，感谢！

[popduke]

共享订阅是MQTT5的能力，vertx mqtt client只支持3.1.1，你需要使用其他Client验证。

[smalleel-qiks]

为什么您用vertx mqtt client 能够实现共享订阅呢。
Integration SDK

[smalleel-qiks]

使用eclipse.paho.mqttv5 还是存在无法持续订阅数据问题

[popduke]

pub

share sub 1

share sub 2

用第三方工具测试并没有出现你的说的情况

[smalleel-qiks]

使用mqttx同样会出现无法持续订阅

[smalleel-qiks]

cleanSession=false 可以保持持续订阅