modify

_celery.py

+# !/usr/bin/env python
+# encoding=utf-8
+from __future__ import absolute_import
+
+import os
+# set the default Django settings module for the 'celery' program.
+# os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings')
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'physical.settings')
+
+import raven
+from raven.contrib.celery import register_signal, register_logger_signal
+from celery import Celery
+from django.conf import settings
+
+
+class Celery(Celery):
+    """wrap for celery.Celery."""
+
+    def on_configure(self):
+        # check if sentry settings provided
+        if not settings.SENTRY_CELERY_ENDPOINT:
+            return
+
+        client = raven.Client(settings.SENTRY_CELERY_ENDPOINT)
+
+        # register a custom filter to filter out duplicate logs
+        register_logger_signal(client)
+
+        # hook into the Celery error handler
+        register_signal(client)
+
+
+app = Celery('physical_tasks')
+
+
+# Using a string here means the worker will not have to
+# pickle the object when using Windows.
+app.config_from_object('django.conf:settings')
+app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
+

search_tips/__init__.py

 from __future__ import unicode_literals, absolute_import, print_function
 
 import pymysql
+from _celery import app as celery_app
 
 pymysql.install_as_MySQLdb()
\ No newline at end of file

search_tips/celery.py

+from __future__ import absolute_import, unicode_literals
+import os
+from celery import Celery
+from django.conf import settings
+
+# set the default Django settings module for the 'celery' program.
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'search_tips.settings')
+
+app = Celery('search_tips')
+
+# Using a string here means the worker doesn't have to serialize
+# the configuration object to child processes.
+# - namespace='CELERY' means all celery-related configuration keys
+#   should have a `CELERY_` prefix.
+app.config_from_object('django.conf:settings', namespace='CELERY')
+
+# Load task modules from all registered Django app configs.
+app.autodiscover_tasks()
+app.conf.broker_url = settings.CELERY_BROKER_URL
+
+
+@app.task(bind=True)
+def debug_task(self):
+    print('Request: {0!r}'.format(self.request))

search_tips/celery_task_router.py

+# coding=utf-8                                                                           
+from __future__ import unicode_literals, print_function, absolute_import
+from django.conf import settings
+import itertools
+import logging
+
+
+class CeleryTaskRouter(object):
+    queue_task_map = {
+        "tapir-search_tips": [
+            'injection.data_sync.tasks.write_to_es',
+        ]
+    }
+
+    # Map[TaskName, QueueName]
+    task_queue_map = dict(itertools.chain.from_iterable(
+        [(task, queue) for task in task_list]
+        for (queue, task_list) in queue_task_map.items()
+    ))
+
+    def route_for_task(self, task, args=None, kwargs=None):
+        """
+        if settings.DEBUG:
+            return None
+
+        if task.startswith("statistic") or task.startswith("api.tasks.export_excel_task"):
+            return "slow"
+        """
+
+        queue_name_or_none = self.task_queue_map.get(task)
+        return queue_name_or_none