fix(helmbroker): rabbitmq sharding err

Author: duanhongyi

rootfs/helmbroker/celery.py

@@ -1,10 +1,9 @@
 import os
+from kombu import Exchange, Queue
 from celery import Celery
 
 
-# The queue and exchange names cannot be the same
-# otherwise an error will occur when enabling the sharding plugin
-class Config:
+class Config(object):
     # Celery Configuration Options
     timezone = "Asia/Shanghai"
     enable_utc = True
@@ -19,20 +18,38 @@ class Config:
     task_time_limit = 30 * 60
     worker_max_tasks_per_child = 200
     result_expires = 24 * 60 * 60
+    broker_url = os.environ.get("DRYCC_RABBITMQ_URL", 'amqp://guest:guest@127.0.0.1:5672/')  # noqa
     broker_connection_retry_on_startup = True
     task_default_queue = 'low'
     task_default_exchange = 'helmbroker.priority'
     task_default_routing_key = 'helmbroker.priority.low'
+    broker_connection_retry_on_startup = True
     worker_cancel_long_running_tasks_on_connection_loss = True
 
 
-app = Celery(
-    'helmbroker',
-    broker=os.environ.get("DRYCC_RABBITMQ_URL"),
-    include=['helmbroker.tasks']
+app = Celery('helmbroker')
+app.config_from_object(Config())
+app.conf.update(
+    task_routes={
+        'helmbroker.tasks': {
+            'queue': 'low',
+            'exchange': 'helmbroker.priority',
+            'routing_key': 'helmbroker.priority.high',
+        },
+    },
+    task_queues=(
+        Queue(
+            'low',
+            exchange=Exchange('helmbroker.priority', type="direct"),
+            routing_key='helmbroker.priority.low',
+            queue_arguments={'x-max-priority': 16},
+        ),
+    ),
 )
+app.autodiscover_tasks()
+
 
-app.config_from_object(Config)
+app.config_from_object(Config())
 
 if __name__ == '__main__':
     app.start()