LINUX.ORG.RU

Как прекратить job (полностью вывалиться из него)?

 ,


0

3

Доброго вечера.

Достаточно глупый вопрос, даже стыдно.

Есть некая очередь с консьюмерами. В процессе обработки таска проводится проверка на соответствие некоторым условиям. В случае соответствия одному из них - дальнейшую обработку таска прерываем.

class Example:
    def __init__(self):
        pass

    async def consumer(self, queue):
        while True:
            i = await queue.get()
            k_list = [1, 2, 3]
            for k in k_list:
                if i == k:
                    queue.task_done()
                    # Как прекратить дальнейшую обработку этого таска?
                    
            await asyncio.sleep(3)  # Псевдонагрузка
            queue.task_done()

    async def producer(self, queue):
        for i in range(0, 100):
            await queue.put(i)

    async def run(self):
        queue = asyncio.Queue()
        consumers = [asyncio.ensure_future(self.consumer(queue)) for _ in range(10)]
        producer = await self.producer(queue)
        await queue.join()

        for consumer_future in consumers:
            consumer_future.cancel()

    def start(self):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(self.run())

if __name__ == '__main__':
    Example = Example()
    Example.start()

Сложность в блоке for. Как в окружении asyncio правильнее прервать дальнейшую обработку? break\continue не подходят, т.к. из while'а не выходим, return же при такой реализации подвесит весь поток.

Где я не там свернул?

Deleted

break\continue не подходят, т.к. из while'а не выходим

Обычно в языке есть способ применить break/continue для любого уровня цикла. Если его нет - тогда тело while сделать отдельной функцией и из нее делать return. Можно еще флаг сделать вместе с break из for и проверять его после.

loz ★★★★★ ()
Последнее исправление: loz (всего исправлений: 2)
Ответ на: комментарий от urxvt

Я вот тоже его увидел, но так и не понял почему его отвергли.

Понятно, что вариант через переменную-«триггер» или exсeption никто не отменял, но должно же быть и какое-то красивое решение.

Deleted ()
Ответ на: комментарий от loz

Пока реализовал проверку через функцию с return True\False.

Но не уверен, что для питона это наиболее красивое решение. Плодить отдельную трехстрочную функцию как уж слишком брутально выглядит для этого языка :)

Deleted ()
async def consumer(self, queue):
    while True:
        i = await queue.get()
        k_list = [1, 2, 3]
        for k in k_list:
            if i == k:
                queue.task_done()
                # Как прекратить дальнейшую обработку этого таска?
                break
        else:
            await asyncio.sleep(3)  # Псевдонагрузка
            queue.task_done()
            continue
        break

Как-то так, если я правильно понял блок-схему.

Virtuos86 ★★★★★ ()
Ответ на: комментарий от vvn_black

Через сокет работать в моё случае сложнее, т.к. скрипт будет запускаться не только в linux-среде. Точнее мои навыки пока не насколько хороши, чтобы воспользоваться такой схемой :)

Deleted ()