6

作成したサーバーに多くの 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())

問題は、すべてのクライアントが次々にジョブを実行することです。

0 Connecting
0 Connected
0 Sending the message
0 Received the echo
0 Closed
1 Connecting
1 Connected
1 Sending the message
1 Received the echo
1 Closed
...

を使用しようとしましasyncio.waitたが、すべてのクライアントが一緒に起動します。それらを徐々に作成し、それぞれが作成されたらすぐにサーバーに接続したいと思います。同時に、新しいクライアントを作成し続けます。

これを達成するには、どのようなアプローチを適用する必要がありますか?

4

1 に答える 1

9

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あなたが望むことをするクラスを実装しました。このセットは先物で埋めることができ、それらが完了すると自動的に削除されます。すべての先物が完了するのを待つこともできます。

class FutureSet:

    def __init__(self, maxsize, *, loop=None):
        self._set = set()
        self._loop = loop
        self._maxsize = maxsize
        self._waiters = []

    @asyncio.coroutine
    def add(self, item):
        if not asyncio.iscoroutine(item) and \
           not isinstance(item, asyncio.Future):
            raise ValueError('Expecting a coroutine or a Future')
        if item in self._set:
            return
        while len(self._set) >= self._maxsize:
            waiter = asyncio.Future(loop=self._loop)
            self._waiters.append(waiter)
            yield from waiter
        item = asyncio.async(item, loop=self._loop)    
        self._set.add(item)
        item.add_done_callback(self._remove)

    def _remove(self, item):
        if not item.done():
            raise ValueError('Cannot remove a pending Future')
        self._set.remove(item)
        if self._waiters:
            waiter = self._waiters.pop(0)
            waiter.set_result(None)

    @asyncio.coroutine
    def wait(self):
        return asyncio.wait(self._set)

例：

@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()
于 2015-10-16T14:35:44.447 に答える