实例 动态添加任务

效果, 多了会卡死

  1. # -*- coding: utf-8 -*-
  3. """
  4. @author: chenchen
  5. @software: PyCharm
  6. @file: task_pool.py
  7. @功能: 
  8. @time: 2022-09-18 20:30
  9. """
  10. """
  11. asyncio 协程介绍:
  12.     - 动态添加任务：
  13.         - 方案是创建一个线程，使事件循环在线程内永久运行
  14.         - 设置守护进程，随着主进程一起关闭
  15.     - 自动停止任务
  16.     - 阻塞任务完成
  17.     - 协程池
  18.         - asyncio.Semaphore() 进行控制
  19. """
  21. import asyncio
  22. import aiohttp
  23. import time
  24. import queue
  25. from threading import Thread
  28. class AsyncPool(object):
  29.     """
  30.     1. 支持动态添加任务
  31.     2. 支持自动停止事件循环
  32.     3. 支持最大协程数
  33.     """
  35.     def __init__(self, maxsize=1, loop=None):
  36.         """
  37.         初始化
  38.         :param loop:
  39.         :param maxsize: 默认为1
  40.         """
  41.         # 在jupyter需要这个，不然asyncio运行出错
  42.         # import nest_asyncio
  43.         # nest_asyncio.apply()
  45.         # 队列，先进先出，根据队列是否为空判断，退出协程
  46.         self.task = queue.Queue()
  48.         # 协程池
  49.         self.loop, _ = self.start_loop(loop)
  50.         # 限制并发量为500
  51.         self.semaphore = asyncio.Semaphore(maxsize, loop=self.loop)
  53.     """ 添加任务 """
  54.     def task_add(self, item=1):
  55.         self.task.put(item)
  57.     """ 任务完成 回调函数 """
  58.     def task_done(self, fn):
  59.         if fn:
  60.             pass
  61.         self.task.get()
  62.         self.task.task_done()
  64.     """ 等待任务执行完毕 """
  65.     def wait(self):
  66.         self.task.join()
  69.     @property  # 获取当前线程数
  70.     def running(self):
  71.         return self.task.qsize()
  73.     @staticmethod  # 运行事件循环
  74.     def _start_thread_loop(loop):
  75.         """
  76.         运行事件循环
  77.         :param loop: loop以参数的形式传递进来运行
  78.         :return:
  79.         """
  80.         # 将当前上下文的事件循环设置为循环。
  81.         asyncio.set_event_loop(loop)
  82.         # 开始事件循环
  83.         loop.run_forever()
  85.     """ 停止协程 关闭线程 """
  86.     async def _stop_thread_loop(self, loop_time=1):
  87.         while True:
  88.             if self.task.empty():
  89.                 self.loop.stop() # 停止协程
  90.                 break
  91.             await asyncio.sleep(loop_time)
  93.     """ 运行事件循环 开启新线程  :param loop: 协程"""
  94.     def start_loop(self, loop):
  95.         # 获取一个事件循环
  96.         if not loop:
  97.             loop = asyncio.new_event_loop()
  99.         loop_thread = Thread(target=self._start_thread_loop, args=(loop,))
  100.         loop_thread.setDaemon(True) # 设置守护进程
  101.         loop_thread.start()  # 运行线程，同时协程事件循环也会运行
  102.         return loop, loop_thread
  104.     """ 队列为空，则关闭线程 :param loop_time """
  105.     def stop_loop(self, loop_time=1):
  106.         # 关闭线程任务
  107.         asyncio.run_coroutine_threadsafe(self._stop_thread_loop(loop_time), self.loop)
  109.     """ 释放线程 """
  110.     def release(self, loop_time=1):
  111.         self.stop_loop(loop_time)
  113.     """ 信号包装 """
  114.     async def async_semaphore_func(self, func):
  115.         async with self.semaphore:
  116.             return await func
  118.     """ 提交任务到事件循环 :param func: 异步函数对象, callback: 回调函数"""
  119.     def submit(self, func, callback=None):
  120.         self.task_add()
  121.         # 将协程注册一个到运行在线程中的循环，thread_loop 会获得一个环任务
  122.         # 注意：run_coroutine_threadsafe 这个方法只能用在运行在线程中的循环事件使用
  123.         # future = asyncio.run_coroutine_threadsafe(func, self.loop)
  124.         future = asyncio.run_coroutine_threadsafe(self.async_semaphore_func(func), self.loop)
  126.         # 添加回调函数,添加顺序调用
  127.         future.add_done_callback(callback)
  128.         future.add_done_callback(self.task_done)
  132. async def thread_example(i):
  133.     url = "http://127.0.0.1:3001/?num={}".format(i)
  134.     async with aiohttp.ClientSession() as session:
  135.         async with session.get(url) as res:
  136.             # print(res.status)
  137.             # print(res.content)
  138.             return await res.text()
  142. def my_callback(future):
  143.     result = future.result()
  144.     print('返回值: ', result)
  148. def main():
  149.     # 任务组， 最大协程数
  150.     pool = AsyncPool(maxsize=100000)
  151.     # 增加任务
  152.     pool.submit(thread_example(1), my_callback)
  153.     # 插入任务任务
  154.     # for i in range(10):
  155.     #     pool.submit(thread_example(i), my_callback)
  157.     print("等待子线程结束1...")
  158.     # 停止事件循环
  159.     pool.release()
  161.     # 获取线程数
  162.     print(pool.running)
  163.     print("等待子线程结束2...")
  164.     # 等待
  165.     pool.wait()
  166.     print("等待子线程结束3...")
  170. if __name__ == '__main__':
  171.     start_time = time.time()
  172.     main()
  173.     end_time = time.time()
  174.     print("run time: ", end_time - start_time)

