websocket - 从消费者外部收听 django channel channel_layer

Question

文档说明可以从消费者外部发布到 channel 层:https://channels.readthedocs.io/en/latest/topics/channel_layers.html#using-outside-of-consumers 我需要做相反的事情。我有一个相当复杂的 python 脚本，它从 pubnub 读取实时数据，对其进行处理，然后通过 channel_layer 上的组将其推送给消费者。这工作正常，但我需要消费者能够向该脚本宣布他们的存在，以便它可以向他们推送数据(它目前仅在从 pubnub 获取新数据时才推送到 channel 层，这可能是每 24 小时一次)。

我决定通过让消费者发布到连接上的“存在” channel 来解决这个问题。我现在需要 pubnub 源脚本来收听这个 channel 。 我已经尝试将以下内容添加到脚本中，它不再抛出错误，但它实际上并不响应消息。它成功加入 channel 层，但消息处理程序 (receive_json) 从未触发。

from channels.generic.websocket import JsonWebsocketConsumer


class channelConsumer(JsonWebsocketConsumer):

    def __init__(self):
        self.channel_name = 'source'
    def join(self):
        async_to_sync(channel_layer.group_add)('presence', self.channel_name)
    def receive_json(self, message):
        print("Presence Detected")
        # do some stuff

进一步在代码中:

global channel_layer
channel_layer = get_channel_layer()

global listener
listener = channelConsumer()
listener.join()

正如我所说，没有明显的错误，它似乎永远不会触发 receive_json。肯定有数据包被发布到“存在”，所以问题肯定在下面的代码中。

Answer 1

要让消费者接收消息，该消费者需要插入路由器。最常见的情况是使用 URLRouter，其中消费者将从连接(例如网络套接字)接收事件。

您可以使用 ChannelNameRouter 创建一个将监听特定 channel 的消费者。例如，消费者可能是:

from channels.consumer import SyncConsumer
class ChannelConsumer(SyncConsumer):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def post_save(self, event):
        print("POSTSAVE", event)

然后在你的路由器定义上

from channels.routing import ProtocolTypeRouter, URLRouter,ChannelNameRouter
application = ProtocolTypeRouter({
    "websocket": TokenAuthMiddlewareStack(
        URLRouter([
            url(r"^.*$", RestConsumer),
        ]),
    ),
    "channel":ChannelNameRouter({
        "signals": ChannelConsumer,
    })

})

现在您可以向该消费者发送消息

from channels.layers import get_channel_layer
channel_layer = get_channel_layer()
async_to_sync(channel_layer.send)("signals", {"type": "post.save", "message":"Hello Consumer"})

特别针对您的情况，如果您希望多个 channel 接收消息，您还需要将“信号” channel 添加到您要使用的组中。

请注意，如果您使用 ChannelNameRouter，您将需要启动将处理它的工作程序。

./manage.py runworker signals