asyncio 队列(Queue)

新的启动事件循环

# coding=utf-8
import asyncio
import random
from contextlib import closing

import aiohttp

NUMBERS = random.sample(range(100), 7)
URL = 'http://httpbin.org/get?a={}'
semaphore = asyncio.Semaphore(3)


async def fetch_async(a):
    async with aiohttp.request('GET', URL.format(a)) as r:
        data = await r.json( )
        return data['args']['a']


async def collect_result(a):
    async with semaphore:
        return await fetch_async(a)


# 生产者
async def produce(queue):
    for num in NUMBERS:
        print(f'生产： {num}')
        item = (num, num)
        await queue.put(item)


# 消费者
async def consume(queue):
    while True:
        item = await queue.get( )
        num = item[0]  # 因为存放的时候是元组，就取第一个数
        rs = await collect_result(num)
        print(f'\t 消费： {rs}...')
        queue.task_done( )


async def main( ):
    queue = asyncio.PriorityQueue( )
    consumer = asyncio.ensure_future(consume(queue))
    await produce(queue)
    # 阻塞并关闭------------------------------------
    await queue.join( )
    consumer.cancel( )


if __name__ == '__main__':
    # 修改了这里
    with closing(asyncio.get_event_loop( )) as loop:
        loop.run_until_complete(main( ))

旧的启动事件循环

# coding=utf-8
import asyncio
import random
import aiohttp

NUMBERS = random.sample(range(100), 7)
URL = 'http://httpbin.org/get?a={}'
semaphore = asyncio.Semaphore(3)


async def fetch_async (a):
    async with aiohttp.request('GET', URL.format(a)) as r:
        data = await r.json()
        return data['args']['a']


async def collect_result (a):
    async with semaphore:
        return await fetch_async(a)


# 生产者
async def produce (queue):
    for num in NUMBERS:
        print(f'生产： {num}')
        item = (num, num)
        await queue.put(item)


# 消费者
async def consume (queue):
    while True:
        item = await queue.get()
        num = item[0]  # 因为存放的时候是元组，就取第一个数
        rs = await collect_result(num)
        print(f'\t 消费： {rs}...')
        queue.task_done()


async def main (loop):
    queue = asyncio.PriorityQueue()
    consumer = asyncio.ensure_future(consume(queue))
    await produce(queue)
    # 阻塞并关闭------------------------------------
    await queue.join()
    consumer.cancel()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    # ---------------------------------------------
    try:
        loop.run_until_complete(main(loop))
    finally:
        loop.close()

输出：

生产： 3
生产： 53
生产： 29
生产： 81
生产： 64
生产： 5
生产： 96
	 消费： 3...
	 消费： 5...
	 消费： 29...
	 消费： 53...
	 消费： 64...
	 消费： 81...
	 消费： 96...