使用协程实现调用接口 验证抽奖概率
原文地址
在原先的基础上使用协程进行改进，提高效率
- 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()
- 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
- 任务列表：tasks 列表包含了 n 个 fetch 任务。
- 解析响应：所有响应返回后，逐个解析 JSON 数据并更新结果字典 d。
- 运行主协程：使用 asyncio.get_event_loop() 和 loop.run_until_complete(main(…)) 启动和运行主协程，获取最终结果。
if __name__ == '__main__':url = ''headers = ''payload = ''n = 100000delay = 0.1 # 单位是sloop = asyncio.get_event_loop()result = loop.run_until_complete(main(url, headers, payload, n, delay))get_lottery(result)加锁
由于是对同一个字典进行操作，需要保证操作的线程安全性，保证数据一致性，需要加锁处理 ，使用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] += 1def get_lottery(d):for key, value in d.items():lottery = value / nprint(key + "的概率是：" + '{:.2%}'.format(lottery))