如何提升OpenFaaS mqtt-connector触发器的消息处理速度？

问题背景 #

我们正尝试将LD-19激光雷达的数据接入运行在OpenFaaS子集Faasd上的无服务器函数，触发逻辑如下：

• 激光雷达检测到物体进入指定距离范围时，向MQTT主题发布数据
• OpenFaaS的mqtt-connector通过触发器调用无服务器函数

当前核心问题：物体进入范围后1秒内，mqtt-connector就出现严重消息堆积

已尝试的无效方案：

• 将MQTT QoS调整为0，未缓解堆积
• 尝试降低LD-19激光雷达的轮询频率，但未找到有效方法
• 试图提升mqtt-connector的CPU利用率，仅找到限制CPU的配置方式，无提升手段

当前函数处理代码 #

import json
import paho.mqtt.client as mqtt

TOPIC = 'controller'
PORT = 1883
BROKER = "172.17.0.1" #mosquitto broker IP

def handle(event):
    message = event.decode()
    print("Received message: {}".format(message))
    msg = 'stop'
    client = mqtt.Client()

    client.connect(BROKER, PORT, 60)
    client.publish(TOPIC, msg)

    return None

解决方案 #

1. 优化无服务器函数的MQTT客户端性能 #

当前函数每次调用都创建新MQTT客户端并建立连接，是极大的性能损耗，会拖慢函数处理速度，间接加剧消息堆积。优化方式：

• 复用全局MQTT客户端（Faasd函数实例温启动时，全局变量会被保留）
• 避免每次调用执行connect操作，改为初始化时连接，后续调用直接复用

优化后的代码示例：

import paho.mqtt.client as mqtt

TOPIC = 'controller'
PORT = 1883
BROKER = "172.17.0.1" #mosquitto broker IP

# 全局MQTT客户端，复用连接
client = None

def init_mqtt_client():
    global client
    if client is None or not client.is_connected():
        client = mqtt.Client()
        client.connect(BROKER, PORT, 60)
        # 启动后台线程处理MQTT网络事件，避免publish阻塞或失败
        client.loop_start()

def handle(event):
    message = event.decode()
    print("Received message: {}".format(message))
    
    # 初始化/复用MQTT客户端
    init_mqtt_client()
    
    msg = 'stop'
    client.publish(TOPIC, msg)

    return None

2. 在MQTT层面做消息限流/去重 #

针对激光雷达消息量过大的问题，可在发布端或Broker层处理：

• Broker端配置：如果使用Mosquitto，可配置max_inflight_messages限制在途消息数，或设置主题消息过期时间，清理旧消息
• 发布端过滤：修改激光雷达数据处理逻辑，仅在物体首次进入范围或**状态变化（进入/离开）**时发布消息，而非持续输出；若必须持续监测，可设置固定间隔（比如每秒1条）

3. 调整LD-19激光雷达的输出频率 #

LD-19默认输出频率为10Hz（每秒10次扫描），可通过串口指令调整：

• 发送串口指令AA 08 00 01 00 00 00 00 09将频率改为1Hz，具体指令格式可参考官方文档
• 若通过驱动程序读取数据，可在驱动层添加采样间隔，比如每100ms读取一次，而非实时读取

4. 优化mqtt-connector的并发处理能力 #

mqtt-connector默认并发较低，可通过以下方式调整：

• 修改mqtt-connector配置文件，调高max-workers参数，提升并发处理的函数调用数量
• 在Faasd中给mqtt-connector分配更多CPU资源：调整容器CPU配额（比如设置cpu_limit: 1000m），取消或提高之前的CPU限制

5. 函数触发策略优化 #

将触发器改为批量处理模式：让mqtt-connector积累一定数量的消息后再调用一次函数，而非每条消息单独触发，减少函数调用开销。

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