0

Как работает Protocol в asyncio при каждом подключении клиента создается новая таска? нужно ли мне например писать в подключении к БД asyncio.create_task()? Или это уже и будет таска?

2
  • "По сути, транспорты и протоколы модуля asyncio следует использовать только в библиотеках и фреймворках, а не в высокоуровневых асинхронных приложениях." - docs-python.ru/standart-library/modul-asyncio-python/… Commented 12 окт. 2022 в 5:19
  • А в клиентских приложениях socket.accept =)
    – eri
    Commented 15 окт. 2022 в 8:46

1 ответ 1

0

Хоть протокол и выполняется в таске в классе доступны только обычные колбэки. Выполнение асинхронного кода возможно в 2х вариантах.

Подключение к базе данных передается в протокол через init, а вот как его использовать ниже.

Самый простой создавать таск при получении данных:

class EchoServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
    
    async def data_received_async(self, data):
        await asyncio.sleep(1)
        self.transport.write(data)

    def data_received(self, data):
        asyncio.create_task(self.data_received_async(data))

Подходит на протоколы "запрос-ответ",например, RADIUS, DNS, NTP, HTTP. В случае UDP сойдет, в случае стрима в синхронном коллбэке нужно собрать буфер по протоколу уровня приложения и уже передавать в обработку.

Посложнее - создавать таск на конекшн.

Пример с обработкой в очереди. Тут уже можно слать клиенту всякое в независимости от запросов.

import asyncio

class EchoServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        
        self.queue = asyncio.Queue()
        self.task = asyncio.create_task(self.connection_task(transport))
        self.ping_task = asyncio.create_task(self.ping(transport))
        

    def connection_lost(self, exc):
        print('The server closed the connection')
        self.ping_task.cancel()
        if not self.task.done():
            self.task.cancel()

    async def ping(self,transport):
        while True:
            await asyncio.sleep(1)
            transport.write(b"ping\n")

    async def connection_task(self, transport):
        transport.write(b"hello\n")
        while True:
            message = await self.queue.get()
            await asyncio.sleep(5)
            transport.write(message.upper().encode())
            self.queue.task_done()
        transport.close()

    def data_received(self, data):
        message = data.decode()
        self.queue.put_nowait(message)



async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    server = await loop.create_server(
        lambda: EchoServerProtocol(),
        '127.0.0.1', 8888)

    async with server:
        await server.serve_forever()
   
asyncio.run(main())

Таск нужно вытащить наружу если асинхронный ресурс не поддерживает обработку нескольких запросов.

Например COM порт или железяка.

3
  • Спасибо, а как еще получить список всех подключенных клиентов? Ведь на каждый коннект создается новый инстанс протокола
    – Injustive
    Commented 18 окт. 2022 в 9:20
  • ещё они переиспользуются. если конект закрыть и снова открыть - можно попасть в старый инстанс.) список всех клиентов храни во внешней переменной отлавливая по connection_made
    – eri
    Commented 18 окт. 2022 в 9:42
  • спасибо большое) а есть какая-то инфа почитать про эти протоколы и транспорты, а то гуглил, очень мало инфы на эту тему, только дока
    – Injustive
    Commented 18 окт. 2022 в 9:59

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.