django 通道 asyncio 无法按顺序运行任务

Question

我正在努力让一些代码使用asyncio 工作。我对它很陌生。对它如此陌生，以至于我不知道如何正确地找出我做错了什么。

我正在使用 Django Channels 运行 AyncJsonWebsocketConsumer，我通过客户端应用程序中的 websockets 连接到该 AyncJsonWebsocketConsumer。我使用 websockets 因为我需要双向通信。我正在创建一个打印过程，在该过程中我开始一系列长时间运行的操作，但我需要暂停、停止等功能。当我使用asyncio.sleep(x) 模拟我的长时间运行的任务（@ 987654324@ 方法）。当我尝试将我的 RPC 添加到队列中而不是 asyncio.sleep 时，它停止按预期工作。

class Command(Enum):
    START = 'start_print'
    PAUSE = 'pause_print'
    STOP = 'stop_print'
    CANCEL = 'cancel_print'
    RESET = 'reset_print'

class State(Enum):
    NOT_STARTED = 0
    STARTING = 1
    RUNNING = 2
    PAUSED = 3
    STOPPED = 4
    COMPLETE = 5
    ERROR = 500

class PrintConsumer(AsyncJsonWebsocketConsumer):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.status = State.NOT_STARTED
        self.publisher = Publisher()
        self.print_instruction = None
        self.current_task = None
        self.loop = asyncio.get_event_loop()
        self.loop.set_debug(enabled=True)

    @property
    def running_task(self):
        return self.current_task is not None and not self.current_task.done() and not self.current_task.cancelled()

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

    async def disconnect(self, close_code):
        self.status = State.STOPPED

    async def receive(self, text_data):
        response = json.loads(text_data)
        event_cmd = response.get('event', None)
        
        if Command(event_cmd) == Command.START:
            if self.status == State.NOT_STARTED:
                sting_uuid = response.get('sting_uuid', None)
                try:
                    await asyncio.wait_for(self.initialize_print(sting_uuid), timeout=10)  # WARNING: this is blocking
                except asyncio.TimeoutError:
                    self.status = State.ERROR
            self.status = State.RUNNING
            if not self.running_task:
                # it would only have a running task in this situation already if someone starts/stops it quickly
                self.current_task = asyncio.create_task(self.resume_print())
        elif Command(event_cmd) == Command.PAUSE:
            self.status = State.PAUSED
        elif Command(event_cmd) == Command.STOP:
            if self.running_task:
                self.current_task.cancel()
        elif Command(event_cmd) == Command.RESET:
            self.status = State.NOT_STARTED
            self.print_instruction = None
            if self.running_task:
                self.current_task.cancel()
            self.current_task = None
            await self.send(json.dumps({ 'message': []}))
        
    async def initialize_print(self, uuid):
        stingfile = await get_file(uuid)
        # This is just an iterator that returns the next task to do
        # hence I use "next" in the resume_print method
        self.print_instruction = StingInstruction(stingfile)


    async def resume_print(self):
        try:
            while self.status == State.RUNNING:
                await self.send(json.dumps({ 'message': self.print_instruction.serialized_action_status}))
                await asyncio.sleep(.2) # It works with this here only
                try:
                    action = next(self.print_instruction)  # step through iterator
                except StopIteration:
                    self.status = State.COMPLETE
                    break;

                # can't seem to get this part to work
                await self.print_layer(action)
        except asyncio.CancelledError:
            # TODO: Add logic here that will send a command out to stop pumps and all motors.
            self.status = State.STOPPED
    
    async def print_layer(self, instruction):
        print_command = instruction['instruction']
        # this publishes using RPC, so it holds until I get a response.
        try:
            await asyncio.wait_for(self.publisher.publish('pumps', json.dumps(print_command)), timeout=10)
        except asyncio.TimeoutError:
            self.status = State.ERROR
        # when I used just this in the function, the code worked as expected
        # await asyncio.sleep(1)

在展示我尝试过的内容时，我不知道从哪里开始...在我看来，我的“最佳”尝试是将print_layer 方法变成一个线程，这样它就不会阻塞执行使用asyncio.to_thread(print_layer).. 但在我尝试的许多事情中，它甚至都不会执行。

