asyncio aiohttp限制並發數量和超時時間

轉載：https://www.jianshu.com/p/6f8980cf0948

主要參考參數設置的一些問題

import asyncio
import random
import traceback

from aiohttp import ClientSession, TCPConnector, client_exceptions
import time

URL = 'http://127.0.0.1:5000/?delay={}'


async def fetch(session, i):
    dly = random.randint(1,8)
    url = URL.format(dly)
    start_time = time.time()
    try:
        async with session.get(url=url) as response:
            r = await response.read()
            end_time = time.time()
            cost = end_time - start_time
            msg = "第{}個查詢請求，花費時間: {}s, 返回信息: {}\n".format(i, cost, r.decode('unicode-escape'))
            print("running %d" % i, msg)
    except client_exceptions.ServerTimeoutError as timeout_error:
        print("request timeout error: {}, url: {}".format(timeout_error, url))
    except Exception as e:
        print("request unknown error: {}".format(traceback.format_exc()))


async def chunks(sem, session, i):
    """
    限制並發數
    """
    # 使用Semaphore， 它會在第一批400個請求發出且返回結果(是否等待返回結果取決於你的fetch方法的定義)后
    # 檢查本地TCP連接池(最大400個)的空閑數(連接池某個插槽是否空閑，在這里，取決於請求是否返回)
    # 有空閑插槽，就PUT入一個請求並發出(完全不同於Jmeter的rame up in period的線性發起機制).
    # 所以，在結果log里，你會看到第一批請求(開始時間)是同一秒發起，而后面的則完全取決於服務器的吞吐量
    async with sem:
        await fetch(session, i)


async def run(num):
    tasks = []
    # Semaphore， 相當於基於服務器的處理速度和測試客戶端的硬件條件，一批批的發
    # 直至發送完全部（下面定義的400）
    sem = asyncio.Semaphore(400)
    # 創建session，且對本地的TCP連接做限制limit=400（不做限制limit=0）
    # 超時時間指定
    # total:全部請求最終完成時間
    # connect: aiohttp從本機連接池里取出一個將要進行的請求的時間
    # sock_connect：單個請求連接到服務器的時間
    # sock_read：單個請求從服務器返回的時間
    import aiohttp
    timeout = aiohttp.ClientTimeout(total=330, connect=2, sock_connect=15, sock_read=10)
    async with ClientSession(connector=TCPConnector(limit=400), timeout=timeout) as session:
        for i in range(0, num):
            # 如果是分批的發，就使用並傳遞Semaphore
            task = asyncio.ensure_future(
                chunks(sem, session, i))
            tasks.append(task)
        responses = asyncio.gather(*tasks)
        await responses


start = time.time()
number = 380
loop = asyncio.get_event_loop()
future = asyncio.ensure_future(run(number))
loop.run_until_complete(future)
end = time.time()
total = end - start
with open("log", "a+", encoding="utf-8") as f:
    f.write('總耗時：{}秒,平均速度：{}秒\n'.format(total, total / number))

 更新：

如果超時，limit=400，驗證第一次同時發起400個請求

import time
import queue
import random
import asyncio
import traceback
import collections

from aiohttp import ClientSession, TCPConnector, client_exceptions, ClientTimeout

queue_data = queue.Queue()
timeout_domains = []
unknown_error_domains = []
start_time_list = []


async def fetch(session, n, url):
    """
    :param session:  aiohttp.ClientSession
    :param n: task編號
    :param url: 請求url
    """
    start_time = time.time()
    # noinspection PyBroadException
    try:
        async with session.get(url=url, verify_ssl=False) as response:
            r = await response.read()
            end_time = time.time()
            cost = end_time - start_time
            msg = "第{}個查詢請求，花費時間: {}s, 返回信息: {}\n".format(n, cost, r.decode('unicode-escape'))
            # print(msg)
            queue_data.put(1)
    except client_exceptions.ServerTimeoutError as timeout_error:
        print("request timeout error: {}, url: {}".format(timeout_error, url))
        timeout_domains.append(url)
    except Exception:
        print("request unknown error: {}".format(traceback.format_exc()))
        unknown_error_domains.append(url)
    start_time_list.append(str(start_time).split(".")[0])


async def chunks(sem, session, i, url):
    """
    限制並發數
    """
    async with sem:
        await fetch(session, i, url)


def get_domains():
    urls = []
    for _ in range(1000):
        urls.append("http://127.0.0.1:5000/?delay={}".format(random.randint(1, 8)))
    return urls


async def main(urls):
    sem = asyncio.Semaphore(400)
    timeout = ClientTimeout(total=10, connect=2, sock_connect=15, sock_read=5)
    async with ClientSession(connector=TCPConnector(limit=400), timeout=timeout) as session:
        tasks = [asyncio.create_task(chunks(sem, session, index, url)) for index, url in enumerate(urls)]
        await asyncio.wait(tasks)


if __name__ == '__main__':
    domains = get_domains()
    asyncio.run(main(domains))
    print("success number: {}, timeout number: {}, unknown_error number: {}".format(queue_data.qsize(),
                                                                                    len(timeout_domains),
                                                                                    len(unknown_error_domains)))

    print(sorted(collections.Counter(start_time_list).items(), key=lambda item:item[0]))
    # 1. 沒有超時的，第一批400個同一秒發起， 再往后就看response相應與讀取速度
    # success number: 1000, timeout number: 0, unknown_error number: 0
    # [('1593246892', 400), ('1593246894', 48), ('1593246895', 55), ('1593246896', 55), ('1593246897', 76),
    #  ('1593246898', 74), ('1593246899', 90), ('1593246900', 96), ('1593246901', 106)]
    # 2. 有超時的
    # success number: 517, timeout number: 483, unknown_error number: 0
    # [('1593248067', 400), ('1593248068', 36), ('1593248069', 43), ('1593248070', 75), ('1593248071', 64),
    # ('1593248072', 168), ('1593248073', 126), ('1593248074', 69), ('1593248075', 19)]