Paho-Mqtt django, on_message() Funktion läuft zweimal

Question

Ich benutze paho-mqtt in Django, um Nachrichten zu empfangen. Alles funktioniert gut. Die Funktion on_message() wird jedoch zweimal ausgeführt.

Ich habe versucht, Debuggen, aber es scheint wie die Funktion einmal aufgerufen wird, aber die Datenbankeinfügung geschieht zweimal, das Drucken von Nachricht geschieht zweimal, alles innerhalb der on_message() - Funktion passiert zweimal, und meine Daten werden eingefügt zweimal für jede Veröffentlichung.

Ich bezweifelte, dass es in einem parallelen Thread passiert, und installiert ein Sellerie-Redis-Backend, um die Einfügung in die Warteschlange zu stellen und doppelte Einfügungen zu vermeiden. aber die Daten werden immer noch zweimal eingefügt.

Ich versuchte auch, die Variablen zu sperren, um Probleme beim parallelen Threading zu vermeiden, aber die Daten werden immer noch zweimal eingefügt.

ich Postgres DB bin mit

Wie löse ich dieses Problem? Ich möchte die ON_MESSAGE() Funktion nur einmal auszuführen, für jede veröffentlichen

mein init Py

from . import mqtt 
mqtt.client.loop_start() 

mein mqtt.py

import ast 
import json 

import paho.mqtt.client as mqtt 


# Broker CONNACK response 
from datetime import datetime 

from raven.utils import logger 

from kctsmarttransport import settings 


def on_connect(client, userdata, flags, rc): 
    # Subcribing to topic and recoonect for 
    client.subscribe("data/gpsdata/server/#") 
    print 'subscribed to data/gpsdata/server/#' 


# Receive message 

def on_message(client, userdata, msg): 
    # from kctsmarttransport.celery import bus_position_insert_task 
    # bus_position_insert_task.delay(msg.payload) 
    from Transport.models import BusPosition 
    from Transport.models import Student, SpeedWarningLog, Bus 
    from Transport.models import Location 
    from Transport.models import IdleTimeLog 
    from pytz import timezone 
    try: 
     dumpData = json.dumps(msg.payload) 
     rawGpsData = json.loads(dumpData) 
     jsonGps = ast.literal_eval(rawGpsData) 
     bus = Bus.objects.get(bus_no=jsonGps['Busno']) 
     student = None 
     stop = None 
     if jsonGps['card'] is not False: 
      try: 
       student = Student.objects.get(rfid_value=jsonGps['UID']) 
      except Student.DoesNotExist: 
       student = None 
     if 'stop_id' in jsonGps: 
      stop = Location.objects.get(pk=jsonGps['stop_id']) 

     dates = datetime.strptime(jsonGps['Date&Time'], '%Y-%m-%d %H:%M:%S') 
     tz = timezone('Asia/Kolkata') 
     dates = tz.localize(dates) 
     lat = float(jsonGps['Latitude']) 
     lng = float(jsonGps['Longitude']) 
     speed = float(jsonGps['speed']) 

     # print msg.topic + " " + str(msg.payload) 
     busPosition = BusPosition.objects.filter(bus=bus, created_at=dates, 
               lat=lat, 
               lng=lng, 
               speed=speed, 
               geofence=stop, 
               student=student) 
     if busPosition.count() == 0: 
      busPosition = BusPosition.objects.create(bus=bus, created_at=dates, 
                lat=lat, 
                lng=lng, 
                speed=speed, 
                geofence=stop, 
                student=student) 
      if speed > 60: 
       SpeedWarningLog.objects.create(bus=busPosition.bus, speed=busPosition.speed, 
               lat=lat, lng=lng, created_at=dates) 
       sendSMS(settings.TRANSPORT_OFFICER_NUMBER, jsonGps['Busno'], jsonGps['speed']) 
      if speed <= 2: 
       try: 
        old_entry_query = IdleTimeLog.objects.filter(bus=bus, done=False).order_by('idle_start_time') 
        if old_entry_query.count() > 0: 
         old_entry = old_entry_query.reverse()[0] 
         old_entry.idle_end_time = dates 
         old_entry.save() 
        else: 
         new_entry = IdleTimeLog.objects.create(bus=bus, idle_start_time=dates, lat=lat, lng=lng) 
       except IdleTimeLog.DoesNotExist: 
        new_entry = IdleTimeLog.objects.create(bus=bus, idle_start_time=dates, lat=lat, lng=lng) 
      else: 
       try: 
        old_entry_query = IdleTimeLog.objects.filter(bus=bus, done=False).order_by('idle_start_time') 
        if old_entry_query.count() > 0: 
         old_entry = old_entry_query.reverse()[0] 
         old_entry.idle_end_time = dates 
         old_entry.done = True 
         old_entry.save() 
       except IdleTimeLog.DoesNotExist: 
        pass 
    except Exception, e: 
     logger.error(e.message, exc_info=True) 


client = mqtt.Client() 
client.on_connect = on_connect 
client.on_message = on_message 
client.connect("10.1.75.106", 1883, 60) 

Answer 1

Ich hatte das gleiche Problem!

Versuchen Sie es mit:

def on_disconnect(client, userdata, rc): 
    client.loop_stop(force=False) 
    if rc != 0: 
     print("Unexpected disconnection.") 
    else: 
     print("Disconnected")