通过线程池推送

Service/CustomizedPushService.py

@@ -4,9 +4,9 @@
 import logging
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
 
 from Model.models import DeviceTypeModel, Device_Info, GatewayPush, CountryModel, SysMsgModel
-from Object.RedisObject import RedisObject
 from Service.CommonService import CommonService
 from Service.HuaweiPushService.HuaweiPushService import HuaweiPushObject
 from Service.PushService import PushObject
@@ -84,7 +84,6 @@ class CustomizedPushObject:
             'n_time': n_time,
             'title': title,
             'msg': msg,
-            'link': link,
             'icon_link': icon_link
         }
 
@@ -95,7 +94,8 @@ class CustomizedPushObject:
             app_bundle_id_list = ['com.ansjer.zccloud_ab', 'com.ansjer.customizede']
 
         try:
-            gateway_push_qs = GatewayPush.objects.filter(user_id__in=user_id_list, app_bundle_id__in=app_bundle_id_list).\
+            gateway_push_qs = GatewayPush.objects.filter(
+                user_id__in=user_id_list, app_bundle_id__in=app_bundle_id_list).\
                 values('user_id', 'app_bundle_id', 'push_type', 'token_val')
             if gateway_push_qs.exists():
                 sys_msg_list = []
@@ -116,7 +116,7 @@ class CustomizedPushObject:
                     'gateway_push_list': gateway_push_list
                 }
                 pre_push_thread = threading.Thread(
-                    target=cls.pre_push,
+                    target=cls.thr_pool_push,
                     kwargs=pre_push_kwargs)
                 pre_push_thread.start()
                 CUSTOMIZED_PUSH_LOGGER.info('customized_push_id:{}推送完成'.format(kwargs['id']))
@@ -125,32 +125,25 @@ class CustomizedPushObject:
                                         'error_line:{},error_msg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
 
     @classmethod
-    def pre_push(cls, **kwargs):
-        CUSTOMIZED_PUSH_LOGGER.info('预推送线程开始')
+    def thr_pool_push(cls, **kwargs):
+        CUSTOMIZED_PUSH_LOGGER.info('线程池推送开始')
         push_kwargs = kwargs['push_kwargs']
         gateway_push_list = kwargs['gateway_push_list']
-        for gateway_push in gateway_push_list:
-            # 异步推送消息
-            push_kwargs['gateway_push'] = gateway_push
-            push_thread = threading.Thread(
-                target=cls.start_push,
-                kwargs=push_kwargs)
-            push_thread.start()
-        CUSTOMIZED_PUSH_LOGGER.info('预推送线程完成')
+        with ThreadPoolExecutor() as executor:
+            executor.map(
+                lambda gateway_push_kwargs: cls.start_push(push_kwargs, gateway_push_kwargs), gateway_push_list)
 
     @classmethod
-    def start_push(cls, **kwargs):
-        gateway_push = kwargs['gateway_push']
-        title = kwargs['title']
-        n_time = kwargs['n_time']
-        msg = kwargs['msg']
-        link = kwargs['link']
-        icon_link = kwargs['icon_link']
+    def start_push(cls, push_kwargs, gateway_push_kwargs):
+        title = push_kwargs['title']
+        n_time = push_kwargs['n_time']
+        msg = push_kwargs['msg']
+        icon_link = push_kwargs['icon_link']
 
-        push_type = gateway_push['push_type']
-        user_id = gateway_push['user_id']
-        app_bundle_id = gateway_push['app_bundle_id']
-        token_val = gateway_push['token_val']
+        push_type = gateway_push_kwargs['push_type']
+        user_id = gateway_push_kwargs['user_id']
+        app_bundle_id = gateway_push_kwargs['app_bundle_id']
+        token_val = gateway_push_kwargs['token_val']
 
         push_succeed = cls.push_msg(push_type, app_bundle_id, token_val, n_time, title, msg, icon_link)
 
@@ -181,7 +174,7 @@ class CustomizedPushObject:
                 return PushObject.android_fcm_push(**push_kwargs)
             # 极光
             elif push_type == 2:
-                push_succeed = PushObject.android_jpush(**push_kwargs)
+                return PushObject.android_jpush(**push_kwargs)
             # 华为
             elif push_type == 3:
                 push_kwargs['image_url'] = icon_link