Python - làm thế nào để chạy nhiều coroutines đồng thời sử dụng asyncio?

Question

Tôi đang sử dụng thư viện websockets để tạo máy chủ websocket trong Python 3.4. Đây là máy chủ echo đơn giản:

import asyncio 
import websockets 

@asyncio.coroutine 
def connection_handler(websocket, path): 
    while True: 
     msg = yield from websocket.recv() 
     if msg is None: # connection lost 
      break 
     yield from websocket.send(msg) 

start_server = websockets.serve(connection_handler, 'localhost', 8000) 
asyncio.get_event_loop().run_until_complete(start_server) 
asyncio.get_event_loop().run_forever() 

Giả sử chúng tôi - ngoài ra - muốn gửi tin nhắn cho khách hàng bất cứ khi nào có sự kiện xảy ra. Để đơn giản, hãy gửi tin nhắn định kỳ 60 giây một lần. Chúng ta nên làm việc đó như thế nào? Ý tôi là, vì connection_handler liên tục chờ tin nhắn đến, máy chủ chỉ có thể thực hiện hành động sau nó đã nhận được tin nhắn từ khách hàng, phải không? Tôi đang thiếu gì ở đây?

Có thể kịch bản này yêu cầu khung dựa trên sự kiện/callbacks thay vì dựa trên coroutines? Tornado?

Answer 1

TL; DR Sử dụng asyncio.ensure_future() để chạy đồng thời nhiều coroutin.

---

Có lẽ kịch bản này đòi hỏi một khuôn khổ dựa trên các sự kiện/callbacks chứ không phải là một dựa trên coroutines? Vòi rồng?

Không, bạn không cần bất kỳ khung công tác nào khác cho việc này. Toàn bộ ý tưởng ứng dụng không đồng bộ so với đồng bộ là nó không chặn, trong khi chờ kết quả. Nó không quan trọng như thế nào nó được thực hiện, sử dụng coroutines hoặc callbacks.

Ý tôi là, vì connection_handler liên tục chờ tin nhắn đến, máy chủ chỉ có thể thực hiện hành động sau khi nhận được tin nhắn từ khách hàng, phải không? Tôi đang thiếu gì ở đây?

Trong ứng dụng đồng bộ, bạn sẽ viết một cái gì đó như msg = websocket.recv(), sẽ chặn toàn bộ ứng dụng cho đến khi bạn nhận được tin nhắn (như bạn mô tả). Nhưng trong ứng dụng không đồng bộ, nó hoàn toàn khác.

Khi bạn làm msg = yield from websocket.recv() bạn nói điều gì đó như: tạm ngừng thực hiện connection_handler() cho đến khi websocket.recv() sẽ tạo ra thứ gì đó. Sử dụng yield from bên trong coroutine trả về kiểm soát trở lại vòng lặp sự kiện, do đó, một số mã khác có thể được thực hiện, trong khi chúng tôi đang chờ kết quả của websocket.recv(). Vui lòng tham khảo documentation để hiểu rõ hơn cách thức hoạt động của coroutines.

Giả sử chúng tôi - ngoài ra - muốn gửi thư cho khách hàng bất cứ khi nào có sự kiện xảy ra. Để đơn giản, hãy gửi tin nhắn định kỳ 60 giây một lần. Chúng ta nên làm việc đó như thế nào?

Bạn có thể sử dụng asyncio.async() để chạy nhiều coroutines tùy ý, trước khi thực hiện cuộc gọi chặn cho starting event loop.

import asyncio 

import websockets 

# here we'll store all active connections to use for sending periodic messages 
connections = [] 


@asyncio.coroutine 
def connection_handler(connection, path): 
    connections.append(connection) # add connection to pool 
    while True: 
     msg = yield from connection.recv() 
     if msg is None: # connection lost 
      connections.remove(connection) # remove connection from pool, when client disconnects 
      break 
     else: 
      print('< {}'.format(msg)) 
     yield from connection.send(msg) 
     print('> {}'.format(msg)) 


@asyncio.coroutine 
def send_periodically(): 
    while True: 
     yield from asyncio.sleep(5) # switch to other code and continue execution in 5 seconds 
     for connection in connections: 
      print('> Periodic event happened.') 
      yield from connection.send('Periodic event happened.') # send message to each connected client 


start_server = websockets.serve(connection_handler, 'localhost', 8000) 
asyncio.get_event_loop().run_until_complete(start_server) 
asyncio.async(send_periodically()) # before blocking call we schedule our coroutine for sending periodic messages 
asyncio.get_event_loop().run_forever() 

Dưới đây là ví dụ về triển khai ứng dụng khách. Nó yêu cầu bạn nhập tên, nhận lại nó từ máy chủ echo, đợi thêm hai tin nhắn từ máy chủ (đó là các tin nhắn định kỳ của chúng tôi) và đóng kết nối.

import asyncio 

import websockets 


@asyncio.coroutine 
def hello(): 
    connection = yield from websockets.connect('ws://localhost:8000/') 
    name = input("What's your name? ") 
    yield from connection.send(name) 
    print("> {}".format(name)) 
    for _ in range(3): 
     msg = yield from connection.recv() 
     print("< {}".format(msg)) 

    yield from connection.close() 


asyncio.get_event_loop().run_until_complete(hello()) 

Những điểm quan trọng:

1. Trong Python 3.4.4 asyncio.async() được đổi tên thành asyncio.ensure_future().
2. Có các phương pháp đặc biệt để lên lịch delayed calls, nhưng chúng không hoạt động với coroutines.

Answer 2

Cùng một vấn đề, khó có thể có giải pháp cho đến khi tôi nhìn thấy những mẫu hoàn hảo ở đây: http://websockets.readthedocs.io/en/stable/intro.html#both

done, pending = await asyncio.wait(
     [listener_task, producer_task], 
     return_when=asyncio.FIRST_COMPLETED) # Important 

Vì vậy, tôi có thể xử lý các tác vụ đa coroutine như nhịp tim và redis đăng ký.