add celery demo

Author: xdai555

demo/celery_demo/celery.py

@@ -0,0 +1,17 @@
+1. 启动 celery 进程，开始监听（角色为消费者，处理任务）
+
+   `celery -A task worker -l info`
+
+   -A 后面跟应用名称，即有 Celery 对象的文件，这里是 task.py
+
+   `worker -c <concurrency>` 指定并发的数量，默认是系统 cpu 数量
+
+   `worker -P [prefork|eventlet|gevent|solo|processes|threads]`  池的实现方式，默认是 prefork
+
+2. 运行任务 exec_task.py（角色为生产者，产生任务）
+
+   `python exec_task.py`
+   
+   运行之后，可以在 celery 进程中看到 worker 已经接受了一系列 task 并开始了处理，一般在运行任务时打印出任务 ID。
+
+3. 运行结果可以通过 fetch_result.py 根据任务 ID 来获取。

demo/celery_demo/celery_demo1.py

@@ -0,0 +1,14 @@
+from task import send_msg, send_msg1
+import time
+
+# for i in range(5):
+#     send_msg1.delay(str(i))
+#     send_msg.delay(str(i+5))
+
+result = send_msg.delay("Hello")
+print(result.id)
+while not result.ready():
+    print("正在执行中...")
+    time.sleep(0.5)
+
+print("执行完成\n" + result.get())

demo/celery_demo/demo1/README.md

@@ -0,0 +1,14 @@
+from task import send_msg
+import time
+from datetime import datetime
+
+
+# 指定时间运行任务，默认是 utc 时区
+time_start = datetime(2022,3,25,16,44,00).timestamp()
+time_start = datetime.utcfromtimestamp(time_start)
+print(time_start)
+
+result = send_msg.apply_async(args=["hello"], eta=time_start)
+print(result.id)
+
+

demo/celery_demo/demo1/exec_task.py

@@ -0,0 +1,12 @@
+from celery.result import AsyncResult
+from task import cel
+
+async_result = AsyncResult(id="be35b203-3f5e-4da2-b3d9-e2d1f9b12dc3", app=cel)
+
+STATUS = ['PENDING', 'STARTED', 'RETRY', 'FAILURE', 'SUCCESS']
+
+if async_result.successful():
+    result = async_result.get()
+    print(result)
+elif async_result.failed():
+    print('任务执行失败')

demo/celery_demo/demo1/exec_task_async.py

@@ -0,0 +1,23 @@
+import time
+from celery import Celery
+from pytz import timezone
+
+broker = "redis://localhost:6379/0"
+backend = "redis://localhost:6379/1"
+
+cel = Celery('tasks', broker=broker,backend=backend)
+
+# 相当于注册一个任务
+@cel.task
+def send_msg(user):
+    print(f"MSG-1 Sending msg to {user}...")
+    time.sleep(2)
+    msg = f"MSG-1 Sent msg to {user} done."
+    return msg
+
+@cel.task
+def send_msg1(user):
+    print(f"MSG-2 Sending msg1 to {user}...")
+    time.sleep(5)
+    msg = f"MSG-2 Sent msg1 to {user} done."
+    return msg

demo/celery_demo/demo1/fetch_result.py

@@ -0,0 +1 @@
+拆分任务模块，在初始化 Celery 对象的时候使用 `include` 来注册任务。

demo/celery_demo/demo1/task.py

@@ -0,0 +1,35 @@
+from datetime import timedelta
+from celery import Celery
+
+broker = "redis://localhost:6379/0"
+backend = "redis://localhost:6379/1"
+
+cel = Celery(
+    "celery_demo",
+    broker=broker,
+    backend=backend,
+    # 通过引入的方式来注册任务，可以将各个任务分散开
+    include=["celery_tasks.send_msg", "celery_tasks.send_mail"],
+)
+
+cel.conf.timezone = "Asia/Shanghai"
+
+# 启动 worker: `celery -A celery_tasks  worker -l info`
+
+
+# 创建定时任务
+cel.conf.beat_schedule = {
+    "run_task1": {
+        "task": "celery_tasks.send_msg.send_msg_1",
+        "schedule": timedelta(seconds=5),
+        "args": ("hello msg",)
+    },
+    "run_task2": {
+        "task": "celery_tasks.send_mail.send_mail_1",
+        "schedule": timedelta(seconds=5),
+        "args": ("hello mail",)
+    },
+}
+
+# 启动定时任务，将任务周期性地添加到 worker 队列中执行
+