add celeries

lambdaplus · lambdaplus · commit 66a25660706b · 2017-03-23T15:40:02.000+08:00
diff --git a/.gitignore b/.gitignore
@@ -88,3 +88,7 @@ ENV/
 
 # Rope project settings
 .ropeproject
+# celery-beat
+*.bak
+*.dat
+*.dir
diff --git a/celeries/proj/__init__.py b/celeries/proj/__init__.py
diff --git a/celeries/proj/celery.py b/celeries/proj/celery.py
@@ -0,0 +1,10 @@
+# coding=utf-8
+from __future__ import absolute_import
+from celery import Celery
+
+app = Celery('proj', include=['proj.tasks'])
+app.config_from_object('proj.celeryconfig')
+
+
+if __name__ == '__main__':
+    app.start()
diff --git a/celeries/proj/celeryconfig.py b/celeries/proj/celeryconfig.py
@@ -0,0 +1,7 @@
+# coding=utf-8
+BROKER_URL = 'amqp://localhost'  # RabbitMQ 作为消息代理
+CELERY_RESULT_BACKEND = 'redis://localhost' # Redis 作为结果存储
+CELERY_TASK_SERIALIZER = 'msgpack'
+# 任务序列化和反序列化格式为 msgpack, 别忘了安装 msgpack-python
+CELERY_RESULT_SERIALIZER = 'json'   # 结果存储序列化格式为 json
+CELERY_ACCEPT_CONTENT = ['msgpack', 'json']  # 任务接受格式类型
diff --git a/celeries/proj/tasks.py b/celeries/proj/tasks.py
@@ -0,0 +1,35 @@
+# coding=utf-8
+from __future__ import absolute_import
+
+from .celery import app
+
+from celery.utils.log import get_task_logger
+
+logger = get_task_logger(__name__)
+
+
+@app.task
+def add(x, y):
+    return x + y
+
+
+@app.task
+def mul(x, y):
+    return x * y
+
+
+@app.task(bind=True)
+def div(self, x, y):
+    logger.info(
+        '''
+        Executing task : {0.id}
+        task.args      : {0.args!r}
+        task.kwargs    : {0.kwargs!r}
+        '''.format(self.request)
+    )
+    try:
+        res = x / y
+    except ZeroDivisionError as e:
+        raise self.retry(exc=e, countdown=3, max_retries=3)
+    else:
+        return res
diff --git a/celeries/projb/__init__.py b/celeries/projb/__init__.py
diff --git a/celeries/projb/celery.py b/celeries/projb/celery.py
@@ -0,0 +1,10 @@
+# coding=utf-8
+from __future__ import absolute_import
+from celery import Celery
+
+app = Celery('projb', include=['projb.tasks'])
+app.config_from_object('projb.celeryconfig')
+
+
+if __name__ == '__main__':
+    app.start()
diff --git a/celeries/projb/celeryconfig.py b/celeries/projb/celeryconfig.py
@@ -0,0 +1,28 @@
+# coding=utf-8
+from kombu import Queue
+
+BROKER_URL = 'amqp://localhost'  # RabbitMQ 作为消息代理
+CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'  # Redis 作为结果存储
+CELERY_TASK_SERIALIZER = 'msgpack'
+# 任务序列化和反序列化格式为 msgpack, 别忘了安装 msgpack-python
+CELERY_RESULT_SERIALIZER = 'json'   # 结果存储序列化格式为 json
+CELERY_ACCEPT_CONTENT = ['msgpack', 'json']  # 任务接受格式类型
+
+CELERY_QUEUES = {
+    Queue('foo', routing_key='task.#'),
+    Queue('feed_task', routing_key='*.feed'),
+}
+CELERY_DEFAULT_QUEUE = 'foo'
+
+CELERY_DEFAULT_EXCHANGE = 'tasks'
+
+CELERY_DEFAULT_EXCHANGE_TYPE = 'topic'
+
+CELERY_DEFAULT_ROUTING_KEY = 'task.foooooo'
+
+CELERY_ROUTES = {
+    'projb.tasks.mul': {
+        'queue': 'feed_task',
+        'routing_key': 'mul.feed',
+    },
+}
diff --git a/celeries/projb/tasks.py b/celeries/projb/tasks.py
@@ -0,0 +1,14 @@
+# coding=utf-8
+from __future__ import absolute_import
+
+from .celery import app
+
+
+@app.task
+def add(x, y):
+    return x + y
+
+
+@app.task
+def mul(x, y):
+    return x * y
diff --git a/celeries/projc/__init__.py b/celeries/projc/__init__.py
diff --git a/celeries/projc/celery.py b/celeries/projc/celery.py
@@ -0,0 +1,10 @@
+# coding=utf-8
+from __future__ import absolute_import
+from celery import Celery
+
+app = Celery('projc', include=['projc.tasks'])
+app.config_from_object('projc.celeryconfig')
+
+
+if __name__ == '__main__':
+    app.start()
diff --git a/celeries/projc/celeryconfig.py b/celeries/projc/celeryconfig.py
@@ -0,0 +1,37 @@
+# coding=utf-8
+from kombu import Queue
+
+BROKER_URL = 'amqp://localhost'  # RabbitMQ 作为消息代理
+CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'  # Redis 作为结果存储
+CELERY_TASK_SERIALIZER = 'msgpack'
+# 任务序列化和反序列化格式为 msgpack, 别忘了安装 msgpack-python
+CELERY_RESULT_SERIALIZER = 'json'   # 结果存储序列化格式为 json
+CELERY_ACCEPT_CONTENT = ['msgpack', 'json']  # 任务接受格式类型
+
+CELERY_QUEUES = {
+    Queue('foo', routing_key='task.#'),
+    Queue('feed_task', routing_key='*.feed'),
+}
+CELERY_DEFAULT_QUEUE = 'foo'
+
+CELERY_DEFAULT_EXCHANGE = 'tasks'
+
+CELERY_DEFAULT_EXCHANGE_TYPE = 'topic'
+
+CELERY_DEFAULT_ROUTING_KEY = 'task.foooooo'
+
+CELERY_ROUTES = {
+    'projb.tasks.mul': {
+        'queue': 'feed_task',
+        'routing_key': 'mul.feed',
+    },
+
+}
+
+CELERYBEAT_SCHEDULE = {
+    'mul-every-30-seconds': {
+        'task': 'projc.tasks.mul',
+        'schedule': 30.0,
+        'args': (2, 2),
+    }
+}
diff --git a/celeries/projc/tasks.py b/celeries/projc/tasks.py
@@ -0,0 +1,14 @@
+# coding=utf-8
+from __future__ import absolute_import
+
+from .celery import app
+
+
+@app.task
+def add(x, y):
+    return x + y
+
+
+@app.task
+def mul(x, y):
+    return x * y