self.print_instruction.serialized_action_status 返回每​​个步骤的状态。我的目标是让它在每个长时间运行的任务之前发送它。这可能看起来像...

# sending status update for each step to client
# running print_layer for first action
# sending status update for each step to client
# running print_layer for second action
...
# sending final update

相反，我一次创建每一个任务，当我添加长时间运行的任务或许多问题时，它会在最后发送所有更新。我可以让长时间运行的任务按顺序运行（看似），但send 实际上不会在层间打印。非常感谢您的帮助.. 提前谢谢您。

这是我的发布者的一些简化的相关代码（不处理连接丢失等）...

class Publisher():
    def on_response(self, ch, method, props, body):
        """when job response set job to inactive"""

    async def publish(routing_key, msg):
        new_corr_id = str(uuid4())
        self.active_jobs[new_corr_id] = False
        self.channel.basic_publish(...)
        white not self.active_jobs[new_corr_id]:
            self._connection.process_data_events()
            sleep(.1)

---

我发现了一个部分有效的 hack.. 如果我添加 await asyncio.sleep(.1) 在我的发送命令之后（即像这样）

await self.send(json.dumps({ 'message': self.print_instruction.serialized_action_status}))
await asyncio.sleep(.2)

然后它似乎按我想要的方式工作（减去中断能力），我仍然能够暂停/启动我的进程。显然，我宁愿在没有 hack 的情况下这样做。为什么这段代码会在.2 asyncio sleep 之后按预期发送状态更新而不是没有？我也不能用我不明白的 STOP 命令打断。我本来希望 django 频道读取停止命令，然后取消正在运行的任务并在 resume_print 方法中强制使用 asyncio.CancelledError。

Answer 1

我能够弄清楚我做错了什么。

我一直假设因为我将一个函数设为协程（将async 放在它前面），所以偶数循环会在执行 io 操作时自动等待。绝对不是这样。就我而言，这是在这里：

class Publisher():
    def on_response(self, ch, method, props, body):
        """when job response set job to inactive"""

    async def publish(routing_key, msg):
        new_corr_id = str(uuid4())
        self.active_jobs[new_corr_id] = False
        self.channel.basic_publish(...)
        white not self.active_jobs[new_corr_id]:
            self._connection.process_data_events()
            sleep(.1)

在publish 方法中。此示例使用pika 并使用basic_publish 进行直接交换。这不是异步的，并且阻塞了事件循环。我将我的发布者改为使用aio-pika，然后一旦我的publish 命令异步，事件循环不再被阻塞，代码按预期工作。

Answer 2

我将从更改代码中的一件事开始：

    if Command(event_cmd) == Command.START:
        if self.status == State.NOT_STARTED:
            sting_uuid = response.get('sting_uuid', None)
            try:
                await asyncio.wait_for(self.initialize_print(sting_uuid), timeout=10)  # WARNING: this is blocking
            except asyncio.TimeoutError:
                self.status = State.ERROR
        self.status = State.RUNNING
        #--> This will always set it to running, as it is outside of the if statement. That will stop your stop problem
        #You might want to do this (see https://www.w3schools.com/python/python_try_except.asp) : 
            else:
                self.status = State.RUNNING
        if not self.running_task:
            # it would only have a running task in this situation already if someone starts/stops it quickly
            self.current_task = asyncio.create_task(self.resume_print())
    elif Command(event_cmd) == Command.PAUSE:
        self.status = State.PAUSED
    elif Command(event_cmd) == Command.STOP:
        if self.running_task:
            self.current_task.cancel()
    elif Command(event_cmd) == Command.RESET:
        self.status = State.NOT_STARTED
        self.print_instruction = None
        if self.running_task:
            self.current_task.cancel()
        self.current_task = None
        await self.send(json.dumps({ 'message': []}))
    

此外，你能用你正在做的导入来完成它吗？

你看了吗：

https://docs.python.org/3/library/asyncio-task.html

这可能会帮助您根据需要获得发送选项。