V2EX = way to explore
V2EX 是一个关于分享和探索的地方
现在注册
已注册用户请  登录
IfEles
V2EX  ›  程序员

mqtt 消费者并发情况下丢消息的情况

  •  
  •   IfEles · 2022-03-12 13:39:47 +08:00 · 2455 次点击
    这是一个创建于 748 天前的主题,其中的信息可能已经有所发展或是发生改变。
    • 折腾了半天,发现 mqtt 消费者端在生产者(物联网设备)并发起来后数据丢失的问题,生产者那边推送频率 1~3s 左右推送一次,消费者这里使用 springboot2.0+spring-integration-mqtt 来接收报文并且解析报文,初步怀疑是接收和解析放一起导致线程阻塞?目前想法是先把数据接下来后期再去处理,不知道有没有大佬遇到过同样的问题
    • emq 里的消费者会话状态如下: bTvHdP.png
    14 条回复    2022-03-14 20:40:00 +08:00
    LLaMA2
        1
    LLaMA2  
       2022-03-12 15:31:46 +08:00
    麻烦提供您发布的主题、订阅的主题、QOS 、遗嘱消息
    yazinnnn
        2
    yazinnnn  
       2022-03-12 15:49:54 +08:00
    提高两侧 qos 级别?
    不知道 spring 集成 mqtt 有没有实现 qos2
    IfEles
        3
    IfEles  
    OP
       2022-03-12 16:33:26 +08:00
    @yazinnnn 客户端这边订阅的 qos 是 0
    IfEles
        4
    IfEles  
    OP
       2022-03-12 16:34:13 +08:00
    @ye4tar 发布和订阅的主题都是 /iot/data, qos 是 0,医嘱消息没配
    wzcloud
        5
    wzcloud  
       2022-03-12 17:21:55 +08:00
    0 表示至多一次,本来就有可能丢消息吧
    yazinnnn
        6
    yazinnnn  
       2022-03-12 17:39:03 +08:00
    不想丢消息的话至少需要 qos1 ,如果想保证只收到一次的话,则需要 qos2

    不知道 emqx 的 broker 对储存消息的数量有没有限制,如果业务不是很复杂的话,其实可以自己实现一个 mqtt server/broker ,做一些定制开发
    LLaMA2
        7
    LLaMA2  
       2022-03-12 17:46:40 +08:00
    我说一些简单的经验吧,也不是很多,做过一些,仔细读过一点文档。

    首先他就是传统 socket 通讯的一种解决方案,如果你有空,你完全可以自己用 socket 仿照他的套路自己实现类似 MQTT 的东西

    其次在传统的 socket 通讯时候,消息的发送者-接受者直接建立链接,发送者是知道接受者已经接收到消息了,而在 MQTT 中发送者和接受者中间隔着一个 broker ,有点像邮局。你只是知道你的消息已经成功被邮局收下了,即发对应主题下消息发布成功。而对方有没有收到消息你是不知道的,如果你想知道,你需要和 broker 协商一个主题,当然,最新的 MQTT 协议已经有现成的实现了。


    再者,QOS 级别有点像邮局会员服务,不同的级别他会按照不同的送达率给你处理消息,这取决于消息的重要程度和你设备的抗造能力。有时候你不得不取舍。

    最后,遗嘱很有用,他会让 MQTT client 知道什么时候发送消息,什么时候该提醒用户设备有一点点故障。例如 broker 服务器有计划的重启了,重启之前发出遗嘱。IOT 设备可以马上就知道 broker 服务器正在重启,这时候 IOT 设备就可以按照事先想好的方式解决这个问题。

    还有,不同家的 MQTT Broker 实现对协议支持的完成度天差地别,选择的时候一定要自己阅读相关的说明文件,EMQ 很好,不过我不确定谁是最好的。我也不知道。
    IfEles
        8
    IfEles  
    OP
       2022-03-12 19:47:33 +08:00
    @ye4tar 感谢,还码了这么多字,主要也是第一次用这个 emq 对原理理解还不到位
    IfEles
        9
    IfEles  
    OP
       2022-03-12 19:48:56 +08:00
    @yazinnnn 好的,我尝试调整一下 qos
    IfEles
        10
    IfEles  
    OP
       2022-03-12 19:49:22 +08:00
    @wzcloud 我去看了一下好像是这样
    wzcloud
        11
    wzcloud  
       2022-03-12 22:51:27 +08:00
    emq 我之前项目用过,开源免费的,基于 Erlang 开发,有收费版;
    RabbitMQ 有 mqtt 插件;Netty 也支持 MQTT 协议;
    其他一些物联网平台也支持 MQTT 协议(他们做了 MQTT 中间件,但是使用的话,就需要你的设备连接到他们的平台了,比如中国移动的 OneNet 平台,电信联通好像也有类似平台).阿里云好像也有 MQTT 队列的服务.
    cshlxm
        12
    cshlxm  
       2022-03-13 22:10:37 +08:00
    所有收到的消息先桥接到消息队列,应该可以解决这个问题,emq 对并发处理还是挺好的,还是看下你消费者是不是处理的慢了,造成缓存队列里已经存不下了
    IfEles
        13
    IfEles  
    OP
       2022-03-14 08:26:43 +08:00
    @cshlxm 多谢,目前并发还是挺高的我也认为是这个问题
    cshlxm
        14
    cshlxm  
       2022-03-14 20:40:00 +08:00
    @IfEles 社区版说是 4C8G 能扛 10w 连接
    关于   ·   帮助文档   ·   博客   ·   API   ·   FAQ   ·   我们的愿景   ·   实用小工具   ·   2741 人在线   最高记录 6543   ·     Select Language
    创意工作者们的社区
    World is powered by solitude
    VERSION: 3.9.8.5 · 25ms · UTC 12:40 · PVG 20:40 · LAX 05:40 · JFK 08:40
    Developed with CodeLauncher
    ♥ Do have faith in what you're doing.