funboost代码测试

Controller/TestApi.py

@@ -37,6 +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
 
 ACCESS_KEY = "AKIA2E67UIMD3CYTIWPA"
 SECRET_KEY = "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"
@@ -122,6 +123,8 @@ class testView(View):
             return self.get_serial_details(request_dict, response, request)
         elif operation == 'find_device_serial':  # 查找设备序列号接口：306低功耗无Wi-Fi产品
             return self.find_device_serial(request_dict, response)
+        elif operation == 'funboost':  # funboost测试
+            return self.funboost_test(request_dict, response)
         else:
             return response.json(414)
 
@@ -901,3 +904,15 @@ class testView(View):
         if password_version == 'V1':
             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)
+        self.task_fun.consume()
+        return response.json(0)