【发布时间】：2017-06-23 06:48:12
【问题描述】：
我在 django 中使用 paho-mqtt 来接收消息。一切正常。但是 on_message() 函数执行了两次。
我试过调试，但似乎函数被调用了一次，但是数据库插入发生了两次，消息的打印发生了两次，on_message() 函数中的所有内容都发生了两次，我的数据被插入了两次每次发布。
我怀疑它发生在并行线程中，并安装了一个 celery redis 后端来对插入进行排队并避免重复插入。但数据仍然被插入两次。
我也尝试锁定变量，以避免并行线程出现问题，但数据仍然插入两次。
我正在使用 Postgres 数据库
我该如何解决这个问题？我希望 on_message() 函数每次发布只执行一次
我的 init.py
from . import mqtt
mqtt.client.loop_start()
我的 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)
【问题讨论】：
-
我唯一知道的是你运行了2次脚本，而不是函数，你真的找到解决方案2吗？我有同样的问题
-
看起来 init 被调用了两次。有人说 Django 有两个进程，一个是主进程，一个是用于实时更改的自动更新。如果您获得进程 ID（导入 OS os.getpid），您将看到两个不同的进程。见stackoverflow.com/questions/2110545/…。使用 --noreload 标志仅加载一个进程。
-
@JavierCampos 实际上我最终对数据库设置了一个唯一约束，以避免重复数据输入。因此，当我尝试保存相同的对象时，数据库会抛出错误，我会捕获该错误并拒绝该消息。这不是最有效的，但那个黑客暂时解决了这个问题。我还没有测试 --noreload 标志。我认为这应该可以解决问题。
标签： python django mqtt postgresql-9.3