Mqtt客户端不能同时处理多条消息答案

【问题标题】：Mqtt client can't handle multiple messages at the same timeMqtt客户端不能同时处理多条消息
【发布时间】：2017-12-26 14:26:58
【问题描述】：

我正在尝试让 mqtt 客户端接收来自它订阅的主题的所有消息，但每次我使用另一个客户端发送消息时它只接收第一个消息。问题是客户端应该像这样处理10 条 qos 2 的消息，但只处理第一个消息。消息以几毫秒的时间间隔同时发送。我不会经常发送消息。我每分钟发送 10 条消息。两个客户端都是持久的。我确定消息会离开发布者，因为每当发送消息时，我都会打印它的有效负载。我使用的是 qos 2，因为收到的消息会保存到数据库中，我不想重复。我使用的代理是activemq。那么问题是为什么会发生这种情况？

from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from sqlalchemy import update
from sqlalchemy.ext.automap import generate_relationship
import sqlalchemy
import paho.mqtt.client as mqtt
import time  
#Function that define what to do on client conenction
def on_connect(client, userdata, rc):
    #Subscribe to all specified topics
    mqttc.subscribe(topic='/+/mysignals/sensors/+/')
def on_message(client,userdata,message):
    #Get the mysignals member id from the topic
    topic_split = message.topic.split('/')
    member_id = topic_split[1]
    session = Session(engine)
    sensor_id = topic_split[4]
    patient = session.query(Patient).filter(Patient.mysignalsid==member_id).first()
    if message.payload == None:
        payload = 0
    else:
        payload = message.payload
    if patient:
        current_time = time.time()
        if patient.id in pending.keys() and (current_time - pending[patient.id]['time_created']) <= 55:
            pending[patient.id]['record'].__dict__[sensor_id] = payload
            print time.time()
        else:
            pending.pop(patient.id,None)
            patientdata = PatientData()
            patientdata.__dict__[sensor_id] = payload
            print patientdata.__dict__[sensor_id]
            print payload
            print patientdata.temp
            patient.patientdata_collection.append(patientdata)
            session.add(patientdata)
            print time.time()
            pending.update({patient.id:{
                                    'time_created':time.time(),
                                    'record':patientdata,
                                    }})
        session.flush()
        session.commit()
        print('Wrote to database.')

pending = {}
Base = automap_base()
engine = create_engine('mysql+mysqlconnector://user:pass@localhost/db')
# reflect the tables
Base.prepare(engine, reflect=True)
Patient = Base.classes.patient
PatientData = Base.classes.patientdata
session = Session(engine)
#Create a mqtt client object
mqttc = mqtt.Client(client_id='database_logger',clean_session=False)
#Set mqtt client callbacks
mqttc.on_connect = on_connect
mqttc.on_message = on_message
#Set mqtt broker username and password
mqttc.username_pw_set('blah','blahblah')
#Connect to the mqtt broker with the specified hostname/ip adress
mqttc.connect('127.0.0.1')
mqttc.loop_forever()

输出：

98
98
None
1500576377.3
Wrote to database.
1500576377.43
Wrote to database.

输出应该是：

98
98
None
1500576377.3
Wrote to database.
25.4
25.4
25.4
1500576377.43
Wrote to database.

【问题讨论】：

编辑问题以显示您的代码
我更新了帖子并添加了我的代码。
首先，您的主题不应以/ 开头和结尾，其次您从代码中得到什么输出？
编辑问题，不要尝试在 cmets 中发布输出
您应该在答案中发布您的解决方案。此外，您不应在标题中添加“已解决”，而应通过单击答案左侧的复选标记来接受您的答案。有关本网站如何运作的更多信息，请阅读tour。

标签： python sqlalchemy activemq mqtt paho

【解决方案1】：

最终不是mqtt客户端的问题。代码错误，第二条消息没有写入数据库。

为了让它工作，我不得不替换以下行：

pending[patient.id]['record'].__dict__[sensor_id] = payload

用这个：

setattr(pending[patient.id]['record'],sensor_id,payload)

同时删除该行：

session = Session(engine)

on_message 函数之外。

我还加了一行：

session.expunge_all()

线下：

session.commit()

为了每次在数据库中完成事务时清理会话。

【讨论】：