使用协程实现调用接口 验证抽奖概率

原文地址
在原先的基础上使用协程进行改进，提高效率

• fetch 函数：用于发送请求并返回响应文本。
  delay 用于指定延迟时间

async def fetch(session, url, headers, payload, delay=0):
    await asyncio.sleep(delay)
    async with session.post(url, headers=headers, data=payload, ssl=False) as response:
        return await response.text()
1
2
3
4

• main 函数：创建一个 aiohttp.ClientSession 会话，并启动多个并发的 fetch 任务。asyncio.gather 用于等待所有任务完成，并收集它们的结果。

async def main(url, headers, payload, n, delay):
    d = defaultdict(int)
    lock = asyncio.Lock()  # 创建锁

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_and_update(session, url, headers, payload, d, lock, delay) for _ in range(n)]
        await asyncio.gather(*tasks)

    return d
1
2
3
4
5
6
7
8
9

• 任务列表：tasks 列表包含了 n 个 fetch 任务。
• 解析响应：所有响应返回后，逐个解析 JSON 数据并更新结果字典 d。
• 运行主协程：使用 asyncio.get_event_loop() 和 loop.run_until_complete(main(…)) 启动和运行主协程，获取最终结果。

if __name__ == '__main__':
	url = ''
	headers = ''
	payload = ''
	n = 100000
	delay = 0.1 # 单位是s
	loop = asyncio.get_event_loop()
	result = loop.run_until_complete(main(url, headers, payload, n, delay))
	get_lottery(result)
	
1
2
3
4
5
6
7
8
9
10

加锁

由于是对同一个字典进行操作，需要保证操作的线程安全性，保证数据一致性，需要加锁处理 ，使用asyncio.Lock

async def fetch_and_update(session, url, headers, payload, d, lock, delay):
    response = await fetch(session, url, headers, payload, delay)
    jsonobj = json.loads(response)
    reward_name = jsonobj['']

    async with lock:  # 确保对字典的访问是线程安全的
        d[reward_name] += 1


def get_lottery(d):
    for key, value in d.items():
        lottery = value / n
        print(key + "的概率是：" + '{:.2%}'.format(lottery))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15