Как работает Protocol в asyncio при каждом подключении клиента создается новая таска? нужно ли мне например писать в подключении к БД asyncio.create_task()? Или это уже и будет таска?
-
"По сути, транспорты и протоколы модуля asyncio следует использовать только в библиотеках и фреймворках, а не в высокоуровневых асинхронных приложениях." - docs-python.ru/standart-library/modul-asyncio-python/…– ВикторCommented 12 окт. 2022 в 5:19
-
А в клиентских приложениях socket.accept =)– eriCommented 15 окт. 2022 в 8:46
Добавить комментарий
|
1 ответ
Хоть протокол и выполняется в таске в классе доступны только обычные колбэки. Выполнение асинхронного кода возможно в 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 порт или железяка.
-
Спасибо, а как еще получить список всех подключенных клиентов? Ведь на каждый коннект создается новый инстанс протокола Commented 18 окт. 2022 в 9:20
-
ещё они переиспользуются. если конект закрыть и снова открыть - можно попасть в старый инстанс.) список всех клиентов храни во внешней переменной отлавливая по connection_made– eriCommented 18 окт. 2022 в 9:42
-
спасибо большое) а есть какая-то инфа почитать про эти протоколы и транспорты, а то гуглил, очень мало инфы на эту тему, только дока Commented 18 окт. 2022 в 9:59