Django从Django Channels Async websocket消费者引发SynchronousOnlyOperation异常

问题描述

目前，我有一个这样的异步使用者：

class AsyncDashConsumer(AsyncJsonWebsocketConsumer):

    async def connect(self):
        await self.accept()

    async def disconnect(self,code):
        await self.close()

    async def receive(self,text_data=None,bytes_data=None,**kwargs):
        print('>>>>>>Data received from client:',text_data)
        # get data
        # sendData.send(self.channel_name)
        sync_to_async(sendData.send(self.channel_name))
        print('>>>>>>>First Worker Start')
        sync_to_async(sendData2.send(self.channel_name))
        print('>>>>>>>Workers started')

    async def Dash_tester(self,event):
        await self.send(text_data=event['text'])

    async def Dash_tester2(self,event):
        await self.send(text_data=event['text'])

sendData中的async def receive函数是戏剧（芹菜替代品）任务：

@dramatiq.actor
def sendData(channelName):
    channel_layer = get_channel_layer()
    for i in range(20):
        async_to_sync(channel_layer.send)(channelName,{'type': 'Dash.tester','text': f'FirstFunction{i}'})


@dramatiq.actor
def sendData2(channelName):
    channel_layer = get_channel_layer()
    for i in range(20):
        async_to_sync(channel_layer.send)(channelName,{'type': 'Dash.tester2','text': f'SecondFunction: {i}'})

---

我的问题是 sync_to_async(sendData.send(self.channel_name))部分。当我尝试使用所有这些方法时，客户端浏览器成功获取了来自Dramatiq任务的文本字符串，但是会出现此错误：

>>>>>>Data received from client: WEBSOCKET OPEN
Unexpected failure in after_enqueue.
Traceback (most recent call last):
  File "C:\Users\Timothee Legros\PycharmProjects\QuadkoRepository\Venv\lib\site-packages\dramatiq\broker.py",line 98,in emit_after
    getattr(middleware,"after_" + signal)(self,*args,**kwargs)
  File "C:\Users\Timothee Legros\PycharmProjects\QuadkoRepository\Venv\lib\site-packages\django_dramatiq\middleware.py",line 21,in after_enqueue
    Task.tasks.create_or_update_from_message(
  File "C:\Users\Timothee Legros\PycharmProjects\QuadkoRepository\Venv\lib\site-packages\django_dramatiq\models.py",line 16,in create_or_update_from_message
    task,_ = self.using(DATABASE_LABEL).update_or_create(
  File "C:\Users\Timothee Legros\PycharmProjects\QuadkoRepository\Venv\lib\site-packages\django\db\models\query.py",line 587,in update_or_create
    with transaction.atomic(using=self.db):
  File "C:\Users\Timothee Legros\PycharmProjects\QuadkoRepository\Venv\lib\site-packages\django\db\transaction.py",line 175,in __enter__
    if not connection.get_autocommit():
  File "C:\Users\Timothee Legros\PycharmProjects\QuadkoRepository\Venv\lib\site-packages\django\db\backends\base\base.py",line 389,in get_autocommit
    self.ensure_connection()
  File "C:\Users\Timothee Legros\PycharmProjects\QuadkoRepository\Venv\lib\site-packages\django\utils\asyncio.py",line 24,in inner
    raise SynchronousOnlyOperation(message)
django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
>>>>>>>First Worker Start
Unexpected failure in after_enqueue.
Traceback (most recent call last):
  File "C:\Users\Timothee Legros\PycharmProjects\QuadkoRepository\Venv\lib\site-packages\dramatiq\broker.py",in inner
    raise SynchronousOnlyOperation(message)
django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.
>>>>>>>Workers started

我认为将sync_to_async包装器添加到sendData.send函数中可以消除此错误，但不会。我尝试了讨论该问题的其他线程上提供的解决方案，但没有一个可以阻止错误的发生。我对python asyncio有非常基本的了解，如果缺少简单的内容，对不起。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）