django - 如何从 django-celery 3 任务发送 channel 2.x 组消息？

Question

我需要推迟发送 channel 消息。这是我的代码:

# consumers.py
class ChatConsumer(WebsocketConsumer):
    def chat_message(self, event):
        self.send(text_data=json.dumps(event['message']))

    def connect(self):
        self.channel_layer.group_add(self.room_name, self.channel_name)
        self.accept()

    def receive(self, text_data=None, bytes_data=None):
        send_message_task.apply_async(
            args=(
                self.room_name,
                {'type': 'chat_message',
                 'message': 'the message'}
            ),
            countdown=10
        )

# tasks.py
@shared_task
def send_message_task(room_name, message):
    layer = get_channel_layer()
    layer.group_send(room_name, message)

任务正在执行，我看不到任何错误，但消息未发送。只有当我从消费者类方法发送它时它才有效。

我还尝试使用 AsyncWebsocketConsumer 并使用 AsyncToSync(layer.group_send) 发送。它错误为“您不能在与异步事件循环相同的线程中使用 AsyncToSync - 直接等待异步函数即可。”

然后我尝试将 send_message_task 声明为异步并使用 await。什么也没有发生(没有错误)，我不确定任务是否完全执行。

以下是版本:

Django==1.11.13
redis==2.10.5
django-celery==3.2.2
channels==2.1.2
channels_redis==2.2.1

设置:

REDIS_HOST = os.getenv('REDIS_HOST', '127.0.0.1')
BROKER_URL = 'redis://{}:6379/0'.format(REDIS_HOST)
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": ['redis://{}:6379/1'.format(REDIS_HOST)],
        },
    },
}

有什么想法吗？

UPD:刚发现redis channel layer被retrieved但是group_send方法没有被调用，直接跳过了。

UPD 2:使用 AsyncToSync(layer.group_send) 从控制台发送。不使用 apply_async 调用任务也可以。但是使用 apply_async 运行它会导致错误 您不能在与异步事件循环相同的线程中使用 AsyncToSync - 直接等待异步函数。当然，将任务定义为异步并使用 await 也会破坏一切。

Answer 1

也许这不是对起始问题的直接回答，但这可能会有所帮助。如果您遇到异常“您不能在与异步事件循环相同的线程中使用 AsyncToSync - 只需直接等待异步函数”，那么您可能会执行以下操作:

1. 事件循环在某处创建
2. 一些异步代码开始
3. 一些SYNC代码被ASYNC代码调用
4. SYNC 代码正在尝试使用 AsyncToSync 调用 ASYNC 代码以防止这种情况发生

似乎 AsyncToSync 检测到外部事件循环并决定不干扰它。

解决方案是直接将异步调用包含在外部事件循环中。示例代码如下，但最好检查您的情况以及外循环是否正在运行...

loop = asyncio.get_event_loop()
loop.create_task(layer.group_send(room_name, {'type': 'chat_message', 'message': message}))