Как обрабатывать большое количество сообщений MQTT за очень короткое время в Python с помощью paho, проблемы с правильны ⇐ Python

[Anonymous]

Я использую Raspbery Pi 4b (Raspbian OS, Bullseye) с работающим на нем брокером Mosquitto. В моей сети есть устройство Shelly (shelly pm mini), которое измеряет мощность и отправляет сообщение MQTT при каждом изменении мощности. На My Raspberry я хочу получать сообщения и сохранять данные в банке данных, чтобы позже отображать их на веб-сайте. Я использую paho в Python (3.10) и подписался на мою тему, зарегистрировав функцию обратного вызова on_message. Проблема, с которой я столкнулся, заключается в том, что когда 3 или 4 сообщения отправляются через MQTT за короткий период времени (1-10 мс), мой код не работает. Кажется, что сообщения обрабатываются в моей функции обратного вызова параллельно, а не одно за другим. Это проблема, потому что эта проблема со временем приводит к тому, что я не могу правильно обработать полученные данные. Я добавил счетчик, который должен подсчитываться при каждом вызове функции on_message, однако, когда сообщения приходят почти в одно и то же время (см. временную метку под этим), счетчик не увеличивается, что заставляет меня думать, что процесс не идет. правильно друг за другом.
Это функция «Мой обратный вызов»:

Код: Выделить всё

 def on_message(self, client, userdata, msg):
timestamp_now = time.time()
print(timestamp_now)
print(self.counter)
self.counter = self.counter +1

И это мой результат, когда 3 сообщения приходят с разницей всего в мс.
1732657116.8512473
1
1732657116.8514547
1
1732657116.8516352
1
1732657116.8521552
1
1732657133.3355079
21732657133.3358626
2
1732657133.3359528
2
1732657137.4190025
1732657137.419096
3< /p>
1732657137.4203782
3
Как вы можете видеть, временные метки имеют разные значения миллисекунд, поэтому они разные, но счетчик не увеличивается для каждого сообщения, поэтому я думаю, что вместо этого функции можно называть параллельными. одного вызова одной и той же функции за другим.
Сначала я пытался сохранить полученные данные в свой банк данных, но мне хотелось сохранять их только каждые 60 секунд, поэтому я сравнил самую новую временную метку с последний раз сохранялся и когда эта разница была более 60, я пытался сохранить данные. Однако, когда 3 сообщения пришли быстро, между ними почти не было времени, и данные должны были быть сохранены, поскольку 60 секунд прошли, данные были сохранены три раза (все 3 SATA из 3 сообщений были сохранены).
Затем я сделал минимальный пример со счетчиком, чтобы проверить это, и обнаружил, что что-то работает не так, как ожидалось. На данный момент я понятия не имею.

Подробнее здесь: https://stackoverflow.com/questions/792 ... h-paho-pro