funboost代码测试

Controller/TestApi.py

@@ -37,7 +37,7 @@ from Object.TokenObject import TokenObject
 from Object.m3u8generate import PlaylistGenerator
 from Service.CommonService import CommonService
 from Service.VodHlsService import SplitVodHlsObject
-# from funboost import boost, BrokerEnum, funboost_aps_scheduler
+from funboost import boost, BrokerEnum, funboost_aps_scheduler
 
 ACCESS_KEY = "AKIA2E67UIMD3CYTIWPA"
 SECRET_KEY = "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"
@@ -905,17 +905,18 @@ class testView(View):
             password = make_password(password)
         return response.json(0)
 
-    # @staticmethod
-    # @boost("task_queue_name1", qps=0.5, broker_kind=BrokerEnum.REDIS_ACK_ABLE)  # 入参包括20种，运行控制方式非常多，想得到的控制都会有。
-    # def task_fun(x, y):
-    #     print(f'{x} + {y} = {x + y}')
-    #     return True
-    #
-    #
-    # def funboost_test(self, request_dict, response):
-    #     # self.task_fun.push(x=5, y=6)
-    #     funboost_aps_scheduler.start()
-    #     funboost_aps_scheduler.add_push_job(self.task_fun, 'interval', id='3_second_job', seconds=60,
-    #                                         kwargs={"x": 5, "y": 6})  # 每隔3秒发布一次任务，自然就能每隔3秒消费一次任务了。
-    #     self.task_fun.consume()
-    #     return response.json(0)
+    @staticmethod
+    @boost("task_queue_name1", qps=0.5, broker_kind=BrokerEnum.REDIS_ACK_ABLE)  # 入参包括20种，运行控制方式非常多，想得到的控制都会有。
+    def task_fun(x, y):
+        print(f'{x} + {y} = {x + y}')
+        return True
+
+
+    def funboost_test(self, request_dict, response):
+        # self.task_fun.push(x=5, y=6)
+        self.task_fun.clear()
+        funboost_aps_scheduler.start()
+        funboost_aps_scheduler.add_push_job(self.task_fun, 'interval', id='3_second_job', seconds=60,
+                                            kwargs={"x": 5, "y": 6})  # 每隔3秒发布一次任务，自然就能每隔3秒消费一次任务了。
+        self.task_fun.consume()
+        return response.json(0)