WMMSE方法的使用笔记

业务背景

数据采集架构

仅看MQTT数据采集，整个过程是这样的，

• kepServer。kepServer上配置设备需要采集的plc数据点位，kepServer自带的 IOT gateway，可以针对任意配置点位推送至EMQX服务器topic以及点位推送速率，为了调试配置了几个点位有虚拟点位和真实点位
• EMQX。启动EMQX：MQTT服务器（业内比较常用的MQTT服务器）
• IOT服务。 kepServer IOT配置点位数据发生变化后，将数据推送给EMQX，IOT服务监听对应topic解析组装数据直接将消息推送至Kafka。
• kafka。Kafka为所有类型数据的入口，所有类型数据统一推送至Kafka，如ModBus、MQTT、HttpApi…
• Flink服务。Flink服务实时消费Kafka数据，根据IOT服务中配置与kepServer上对应的点位以及针对不同点位配置的告警模板，根据阈值或者状态等其他规则实时处理数据。
• WebSocket。目前Flink集成WebSocket，根据阈值实时推送给前端，实时展示数据，如传感器温湿度、设备状态…

现象

调试真实点位

分析

查看现有日志，初步分析发现消息生产到推送到Kafka有10s延迟

初步分析：

• kepserver 消息产生有时间
• emqx可以配置日志级别为 debug，查看接受到消息的时间 延迟没问题
• 发送Kafka之前的逻辑比较简单，不会有延迟

后续，又将IOT监听MQTT消息初打日志，发送Kafka消息耗时时间打印。

对比多个关键节点时间，发现两个延迟点：

1. kepServer数据发生变化的时间和EMQX接收到kepServer推送的消息的时间对比：发现有个5-8s的延迟【kepServer->EMQX】
2. Kafka发送消息到成功回调：有个固定2s左右的延迟【Kafka】

最终解决

全链路排查

1. kepServer

   • kepServer IOT gateway -> rate 速率设置，由10000 -> 1000

   • 这个参数改为1000之后呢（kepServer延迟得到解决），原本8-10s的延迟，变为了3s延迟左右

2. kafka

   • 通过代码中各个关键节点打的日志，发现Kafka发消息到成功回调基本稳定在2s延迟左右，偶尔会有基本无延迟的情况（这种情况有点意思）

   • 各种查资料发现Kafka有如下几个与消息缓存区相关的参数

         kafka:
             ...
             producer:
                 batch-size: 16384 # kafka本地线程会去缓冲区中⼀次拉16k的数据，发送到broker
                 buffer-memory: 33554432 # 消息缓冲区默认32m
                 ...
                 properties:
                     linger:
                         ms: 10 # 默认 10ms
     

   原本这个时间参数刚好设置的就是2000ms(kafka.producer.properties.linger.ms)，与上述刚好固定两秒延迟相符，偶尔有无延迟现象说明该消息刚发送就到了2s的频次直接就发生了无延迟

   • 该时间参数不设置或者设置小一些对延迟有明显的提升，但这样就会频繁发送消息增大网络开销，自行根据业务取舍
   • 整个默认10ms吧