逐步创建异步任务，等待所有任务完成

Question

我正在尝试制作一个程序，使许多web套接字连接到我创建的服务器：

class WebSocketClient():

    @asyncio.coroutine
    def run(self):
        print(self.client_id, 'Connecting')
        ws = yield from aiohttp.ws_connect(self.url)
        print(self.client_id, 'Connected')
        print(self.client_id, 'Sending the message')
        ws.send_str(self.make_new_message())

        while not ws.closed:
            msg = yield from ws.receive()

            if msg.tp == aiohttp.MsgType.text:
                print(self.client_id, 'Received the echo')
                yield from ws.close()
                break

        print(self.client_id, 'Closed')


@asyncio.coroutine
def make_clients():

    for client_id in range(args.clients):
        yield from WebSocketClient(client_id, WS_CHANNEL_URL.format(client_id=client_id)).run()


event_loop.run_until_complete(make_clients())

问题是所有的客户都在一个接一个地完成他们的工作：

^{pr2}$

我尝试过使用^{}，但所有客户机都是一起启动的。我希望它们逐步创建，并在每个创建完成后立即连接到服务器。同时继续创建新客户。在

我应该采用什么方法来实现这一点？在

Answer 1

使用asyncio.wait是一个很好的方法。您可以将其与asyncio.ensure_future和asyncio.sleep组合以逐步创建任务：

@asyncio.coroutine
def make_clients(nb_clients, delay):
    futures = []
    for client_id in range(nb_clients):
        url = WS_CHANNEL_URL.format(client_id=client_id)
        coro = WebSocketClient(client_id, url).run()
        futures.append(asyncio.ensure_future(coro))
        yield from asyncio.sleep(delay)
    yield from asyncio.wait(futures)

---

EDIT：我实现了一个FutureSet类，它应该做你想要的。这个集合可以填充futures并在完成后自动删除它们。也可以等待所有的未来完成。在

^{pr2}$

示例：

@asyncio.coroutine
def make_clients(nb_clients, limit=0):
    futures = FutureSet(maxsize=limit)
    for client_id in range(nb_clients):
        url = WS_CHANNEL_URL.format(client_id=client_id)
        client = WebSocketClient(client_id, url)
        yield from futures.add(client.run())
    yield from futures.wait()