为什么脚本没有从 mosquitto 接收数据？

Question

我有以下代码接收来自不同 MQTT 主题的数据。

#!/usr/bin/python3
import paho.mqtt.client as mqtt
import pymysql
import json
import sys
import time

#-### Constantes ####
PUERTO_MQTT = ...
USUARIO = ...
CLAVE = ...
#DB_LOCAL = ...
DB_LOCAL = ...
RETARDO_SUSCRIP = 30 #s

#-### Variables ####
idsNgsConectados = set()
clienteMQTT = mqtt.Client("receptor_datos_ngs")
db_local = None
t1 = 0
t2 = 0

#-### Funciones ####
def on_connect(clienteMQTT, userdata, flags, rc):
    print("Conexion establecida con el broker MQTT correctamente")

def on_message(clienteMQTT, userdata, msg):
    msg_str = msg.payload.decode('utf-8')

    # procesamiento de los mensajes MQTT
    posPrimeraBarra = msg.topic.find('/')

    print("topic: {}".format(msg.topic))
    print("datos: {}".format(msg_str))

    if posPrimeraBarra != -1:
        id_ngs = int(msg.topic[:posPrimeraBarra])
        ref_medicion = msg.topic[posPrimeraBarra+1:]
        variables = json.loads(msg_str)

        if type(variables) is dict:
            if id_ngs in idsNgsConectados:
                try:
                    cursor = db_local.cursor()
                    cursor.execute("INSERT INTO mediciones (id_ngs, referencia) VALUES ({}, '{}')".format(id_ngs, ref_medicion))
                    db_local.commit()
                    cursor.execute("SELECT id FROM mediciones WHERE (id_ngs={} AND referencia='{}') ORDER BY id DESC LIMIT 1".format(id_ngs, ref_medicion))
                    id_medicion = cursor.fetchone()[0]
                    for var in variables:
                        cursor.execute("INSERT INTO valores_mediciones (id_medicion, variable, valor) VALUES ({},'{}',{})".format(id_medicion, var, variables[var]))
                    db_local.commit()
                    cursor.close()
                except Exception as e:
                    db_local.rollback()
                    print("Error 1: fallo el procesamiento de un mensaje MQTT: " + str(e), file=sys.stderr)
                    sys.exit(1)
            else:
                print("Advertencia: un ngs envio datos sin haberse presentado, por lo que se le indicara que salude primero", file=sys.stderr)
                clienteMQTT.publish(str(id_ngs),"saludar",qos=1)
        else:
            print("Advertencia: un ngs envio datos con un formato incorrecto por lo que se le indicara que se reinicie", file=sys.stderr)
            clienteMQTT.publish(str(id_ngs),"reiniciar",qos=1)

def sub_topics():
    global clienteMQTT
    cursor = db_local.cursor()
    cursor.execute('SELECT id FROM ngs')
    ids_ngs = cursor.fetchall()
    cursor.close()
    print("Topics suscriptos:")
    for id_ngs in ids_ngs:
        id_ngs = id_ngs[0]
        idsNgsConectados.add(id_ngs)
        topic = str(id_ngs) + '/#'
        print("\t{}".format(topic))
        clienteMQTT.subscribe(topic)

try:
    db_local = pymysql.connect( unix_socket='/run/mysqld/mysqld.sock',
                    user=USUARIO,
                    password=CLAVE,
                    db=DB_LOCAL )

    clienteMQTT.on_connect = on_connect
    clienteMQTT.on_message = on_message
    clienteMQTT.username_pw_set(username=USUARIO, password=CLAVE)

    clienteMQTT.connect("localhost", PUERTO_MQTT)
    clienteMQTT.loop_start()

    # Bucle infinito
    while True:
        time.sleep(RETARDO_SUSCRIP)
        # Suscripcion a los topics de todos los ngs
        sub_topics()
except Exception as e:
    db_local.close()
    clienteMQTT.loop_stop()
    clienteMQTT.disconnect()
    print("Error 2: desconocido: " + str(e), file=sys.stderr)
    sys.exit(2)

脚本动态订阅主题。如果我从 shell 运行脚本，它运行良好，但如果将它设置为使用 systemd 在启动时运行，它会失败。我已将单元文件设置为需要并在 mosquitto、mariadb、dhcpcd 和 wpa_supplicant 服务之后运行。我确定数据是从另一台设备发布的，因为我可以使用“mosquitto_sub”接收它。 可能是什么原因？

它在 Raspberry pi 零 w 中运行，Mosquitto 版本为 1.5.7。

Answer 1

我可以找到问题的原因，不是因为mosquitto 或paho-mqtt，而是因为pymysql。代码中的失败是我没有在 sql 事务中使用方法commit()。这导致代码无法获取数据库中另一个进程引入的新数据。这是因为 InnoDB 的默认隔离级别 REPEATABLE READ。您可以阅读更多关于 here 的信息。进行该 sql 事务的正确方法如下：

def sub_topics():
    cursor = db_local.cursor()
    cursor.execute('SELECT id FROM ngs')
    ids_ngs = cursor.fetchall()
    print("Topics suscriptos:")
    for id_ngs in ids_ngs:
        id_ngs = id_ngs[0]
        idsNgsConectados.add(id_ngs)
        topic = str(id_ngs) + '/#'
        print("\t{}".format(topic))
        clienteMQTT.subscribe(topic)
    db_local.commit()
    cursor.close()

此代码使用该数据来确定它必须订阅哪些主题，因此该代码没有订阅预期的主题，因此它没有从 mosquitto 接收相关数据。

pymysql 文档未涵盖该主题，因此我建议新手（我意识到我是 50% 新手 50% 专业 jaja）阅读有关 ACID 和您正在使用的 RDBMS 的实现细节。