是否可以在使用Redis后端的芹菜Worker中使用mqttasgi发送MQTT消息?

Question

我在Django中使用mqttasgi库来接收大量消息，并使用REDIS队列处理它们，我想将此信息发布回另一个TOPIC。这可能吗?如果是，我该怎么做?目前，我只将publish函数覆盖到我的消费者中，如下所示。

from mqttasgi.consumers import MqttConsumer
from mqtt_handler.tasks import processmqttmessage
import json
class MyMqttConsumer(MqttConsumer):
async def connect(self):
await self.subscribe('application/5/device/+/event/up', 2)
async def receive(self, mqtt_message):
print('Received a message at topic:', mqtt_message['topic'])
print('With payload', mqtt_message['payload'])
print('And QOS:', mqtt_message['qos'])
print(type(mqtt_message['payload']))
dictresult = json.loads(mqtt_message['payload'])
print(type(dictresult))
print(dictresult)
jsonresult = json.dumps(dictresult)
print(type(jsonresult))
print(jsonresult)
processmqttmessage.delay(jsonresult)
print("test")
pass
async def publish(self, topic, payload, qos=1, retain=False):
await self.send({
'type': 'mqtt.pub',
'mqtt': {
'topic': topic,
'payload': payload,
'qos': qos,
'retain': retain,
}
})
async def disconnect(self):
await self.unsubscribe('application/5/device/+/event/up')

我想能够能够现在发布，但从我的任务processmqttmessage的内部。

谢谢。

Pd: @Santiago Ivulich也许你能帮我。

Answer 1

是的，这是可能的，没有必要覆盖基本消费者的发布。我建议将需要发布的结果返回给MQTTAsgi，以便维护单个MQTT连接。为此，您可以在通道层中使用一个组，以便将需要发送的内容发送回mqttasgi。

from mqttasgi.consumers import MqttConsumer
from mqtt_handler.tasks import processmqttmessage
import json
class MyMqttConsumer(MqttConsumer):
async def connect(self):
await self.subscribe('application/5/device/+/event/up', 2)
# Subscribe consumer to channel layer group.
await self.channel_layer.group_add("my.group", self.channel_name)
async def receive(self, mqtt_message):
print('Received a message at topic:', mqtt_message['topic'])
print('With payload', mqtt_message['payload'])
print('And QOS:', mqtt_message['qos'])
print(type(mqtt_message['payload']))
dictresult = json.loads(mqtt_message['payload'])
print(type(dictresult))
print(dictresult)
jsonresult = json.dumps(dictresult)
print(type(jsonresult))
print(jsonresult)
processmqttmessage.delay(jsonresult)
print("test")
pass
async def publish_results(self, event):
data = event['text']
self.publish('my/publish/topic', data, qos=2, retain=False)

async def disconnect(self):
await self.unsubscribe('application/5/device/+/event/up')

从芹菜任务:

from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync
def processmqttmessage():
...
channel_layer = get_channel_layer()
async_to_sync(channel_layer.group_send)("my.group", 
{"type": "publish.results", "text":"Hi from outside of the consumer"})

如果多个消费者将同时运行，您可以通过编程为组生成一个名称并将其作为参数传递给任务。

重要提示:确保您在芹菜和mqttasgi项目中使用相同的通道后端。