0

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

2
  • "По сути, транспорты и протоколы модуля asyncio следует использовать только в библиотеках и фреймворках, а не в высокоуровневых асинхронных приложениях." - docs-python.ru/standart-library/modul-asyncio-python/… 12 окт 2022 в 5:19
  • А в клиентских приложениях socket.accept =)
    – eri
    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
    18 окт 2022 в 9:20
  • ещё они переиспользуются. если конект закрыть и снова открыть - можно попасть в старый инстанс.) список всех клиентов храни во внешней переменной отлавливая по connection_made
    – eri
    18 окт 2022 в 9:42
  • спасибо большое) а есть какая-то инфа почитать про эти протоколы и транспорты, а то гуглил, очень мало инфы на эту тему, только дока
    – Injustive
    18 окт 2022 в 9:59

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

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