完美的异步网络发送

只要传递 session 即可完成一个数据, 发起多个不同的接口

  1. # -*- coding: utf-8 -*-
  3. """
  4. @author: chenchen
  5. @software: PyCharm
  6. @file: Thread.py
  7. @功能: 
  8. @time: 2022-09-18 1:07
  9. """
  10. import json, platform, time, re
  11. from PyQt5 import QtCore
  12. from PyQt5.QtCore import Qt, pyqtSignal
  13. # uvloop  linux系统的时候安装这个   pip install uvloop -i https://pypi.douban.com/simple
  14. import aiohttp, asyncio, queue, traceback
  15. from aiohttp import ClientSession, client_exceptions, TCPConnector, ClientTimeout
  16. from Meal import Meal, Shop
  17. from utils import api
  19. """ 这是第一个执行的线程 整个循环都在这处理 """
  20. class RunTherad(QtCore.QThread):
  21.     start_prin = pyqtSignal(dict)
  22.     start_prin_status = pyqtSignal(dict)
  24.     def __init__(self):
  25.         super(RunTherad, self).__init__()
  26.         self.dataList = []
  27.         self.Meal = Meal()
  28.         self.Shop = Shop()
  29.         """
  30.             linux: uvloop.EventLoopPolicy())
  31.             win: asyncio.WindowsSelectorEventLoopPolicy()
  32.         """
  33.         sys = platform.system()
  34.         if sys == 'Windows':
  35.             print('windows系统')
  36.             asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
  37.         elif sys == 'Linux':
  38.             # asyncio.set_event_loop_policy(uvloop.EventLoopPolicy)
  39.             print('linux系统')
  40.         else:
  41.             pass
  42.         self.new_loop = asyncio.new_event_loop()
  43.         self.queue_data = queue.Queue()  # 任务队列
  44.         self.timeout_domains = []  # 超时任务队列存放
  45.         self.unknown_error_domains = []  # 超时未知错误队列存放
  47.     def run(self):
  48.         """ 单词运行代码 """
  49.         # asyncio.run(self.main(self.dataList))
  50.         # print("成功连接 number: {}, 超时连接 number: {}, 失败连接数 number: {} ====================".format(self.queue_data.qsize(), len(self.timeout_domains),len(self.unknown_error_domains)))
  51.         # mtshop = self.Meal.getData()
  52.         # if len(mtshop['data']) > 0:
  53.         #     mtshop = self.Meal.getData()
  54.         #     mtshop = json.dumps(mtshop['data'], ensure_ascii=False)
  55.         #     mtshop = mtshop.replace('\'', '\"')
  56.         #     mtshop= json.loads(json.dumps(mtshop, ensure_ascii=False))
  57.         #     self.dataList = json.loads(mtshop)
  58.         #     print(self.dataList, type(self.dataList))
  59.         #     asyncio.run(self.main(self.dataList))
  60.         #     print("成功连接 number: {}, 超时连接 number: {}, 失败连接数 number: {} ====================".format(self.queue_data.qsize(), len(self.timeout_domains),  len( self.unknown_error_domains)))
  61.         #     self.timeout_domains = []  # 超时任务队列存放
  62.         #     self.unknown_error_domains = []  # 超时未知错误队列存放
  63.         #     self.queue_data = queue.Queue()
  64.             # time.sleep(1)
  65.         try:
  66.             """ 批量运行代码 """
  67.             while True:
  68.                 mtshop = self.Meal.getData()
  69.                 print('本次处理数据:', len(mtshop['data']))
  70.                 if len(mtshop['data']) > 0:
  71.                     # mtshop = self.Meal.getData()
  72.                     mtshop = json.dumps(mtshop['data'], ensure_ascii=False)
  73.                     mtshop = mtshop.replace('\'', '\"')
  74.                     mtshop = json.loads(json.dumps(mtshop, ensure_ascii=False))
  75.                     self.dataList = json.loads(mtshop)
  76.                     """ 启动异步网络请求 """
  77.                     asyncio.run(self.main(self.dataList))
  78.                     print("成功连接 number: {}, 超时连接 number: {}, 失败连接数 number: {} ====================".format(self.queue_data.qsize(),  len(self.timeout_domains), len( self.unknown_error_domains)))
  79.                     self.timeout_domains = []  # 超时任务队列存放
  80.                     self.unknown_error_domains = []  # 超时未知错误队列存放
  81.                     self.queue_data = queue.Queue()
  82.                     time.sleep(5)
  83.                 else:
  84.                     print('数据库没有任务, 等待10秒后再次检测=====>>>>>')
  85.                     time.sleep(10)
  87.         except Exception as e:
  88.             print('run报错:', e)
  91.     async def fetch(self, session, n, linedata):
  92.         """
  93.         :param session:  aiohttp.ClientSession
  94.         :param n: 数量 序号
  95.         :param linedata: 整条的onedata
  96.         """
  97.         cookie = linedata['cookie']
  98.         _state = linedata['state']
  99.         _expire = linedata['expire']
  100.         _mark = linedata['mark']
  101.         _expiretime = linedata['expiretime']
  102.         _shop_id = linedata['shop_id']
  103.         region_id = re.findall('region_id=(.*?);', cookie)[0]
  104.         region_version = re.findall('region_version=(.*?);', cookie)[0]
  105.         start_time = time.time()
  107.         try:
  108.             """ 判断店铺 是否过期, 即将过期 """
  109.             _expiretime = _expiretime + ' 23:59:59'
  110.             _expiretime = int(time.mktime(time.strptime(_expiretime, '%Y-%m-%d %H:%M:%S')))
  111.             _currenttime =int(time.time())
  112.             if _currenttime > _expiretime:
  113.                 # print('过期了:', _shop_id)
  114.                 if _expire != 1:
  115.                     await self.Meal.shop_expire_update(self, session,_shop_id, 1)
  116.             elif _expiretime - _currenttime < 86399:
  117.                 # print('即将过期:', _shop_id)
  118.                 if _expire != 2:
  119.                     await self.Meal.shop_expire_update(self, session, _shop_id, 2)
  121.             """ 不是营业中状态的 就只查询店铺状态, 否则就查询店铺订单出餐并出单"""
  122.             if _state != 1:
  123.                 await self.Shop.businessSatus( session, cookie, region_id, region_version, _state)
  124.                 end_time = time.time()
  125.                 cost = end_time - start_time
  126.                 print(f"营业状态请求:{n}, 耗时:{cost}, 店铺: {linedata['shop_id']}")
  127.             else:
  128.                 url, param, headers = api.dataShopOrder(self, cookie, region_id, region_version)
  129.                 async with session.get(url=url, params=param, headers=headers,  timeout=5, verify_ssl=False) as response:
  130.                     d = await response.json()  # 返回订单的数据 是否有
  131.                     if d['code'] == 1001:      # """ cookie 失效请重新登录 """
  132.                         await    self.Meal.shop_cookie_update(session, {'shop_id': linedata['shop_id']})
  133.                     elif d['code'] == 0:       # """ cookie 正常 """
  134.                         if len(d['data']['wmOrderList']) > 0:  # 判断是否有订单数据, 循环遍历订单是否可以出餐
  135.                             for index, item in enumerate(d['data']['wmOrderList']):
  136.                                 js = json.loads(item['orderInfo'])
  137.                                 ts = time.localtime().tm_hour  # 1. 这里还需要判断 当前的时间, 选择使用 指定时间
  138.                                 num = js['num']
  139.                                 order_time = js['order_time']
  140.                                 order_num = js['wm_order_id_view_str']
  141.                                 cityId = js['poi_city_id']
  142.                                 wmPoiId = js['wm_poi_id']
  143.                                 utime = js['utime']
  144.                                 # 按指定使用出餐
  145.                                 vals = {'region_id': region_id, 'region_version': region_version, 'wmPoiId': wmPoiId, 'num': num, 'order_num': order_num, 'order_time': order_time, 'cityId': cityId, 'utime': utime, 'cookie': cookie }
  146.                                 if ts == 11:
  147.                                     if utime + linedata['meal_morning_time'] < int(time.time()):
  148.                                         print(f"午高峰时间出餐   订餐号: {num}, 下单时间: {order_time}, 订单号: {order_num}, 店铺id: {wmPoiId}, 城市id: {cityId}")
  149.                                         await self.Meal.Receive_send(session, vals)
  150.                                 elif ts == 17 or ts == 18:
  151.                                     if utime + linedata['meal_night_time'] < int(time.time()):
  152.                                         print(f"晚高峰时间出餐   订餐号: {num}, 下单时间: {order_time}, 订单号: {order_num}, 店铺id: {wmPoiId}, 城市id: {cityId}")
  153.                                         await self.Meal.Receive_send(session, vals)
  154.                                 else:
  155.                                     if utime + linedata['meal_time'] < int(time.time()):
  156.                                         print(f"正常时间出餐   订餐号: {num}, 下单时间: {order_time}, 订单号: {order_num}, 店铺id: {wmPoiId}, 城市id: {cityId}")
  157.                                         await self.Meal.Receive_send(session, vals)
  158.                         end_time = time.time()
  159.                         cost = end_time - start_time
  161.                         print(f"本次请求:{n}, 耗时:{cost}, 店铺: {linedata['shop_id']}")
  162.                     else:
  163.                         print('查询店铺订单, 网络错误, 请检查是否有网络信号')
  164.             self.queue_data.put(n)
  165.         except client_exceptions.ServerTimeoutError as timeout_error:  # 记录超时的店铺请求
  166.             print("fetch函数 请求超时的店铺 error: {}, url: {}".format(timeout_error, linedata['shop_id']))
  167.             self.timeout_domains.append(linedata['shop_id'])
  168.         except Exception:
  169.             print("fetc函数 错误捕捉 unknown error: {}".format(traceback.format_exc()))
  170.             self.unknown_error_domains.append(linedata['shop_id'])
  175.     async def chunks(self, sem, session, i, oneData):
  176.         try:
  177.             async with sem:
  178.                 await self.fetch(session, i + 1, oneData)
  179.                 # self.new_loop.call_soon_threadsafe(self.new_loop.stop)
  180.         except Exception as e:
  181.             print("chunks报错:", e)
  183.     async def main(self, data):
  184.         sem = asyncio.Semaphore(100)
  185.         # timeout = ClientTimeout(total=5, connect=5, sock_connect=15, sock_read=5)
  186.         timeout = ClientTimeout(total=10, connect=10000, sock_connect=5, sock_read=5)
  187.         async with ClientSession(connector=TCPConnector(limit=100), timeout=timeout) as session:
  188.             tasks = [asyncio.create_task(self.chunks(sem, session, index, shop_data)) for index, shop_data in enumerate(data)]
  189.             await asyncio.wait(tasks)
  190.             # print('结束:' ,await asyncio.wait(tasks))