{"id":5714,"date":"2022-04-03T07:18:33","date_gmt":"2022-04-02T23:18:33","guid":{"rendered":"https:\/\/egonlin.com\/?p=5714"},"modified":"2022-04-03T07:30:57","modified_gmt":"2022-04-02T23:30:57","slug":"07-01-celery","status":"publish","type":"post","link":"https:\/\/egonlin.com\/?p=5714","title":{"rendered":"07-01 Celery"},"content":{"rendered":"

Celery<\/h1>\n

1.\u4ec0\u4e48\u662fClelery<\/h2>\n

\"\"<\/div><\/p>\n

Celery\u662f\u4e00\u4e2a\u7b80\u5355\u3001\u7075\u6d3b\u4e14\u53ef\u9760\u7684\uff0c\u5904\u7406\u5927\u91cf\u6d88\u606f\u7684\u5206\u5e03\u5f0f\u7cfb\u7edf<\/p>\n

\u4e13\u6ce8\u4e8e\u5b9e\u65f6\u5904\u7406\u7684\u5f02\u6b65\u4efb\u52a1\u961f\u5217<\/p>\n

\u540c\u65f6\u4e5f\u652f\u6301\u4efb\u52a1\u8c03\u5ea6<\/p>\n

Celery\u67b6\u6784<\/h3>\n

\"\"<\/div><\/p>\n

Celery\u7684\u67b6\u6784\u7531\u4e09\u90e8\u5206\u7ec4\u6210\uff0c\u6d88\u606f\u4e2d\u95f4\u4ef6\uff08message broker\uff09\uff0c\u4efb\u52a1\u6267\u884c\u5355\u5143\uff08worker\uff09\u548c\u4efb\u52a1\u6267\u884c\u7ed3\u679c\u5b58\u50a8\uff08task result store\uff09\u7ec4\u6210\u3002<\/p>\n

\u6d88\u606f\u4e2d\u95f4\u4ef6<\/h4>\n

Celery\u672c\u8eab\u4e0d\u63d0\u4f9b\u6d88\u606f\u670d\u52a1\uff0c\u4f46\u662f\u53ef\u4ee5\u65b9\u4fbf\u7684\u548c\u7b2c\u4e09\u65b9\u63d0\u4f9b\u7684\u6d88\u606f\u4e2d\u95f4\u4ef6\u96c6\u6210\u3002\u5305\u62ec\uff0cRabbitMQ, Redis\u7b49\u7b49<\/p>\n

\u4efb\u52a1\u6267\u884c\u5355\u5143<\/h4>\n

Worker\u662fCelery\u63d0\u4f9b\u7684\u4efb\u52a1\u6267\u884c\u7684\u5355\u5143\uff0cworker\u5e76\u53d1\u7684\u8fd0\u884c\u5728\u5206\u5e03\u5f0f\u7684\u7cfb\u7edf\u8282\u70b9\u4e2d\u3002<\/p>\n

\u4efb\u52a1\u7ed3\u679c\u5b58\u50a8<\/h4>\n

Task result store\u7528\u6765\u5b58\u50a8Worker\u6267\u884c\u7684\u4efb\u52a1\u7684\u7ed3\u679c\uff0cCelery\u652f\u6301\u4ee5\u4e0d\u540c\u65b9\u5f0f\u5b58\u50a8\u4efb\u52a1\u7684\u7ed3\u679c\uff0c\u5305\u62ecAMQP, redis\u7b49<\/p>\n

\u7248\u672c\u652f\u6301\u60c5\u51b5<\/h3>\n
Celery version 4.0 runs on\n        Python \u27682.7, 3.4, 3.5\u2769\n        PyPy \u27685.4, 5.5\u2769\n    This is the last version to support Python 2.7, and from the next version (Celery 5.x) Python 3.5 or newer is required.\n\n    If you\u2019re running an older version of Python, you need to be running an older version of Celery:\n\n        Python 2.6: Celery series 3.1 or earlier.\n        Python 2.5: Celery series 3.0 or earlier.\n        Python 2.4 was Celery series 2.2 or earlier.\n\n    Celery is a project with minimal funding, so we don\u2019t support Microsoft Windows. Please don\u2019t open any issues related to that platform.<\/code><\/pre>\n
\"\"<\/div><\/p>\n
2.\u4f7f\u7528\u573a\u666f<\/h2>\n
\u5f02\u6b65\u4efb\u52a1\uff1a\u5c06\u8017\u65f6\u64cd\u4f5c\u4efb\u52a1\u63d0\u4ea4\u7ed9Celery\u53bb\u5f02\u6b65\u6267\u884c\uff0c\u6bd4\u5982\u53d1\u9001\u77ed\u4fe1\/\u90ae\u4ef6\u3001\u6d88\u606f\u63a8\u9001\u3001\u97f3\u89c6\u9891\u5904\u7406\u7b49\u7b49<\/p>\n
\u5b9a\u65f6\u4efb\u52a1\uff1a\u5b9a\u65f6\u6267\u884c\u67d0\u4ef6\u4e8b\u60c5\uff0c\u6bd4\u5982\u6bcf\u5929\u6570\u636e\u7edf\u8ba1<\/p>\n
3.Celery\u7684\u5b89\u88c5\u914d\u7f6e<\/h2>\n
pip install celery<\/p>\n
\u6d88\u606f\u4e2d\u95f4\u4ef6\uff1aRabbitMQ\/Redis<\/p>\n
app=Celery(‘\u4efb\u52a1\u540d’\uff0cbackend=’xxx’,broker=’xxx’)<\/p>\n
4.Celery\u6267\u884c\u5f02\u6b65\u4efb\u52a1<\/h2>\n
\u57fa\u672c\u4f7f\u7528<\/h4>\n
\u521b\u5efa\u9879\u76eecelerytest<\/p>\n
\u521b\u5efapy\u6587\u4ef6\uff1acelery_app_task.py<\/p>\n
import celery\nimport time\n# broker='redis:\/\/127.0.0.1:6379\/2' \u4e0d\u52a0\u5bc6\u7801\nbackend='redis:\/\/:123456@127.0.0.1:6379\/1'\nbroker='redis:\/\/:123456@127.0.0.1:6379\/2'\ncel=celery.Celery('test',backend=backend,broker=broker)\n@cel.task\ndef add(x,y):\n    return x+y<\/code><\/pre>\n
\u521b\u5efapy\u6587\u4ef6\uff1aadd_task.py,\u6dfb\u52a0\u4efb\u52a1<\/p>\n
from celery_app_task import add\nresult = add.delay(4,5)\nprint(result.id)<\/code><\/pre>\n
\u521b\u5efapy\u6587\u4ef6\uff1arun.py\uff0c\u6267\u884c\u4efb\u52a1\uff0c\u6216\u8005\u4f7f\u7528\u547d\u4ee4\u6267\u884c\uff1acelery worker -A celery_app_task -l info<\/p>\n
\u6ce8\uff1awindows\u4e0b\uff1acelery worker -A celery_app_task -l info -P eventlet<\/p>\n
from celery_app_task import cel\nif __name__ == '__main__':\n    cel.worker_main()\n    # cel.worker_main(argv=['--loglevel=info')<\/code><\/pre>\n
\u521b\u5efapy\u6587\u4ef6\uff1aresult.py\uff0c\u67e5\u770b\u4efb\u52a1\u6267\u884c\u7ed3\u679c<\/p>\n
from celery.result import AsyncResult\nfrom celery_app_task import cel\n\nasync = AsyncResult(id="e919d97d-2938-4d0f-9265-fd8237dc2aa3", app=cel)\n\nif async.successful():\n    result = async.get()\n    print(result)\n    # result.forget() # \u5c06\u7ed3\u679c\u5220\u9664\nelif async.failed():\n    print('\u6267\u884c\u5931\u8d25')\nelif async.status == 'PENDING':\n    print('\u4efb\u52a1\u7b49\u5f85\u4e2d\u88ab\u6267\u884c')\nelif async.status == 'RETRY':\n    print('\u4efb\u52a1\u5f02\u5e38\u540e\u6b63\u5728\u91cd\u8bd5')\nelif async.status == 'STARTED':\n    print('\u4efb\u52a1\u5df2\u7ecf\u5f00\u59cb\u88ab\u6267\u884c')<\/code><\/pre>\n
\u6267\u884c add_task.py\uff0c\u6dfb\u52a0\u4efb\u52a1\uff0c\u5e76\u83b7\u53d6\u4efb\u52a1ID<\/p>\n
\u6267\u884c run.py \uff0c\u6216\u8005\u6267\u884c\u547d\u4ee4\uff1acelery worker -A celery_app_task -l info<\/p>\n
\u6267\u884c result.py,\u68c0\u67e5\u4efb\u52a1\u72b6\u6001\u5e76\u83b7\u53d6\u7ed3\u679c<\/p>\n
\u591a\u4efb\u52a1\u7ed3\u6784<\/h4>\n
\"\"<\/div><\/p>\n
pro_cel\n    \u251c\u2500\u2500 celery_task# celery\u76f8\u5173\u6587\u4ef6\u5939\n    \u2502   \u251c\u2500\u2500 celery.py   # celery\u8fde\u63a5\u548c\u914d\u7f6e\u76f8\u5173\u6587\u4ef6,\u5fc5\u987b\u53eb\u8fd9\u4e2a\u540d\u5b57\n    \u2502   \u2514\u2500\u2500 tasks1.py    #  \u6240\u6709\u4efb\u52a1\u51fd\u6570\n    \u2502   \u2514\u2500\u2500 tasks2.py    #  \u6240\u6709\u4efb\u52a1\u51fd\u6570\n    \u251c\u2500\u2500 check_result.py # \u68c0\u67e5\u7ed3\u679c\n    \u2514\u2500\u2500 send_task.py    # \u89e6\u53d1\u4efb\u52a1<\/code><\/pre>\n
celery.py<\/p>\n
from celery import Celery\n\ncel = Celery('celery_demo',\n             broker='redis:\/\/127.0.0.1:6379\/1',\n             backend='redis:\/\/127.0.0.1:6379\/2',\n             # \u5305\u542b\u4ee5\u4e0b\u4e24\u4e2a\u4efb\u52a1\u6587\u4ef6\uff0c\u53bb\u76f8\u5e94\u7684py\u6587\u4ef6\u4e2d\u627e\u4efb\u52a1\uff0c\u5bf9\u591a\u4e2a\u4efb\u52a1\u505a\u5206\u7c7b\n             include=['celery_task.tasks1',\n                      'celery_task.tasks2'\n                      ])\n\n# \u65f6\u533a\ncel.conf.timezone = 'Asia\/Shanghai'\n# \u662f\u5426\u4f7f\u7528UTC\ncel.conf.enable_utc = False<\/code><\/pre>\n
tasks1.py<\/p>\n
import time\nfrom celery_task.celery import cel\n\n@cel.task\ndef test_celery(res):\n    time.sleep(5)\n    return "test_celery\u4efb\u52a1\u7ed3\u679c:%s"%res<\/code><\/pre>\n
tasks2.py<\/p>\n
import time\nfrom celery_task.celery import cel\n@cel.task\ndef test_celery2(res):\n    time.sleep(5)\n    return "test_celery2\u4efb\u52a1\u7ed3\u679c:%s"%res<\/code><\/pre>\n
check_result.py<\/p>\n
from celery.result import AsyncResult\nfrom celery_task.celery import cel\n\nasync = AsyncResult(id="08eb2778-24e1-44e4-a54b-56990b3519ef", app=cel)\n\nif async.successful():\n    result = async.get()\n    print(result)\n    # result.forget() # \u5c06\u7ed3\u679c\u5220\u9664,\u6267\u884c\u5b8c\u6210\uff0c\u7ed3\u679c\u4e0d\u4f1a\u81ea\u52a8\u5220\u9664\n    # async.revoke(terminate=True)  # \u65e0\u8bba\u73b0\u5728\u662f\u4ec0\u4e48\u65f6\u5019\uff0c\u90fd\u8981\u7ec8\u6b62\n    # async.revoke(terminate=False) # \u5982\u679c\u4efb\u52a1\u8fd8\u6ca1\u6709\u5f00\u59cb\u6267\u884c\u5462\uff0c\u90a3\u4e48\u5c31\u53ef\u4ee5\u7ec8\u6b62\u3002\nelif async.failed():\n    print('\u6267\u884c\u5931\u8d25')\nelif async.status == 'PENDING':\n    print('\u4efb\u52a1\u7b49\u5f85\u4e2d\u88ab\u6267\u884c')\nelif async.status == 'RETRY':\n    print('\u4efb\u52a1\u5f02\u5e38\u540e\u6b63\u5728\u91cd\u8bd5')\nelif async.status == 'STARTED':\n    print('\u4efb\u52a1\u5df2\u7ecf\u5f00\u59cb\u88ab\u6267\u884c')<\/code><\/pre>\n
send_task.py<\/p>\n
from celery_task.tasks1 import test_celery\nfrom celery_task.tasks2 import test_celery2\n\n# \u7acb\u5373\u544a\u77e5celery\u53bb\u6267\u884ctest_celery\u4efb\u52a1\uff0c\u5e76\u4f20\u5165\u4e00\u4e2a\u53c2\u6570\nresult = test_celery.delay('\u7b2c\u4e00\u4e2a\u7684\u6267\u884c')\nprint(result.id)\nresult = test_celery2.delay('\u7b2c\u4e8c\u4e2a\u7684\u6267\u884c')\nprint(result.id)<\/code><\/pre>\n
\u6dfb\u52a0\u4efb\u52a1\uff08\u6267\u884csend_task.py\uff09\uff0c\u5f00\u542fwork\uff1acelery worker -A celery_task -l info  -P  eventlet\uff0c\u68c0\u67e5\u4efb\u52a1\u6267\u884c\u7ed3\u679c\uff08\u6267\u884ccheck_result.py\uff09<\/p>\n
\"\"<\/div><\/p>\n
5.Celery\u6267\u884c\u5b9a\u65f6\u4efb\u52a1<\/h2>\n
\"\"<\/div><\/p>\n
\u8bbe\u5b9a\u65f6\u95f4\u8ba9celery\u6267\u884c\u4e00\u4e2a\u4efb\u52a1<\/h4>\n
add_task.py<\/p>\n
from celery_app_task import add\nfrom datetime import datetime\n\n# \u65b9\u5f0f\u4e00\n# v1 = datetime(2019, 2, 13, 18, 19, 56)\n# print(v1)\n# v2 = datetime.utcfromtimestamp(v1.timestamp())\n# print(v2)\n# result = add.apply_async(args=[1, 3], eta=v2)\n# print(result.id)\n\n# \u65b9\u5f0f\u4e8c\nctime = datetime.now()\n# \u9ed8\u8ba4\u7528utc\u65f6\u95f4\nutc_ctime = datetime.utcfromtimestamp(ctime.timestamp())\nfrom datetime import timedelta\ntime_delay = timedelta(seconds=10)\ntask_time = utc_ctime + time_delay\n\n# \u4f7f\u7528apply_async\u5e76\u8bbe\u5b9a\u65f6\u95f4\nresult = add.apply_async(args=[4, 3], eta=task_time)\nprint(result.id)<\/code><\/pre>\n
\u7c7b\u4f3c\u4e8econtab\u7684\u5b9a\u65f6\u4efb\u52a1<\/h4>\n
\u591a\u4efb\u52a1\u7ed3\u6784\u4e2dcelery.py\u4fee\u6539\u5982\u4e0b<\/p>\n
from datetime import timedelta\nfrom celery import Celery\nfrom celery.schedules import crontab\n\ncel = Celery('tasks', broker='redis:\/\/127.0.0.1:6379\/1', backend='redis:\/\/127.0.0.1:6379\/2', include=[\n    'celery_task.tasks1',\n    'celery_task.tasks2',\n])\ncel.conf.timezone = 'Asia\/Shanghai'\ncel.conf.enable_utc = False\n\ncel.conf.beat_schedule = {\n    # \u540d\u5b57\u968f\u610f\u547d\u540d\n    'add-every-10-seconds': {\n        # \u6267\u884ctasks1\u4e0b\u7684test_celery\u51fd\u6570\n        'task': 'celery_task.tasks1.test_celery',\n        # \u6bcf\u96942\u79d2\u6267\u884c\u4e00\u6b21\n        # 'schedule': 1.0,\n        # 'schedule': crontab(minute="*\/1"),\n        'schedule': timedelta(seconds=2),\n        # \u4f20\u9012\u53c2\u6570\n        'args': ('test',)\n    },\n    # 'add-every-12-seconds': {\n    #     'task': 'celery_task.tasks1.test_celery',\n    #     \u6bcf\u5e744\u670811\u53f7\uff0c8\u70b942\u5206\u6267\u884c\n    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),\n    #     'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4),\n    #     'args': (16, 16)\n    # },\n}<\/code><\/pre>\n
\u542f\u52a8\u4e00\u4e2abeat\uff1acelery beat -A celery_task -l info<\/p>\n
\u542f\u52a8work\u6267\u884c\uff1acelery worker -A celery_task -l info -P  eventlet<\/p>\n
\"\"<\/div><\/p>\n
6.Django\u4e2d\u4f7f\u7528Celery<\/h2>\n
\u5728\u9879\u76ee\u76ee\u5f55\u4e0b\u521b\u5efaceleryconfig.py<\/p>\n
import djcelery\ndjcelery.setup_loader()\nCELERY_IMPORTS=(\n    'app01.tasks',\n)\n#\u6709\u4e9b\u60c5\u51b5\u53ef\u4ee5\u9632\u6b62\u6b7b\u9501\nCELERYD_FORCE_EXECV=True\n# \u8bbe\u7f6e\u5e76\u53d1worker\u6570\u91cf\nCELERYD_CONCURRENCY=4\n#\u5141\u8bb8\u91cd\u8bd5\nCELERY_ACKS_LATE=True\n# \u6bcf\u4e2aworker\u6700\u591a\u6267\u884c100\u4e2a\u4efb\u52a1\u88ab\u9500\u6bc1\uff0c\u53ef\u4ee5\u9632\u6b62\u5185\u5b58\u6cc4\u6f0f\nCELERYD_MAX_TASKS_PER_CHILD=100\n# \u8d85\u65f6\u65f6\u95f4\nCELERYD_TASK_TIME_LIMIT=12*30<\/code><\/pre>\n
\u5728app01\u76ee\u5f55\u4e0b\u521b\u5efatasks.py<\/p>\n
from celery import task\n@task\ndef add(a,b):\n    with open('a.text', 'a', encoding='utf-8') as f:\n        f.write('a')\n    print(a+b)<\/code><\/pre>\n
\u89c6\u56fe\u51fd\u6570views.py<\/p>\n
from django.shortcuts import render,HttpResponse\nfrom app01.tasks import add\nfrom datetime import datetime\ndef test(request):\n    # result=add.delay(2,3)\n    ctime = datetime.now()\n    # \u9ed8\u8ba4\u7528utc\u65f6\u95f4\n    utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())\n    from datetime import timedelta\n    time_delay = timedelta(seconds=5)\n    task_time = utc_ctime + time_delay\n    result = add.apply_async(args=[4, 3], eta=task_time)\n    print(result.id)\n    return HttpResponse('ok')<\/code><\/pre>\n
settings.py<\/p>\n
\nINSTALLED_APPS = [\n    ...\n    'djcelery',\n    'app01'\n]\n\n...\n\nfrom djagocele import celeryconfig\nBROKER_BACKEND='redis'\nBOOKER_URL='redis:\/\/127.0.0.1:6379\/1'\nCELERY_RESULT_BACKEND='redis:\/\/127.0.0.1:6379\/2'<\/code><\/pre>\n
\"\"<\/div><\/p>\n","protected":false},"excerpt":{"rendered":"
Celery 1.\u4ec0\u4e48\u662fClelery Celery\u662f\u4e00\u4e2a\u7b80\u5355\u3001\u7075\u6d3b\u4e14\u53ef\u9760\u7684\uff0c\u5904\u7406\u5927\u91cf\u6d88\u606f\u7684\u5206\u5e03\u5f0f\u7cfb\u7edf \u4e13\u6ce8\u4e8e […]<\/p>\n","protected":false},"author":3,"featured_media":5716,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":[],"categories":[385,387],"tags":[],"_links":{"self":[{"href":"https:\/\/egonlin.com\/index.php?rest_route=\/wp\/v2\/posts\/5714"}],"collection":[{"href":"https:\/\/egonlin.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/egonlin.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/egonlin.com\/index.php?rest_route=\/wp\/v2\/users\/3"}],"replies":[{"embeddable":true,"href":"https:\/\/egonlin.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=5714"}],"version-history":[{"count":0,"href":"https:\/\/egonlin.com\/index.php?rest_route=\/wp\/v2\/posts\/5714\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/egonlin.com\/index.php?rest_route=\/wp\/v2\/media\/5716"}],"wp:attachment":[{"href":"https:\/\/egonlin.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=5714"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/egonlin.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=5714"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/egonlin.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=5714"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}