modify

injection/data_sync/tasks.py

@@ -6,12 +6,15 @@ from django.core import serializers
 from trans2es.type_info import get_type_info_map
 from rpc.all import get_rpc_remote_invoker
 from libs.es import ESPerform
+import logging
 
 @shared_task
 def write_to_es(es_type, pk_list, configuration, use_batch_query_set=False):
     pk_list = list(frozenset(pk_list))
     type_info_map = get_type_info_map()
     type_info = type_info_map[es_type]
+
+    logging.info("duan add,es_type:%s" % str(es_type))
     type_info.insert_table_by_pk_list(
         sub_index_name=es_type,
         pk_list=pk_list,

physical/__init__.py

 import pymysql
+from .celery import app as celery_app
+
 
 pymysql.install_as_MySQLdb()
+__all__ = ('celery_app',)

physical/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', 'physical.settings')
+
+app = Celery('physical')
+
+# 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))

physical/settings.py

@@ -30,7 +30,7 @@ ALLOWED_HOSTS = []
 
 
 # Application definition
-BROKER_URL = "redis://127.0.0.1:6379/0"
+CELERY_BROKER_URL = "redis://127.0.0.1:6379/2"
 CELERY_TIMEZONE = 'Asia/Shanghai'
 CELERY_ROUTES = ['physical.celery_task_router.CeleryTaskRouter']