zoukankan      html  css  js  c++  java

  • 【转】Celery 分布式任务队列快速入门

    Celery 分布式任务队列快速入门

    本节内容

    Celery介绍和基本使用

    在项目中如何使用celery

    启用多个workers

    Celery 分布式

    Celery 定时任务

    与django结合

    通过django配置celery periodic task

    一、Celery介绍和基本使用 

    Celery 是一个 基于python开发的分布式异步消息任务队列，通过它可以轻松的实现任务的异步处理， 如果你的业务场景中需要用到异步任务，就可以考虑使用celery， 举几个实例场景中可用的例子:

    1. 你想对100台机器执行一条批量命令，可能会花很长时间 ，但你不想让你的程序等着结果返回，而是给你返回 一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果， 在任务执行ing进行时，你可以继续做其它的事情。 
    2. 你想做一个定时任务，比如每天检测一下你们所有客户的资料，如果发现今天 是客户的生日，就给他发个短信祝福

    Celery 在执行任务时需要通过一个消息中间件来接收和发送任务消息，以及存储任务结果， 一般使用rabbitMQ or Redis,后面会讲

    1.1 Celery有以下优点：

    1. 简单：一单熟悉了celery的工作流程后，配置和使用还是比较简单的
    2. 高可用：当任务执行失败或执行过程中发生连接中断，celery 会自动尝试重新执行任务
    3. 快速：一个单进程的celery每分钟可处理上百万个任务
    4. 灵活： 几乎celery的各个组件都可以被扩展及自定制

    Celery基本工作流程图

    1.2 Celery安装使用

    Celery的默认broker是RabbitMQ, 仅需配置一行就可以

    1
    broker_url = 'amqp://guest:guest@localhost:5672//'

    rabbitMQ 没装的话请装一下，安装看这里  http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html#id3

    使用Redis做broker也可以

    安装redis组件

    1
    $ pip install -U "celery[redis]"

    配置

    Configuration is easy, just configure the location of your Redis database:

    app.conf.broker_url = 'redis://localhost:6379/0'

    Where the URL is in the format of:

    redis://:password@hostname:port/db_number

    all fields after the scheme are optional, and will default to localhost on port 6379, using database 0.

    如果想获取每个任务的执行结果，还需要配置一下把任务结果存在哪

    If you also want to store the state and return values of tasks in Redis, you should configure these settings:

    app.conf.result_backend = 'redis://localhost:6379/0'


    注意：如果没有在这里进行配置，直接在代码里写上即可：
    app = Celery('tasks',
                 broker='redis://localhost',
                 backend='redis://localhost')
     

     

    1. 3 开始使用Celery啦　　

    安装celery模块

    1
    $ pip install celery

    创建一个celery application 用来定义你的任务列表

    创建一个任务文件就叫tasks.py吧

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    from celery import Celery
     
    app = Celery('tasks',
                 broker='redis://localhost',
                 backend='redis://localhost')
     
    @app.task
    def add(x,y):
        print("running...",x,y)
        return x+y

    启动Celery Worker来开始监听并执行任务

    1
    $ celery -A tasks worker --loglevel=info

    　　

    调用任务

    再打开一个终端， 进行命令行模式，调用任务　　

    1
    2
    >>> from tasks import add
    >>> add.delay(4, 4)

    看你的worker终端会显示收到 一个任务，此时你想看任务结果的话，需要在调用 任务时　赋值个变量

    1
    >>> result = add.delay(4, 4)

    The ready() method returns whether the task has finished processing or not:

    >>> result.ready()
False

    You can wait for the result to complete, but this is rarely used since it turns the asynchronous call into a synchronous one:

    >>> result.get(timeout=1)
8

    In case the task raised an exception, get() will re-raise the exception, but you can override this by specifying the propagate argument:

    >>> result.get(propagate=False)

    If the task raised an exception you can also gain access to the original traceback:

    >>> result.traceback
…

    　　

    二、在项目中如何使用celery　

    可以把celery配置成一个应用

    目录格式如下

    1
    2
    3
    proj/__init__.py
        /celery.py
        /tasks.py

    proj/celery.py内容

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    from __future__ import absolute_import, unicode_literals
    from celery import Celery
     
    app = Celery('proj',
                 broker='amqp://',
                 backend='amqp://',
                 include=['proj.tasks'])
     
    # Optional configuration, see the application user guide.
    app.conf.update(
        result_expires=3600,
    )
     
    if __name__ == '__main__':
        app.start()

    proj/tasks.py中的内容

    复制代码
    from __future__ import absolute_import, unicode_literals
from .celery import app


@app.task
def add(x, y):
    return x + y


@app.task
def mul(x, y):
    return x * y


@app.task
def xsum(numbers):
    return sum(numbers)
    复制代码

    启动worker (注意在proj的父目录下启动，否则报错)

    1
    $ celery -A proj worker -l info

    输出

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    -------------- celery@Alexs-MacBook-Pro.local v4.0.2 (latentcall)
    ---- **** -----
    --- * ***  * -- Darwin-15.6.0-x86_64-i386-64bit 2017-01-26 21:50:24
    -- * - **** ---
    - ** ---------- [config]
    - ** ---------- .> app:         proj:0x103a020f0
    - ** ---------- .> transport:   redis://localhost:6379//
    - ** ---------- .> results:     redis://localhost/
    - *** --- * --- .> concurrency: 8 (prefork)
    -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
    --- ***** -----
     -------------- [queues]
                    .> celery           exchange=celery(direct) key=celery

    　　

    后台启动worker

    In the background

    In production you’ll want to run the worker in the background, this is described in detail in the daemonization tutorial.

    The daemonization scripts uses the celery multi command to start one or more workers in the background:

    $ celery multi start w1 -A proj -l info
celery multi v4.0.0 (latentcall)
> Starting nodes...
    > w1.halcyon.local: OK

    You can restart it too:

    $ celery  multi restart w1 -A proj -l info
celery multi v4.0.0 (latentcall)
> Stopping nodes...
    > w1.halcyon.local: TERM -> 64024
> Waiting for 1 node.....
    > w1.halcyon.local: OK
> Restarting node w1.halcyon.local: OK
celery multi v4.0.0 (latentcall)
> Stopping nodes...
    > w1.halcyon.local: TERM -> 64052

    or stop it:

    $ celery multi stop w1 -A proj -l info

    The stop command is asynchronous so it won’t wait for the worker to shutdown. You’ll probably want to use the stopwait command instead, this ensures all currently executing tasks is completed before exiting:

    $ celery multi stopwait w1 -A proj -l info

    　　

    三、分布式Celery

    大家都知道Celery是分布式任务队列，可是在学习过程中，发现基本都是怎么用celery去异步操作，很少有讲到如何分布式。于是把自己的采坑过程分享给大家。

    A、准备工作

    1.选择一个消息中间件Broker

    官方推荐的有RabbitMQ和Redis，二选一安装。具体安装可参考
    http://docs.celeryproject.org/en/latest/getting-started/first-steps-with-celery.html#first-steps
    本人选择Redis做Broker.

    2.准备一个虚拟机

    确保本机和虚拟机都可以连上redis。有其他网络互通的机器也可以。

    B、简单的demo

    1.创建项目和python环境

    mkdir myCelery
cd myCelery
virtualenv -p python2 envp2
source envp2/bin/activate

    2.安装Celery和Redis

    pip install celery==4.0 
pip install redis

    请注意这里安装的是4.0.按照官方的说法4.0是既支持Python2又支持python3的版本。官方内容如下：

    Version Requirements

Celery version 4.0 runs on

Python ❨2.7, 3.4, 3.5❩
PyPy ❨5.4, 5.5❩
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.

If you’re running an older version of Python, you need to be running an older version of Celery:

Python 2.6: Celery series 3.1 or earlier.
Python 2.5: Celery series 3.0 or earlier.
Python 2.4 was Celery series 2.2 or earlier.
Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.

    3.码代码

    vim tasks.py

    conding:

    from celery import Celery

app = Celery('tasks', broker='redis://10.211.55.2:6379/0')

@app.task
def add(x, y):
    return x + y

    4.运行

    把tasks作为worker运行起来

    celery -A tasks worker -l info

    5.发起任务

    另开一个terminal

    cd myCelery
envp2/bin/pyton 
#python terminal 下
>>> from tasks import add
>>> add.delay(4, 4)

    观察运行tasks的terminal，有结果8出来就证明一切正常。熟悉了简单的流程后，我们就在这个基础上进一步深入。

    C、分布式celery
    开启虚拟机，把tasks.py拷贝过去，python环境配置完成后，打开terminal。运行tasks

    celery -A tasks worker -l info

    这时观察本机运行tasks的terminal,多输出如下内容：

    [2018-01-24 21:03:10,114: INFO/MainProcess]  sync with celery@ubuntu

    而虚拟机上的terminal将会输出如下内容：

    [2018-01-24 21:03:09,007: INFO/MainProcess] Connected to redis://10.211.55.2:6379/0
[2018-01-24 21:03:09,017: INFO/MainProcess] mingle: searching for neighbors
[2018-01-24 21:03:10,036: INFO/MainProcess] mingle: sync with 1 nodes
[2018-01-24 21:03:10,037: INFO/MainProcess] mingle: sync complete
[2018-01-24 21:03:10,052: INFO/MainProcess] celery@ubuntu ready.

    至此，集群已经成功连接。下面我们来测试一下。在本机tasks.py同级新建test_tasks.py文件,输入下面内容并新开terminal运行：

    from tasks import add
#循环执行10次add函数
for i in range(10):
    add.delay(i, i)

    来观察一下本机tasks terminal：

    [2018-01-24 21:37:18,360: INFO/MainProcess] Received task: tasks.add[33586201-082a-4e8f-8153-8b9d757990af]
[2018-01-24 21:37:18,361: INFO/ForkPoolWorker-7] Task tasks.add[33586201-082a-4e8f-8153-8b9d757990af] succeeded in 0.0004948119749315083s: 2
[2018-01-24 21:37:18,362: INFO/MainProcess] Received task: tasks.add[a6e363cf-fd25-4b0d-9da4-04bac1c5476c]
[2018-01-24 21:37:18,363: INFO/MainProcess] Received task: tasks.add[7fd2b545-f87b-49d3-941f-b8723bd1b039]
[2018-01-24 21:37:18,365: INFO/ForkPoolWorker-2] Task tasks.add[7fd2b545-f87b-49d3-941f-b8723bd1b039] succeeded in 0.000525131996255368s: 8
[2018-01-24 21:37:18,365: INFO/ForkPoolWorker-8] Task tasks.add[a6e363cf-fd25-4b0d-9da4-04bac1c5476c] succeeded in 0.000518032000400126s: 6
[2018-01-24 21:37:18,366: INFO/MainProcess] Received task: tasks.add[79d1ead3-1077-4bfd-8300-3a335f533b74]
[2018-01-24 21:37:18,368: INFO/MainProcess] Received task: tasks.add[0d0eefab-c6f0-4fa6-945c-6c7931b74e7b]
[2018-01-24 21:37:18,368: INFO/ForkPoolWorker-4] Task tasks.add[79d1ead3-1077-4bfd-8300-3a335f533b74] succeeded in 0.00042340101208537817s: 12
[2018-01-24 21:37:18,369: INFO/MainProcess] Received task: tasks.add[230eb9d1-7fa5-4f18-8fd4-f535e4c190d2]
[2018-01-24 21:37:18,370: INFO/ForkPoolWorker-6] Task tasks.add[0d0eefab-c6f0-4fa6-945c-6c7931b74e7b] succeeded in 0.00048609700752422214s: 16
[2018-01-24 21:37:18,370: INFO/ForkPoolWorker-7] Task tasks.add[230eb9d1-7fa5-4f18-8fd4-f535e4c190d2] succeeded in 0.00046275201020762324s: 18

    在来看一下虚拟机上的terminal：

    [2018-01-24 21:37:17,261: INFO/MainProcess] Received task: tasks.add[f95a4a20-e245-4cdc-b48a-4b79416b14c1]  
[2018-01-24 21:37:17,263: INFO/ForkPoolWorker-1] Task tasks.add[f95a4a20-e245-4cdc-b48a-4b79416b14c1] succeeded in 0.00107001400102s: 0
[2018-01-24 21:37:17,264: INFO/MainProcess] Received task: tasks.add[3ddfbcda-7d75-488c-bc69-243f991bb49a]  
[2018-01-24 21:37:17,267: INFO/MainProcess] Received task: tasks.add[b0a36bfe-87c4-43ef-9e6e-fb8740dd26d0]  
[2018-01-24 21:37:17,267: INFO/ForkPoolWorker-1] Task tasks.add[3ddfbcda-7d75-488c-bc69-243f991bb49a] succeeded in 0.00107501300226s: 4
[2018-01-24 21:37:17,270: INFO/ForkPoolWorker-2] Task tasks.add[b0a36bfe-87c4-43ef-9e6e-fb8740dd26d0] succeeded in 0.00111001500045s: 10
[2018-01-24 21:37:17,272: INFO/MainProcess] Received task: tasks.add[7bcec842-65e5-407d-9e7d-99183956ef3e]  
[2018-01-24 21:37:17,277: INFO/ForkPoolWorker-1] Task tasks.add[7bcec842-65e5-407d-9e7d-99183956ef3e] succeeded in 0.000870012001542s: 14

    我们只观察succeeded in 及后面的值,本机输出了： 2 8 6 12 16 18
    虚拟机上输出了：0 4 10 14
    充分证明了10个add任务已经被分发处理了。

     

     

    四、Celery 定时任务

     celery支持定时任务，设定好任务的执行时间，celery就会定时自动帮你执行， 这个定时任务模块叫celery beat

     
    写一个脚本 叫periodic_task.py
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    from celery import Celery
    from celery.schedules import crontab
     
    app = Celery()
     
    @app.on_after_configure.connect
    def setup_periodic_tasks(sender, **kwargs):
        # Calls test('hello') every 10 seconds.
        sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')
     
        # Calls test('world') every 30 seconds
        sender.add_periodic_task(30.0, test.s('world'), expires=10)
     
        # Executes every Monday morning at 7:30 a.m.
        sender.add_periodic_task(
            crontab(hour=7, minute=30, day_of_week=1),
            test.s('Happy Mondays!'),
        )
     
    @app.task
    def test(arg):
        print(arg)

    add_periodic_task 会添加一条定时任务

    上面是通过调用函数添加定时任务，也可以像写配置文件 一样的形式添加， 下面是每30s执行的任务

    1
    2
    3
    4
    5
    6
    7
    8
    app.conf.beat_schedule = {
        'add-every-30-seconds': {
            'task': 'tasks.add',
            'schedule': 30.0,
            'args': (16, 16)
        },
    }
    app.conf.timezone = 'UTC'

    　　

    任务添加好了，需要让celery单独启动一个进程来定时发起这些任务， 注意， 这里是发起任务，不是执行，这个进程只会不断的去检查你的任务计划， 每发现有任务需要执行了，就发起一个任务调用消息，交给celery worker去执行

    启动任务调度器 celery beat

    1
    $ celery -A periodic_task beat

    输出like below

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    celery beat v4.0.2 (latentcall) is starting.
    __    -    ... __   -        _
    LocalTime -> 2017-02-08 18:39:31
    Configuration ->
        . broker -> redis://localhost:6379//
        . loader -> celery.loaders.app.AppLoader
        . scheduler -> celery.beat.PersistentScheduler
        . db -> celerybeat-schedule
        . logfile -> [stderr]@%WARNING
        . maxinterval -> 5.00 minutes (300s)

    此时还差一步，就是还需要启动一个worker,负责执行celery beat发起的任务

    启动celery worker来执行任务

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    $ celery -A periodic_task worker
      
     -------------- celery@Alexs-MacBook-Pro.local v4.0.2 (latentcall)
    ---- **** -----
    --- * ***  * -- Darwin-15.6.0-x86_64-i386-64bit 2017-02-08 18:42:08
    -- * - **** ---
    - ** ---------- [config]
    - ** ---------- .> app:         tasks:0x104d420b8
    - ** ---------- .> transport:   redis://localhost:6379//
    - ** ---------- .> results:     redis://localhost/
    - *** --- * --- .> concurrency: 8 (prefork)
    -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
    --- ***** -----
     -------------- [queues]
                    .> celery           exchange=celery(direct) key=celery

    　　

    好啦，此时观察worker的输出，是不是每隔一小会，就会执行一次定时任务呢！

    注意：Beat needs to store the last run times of the tasks in a local database file (named celerybeat-schedule by default), so it needs access to write in the current directory, or alternatively you can specify a custom location for this file:

    1
    $ celery -A periodic_task beat -s /home/celery/var/run/celerybeat-schedule

    更复杂的定时配置　　

    上面的定时任务比较简单，只是每多少s执行一个任务，但如果你想要每周一三五的早上8点给你发邮件怎么办呢？哈，其实也简单，用crontab功能，跟linux自带的crontab功能是一样的，可以个性化定制任务执行时间

    linux crontab http://www.cnblogs.com/peida/archive/2013/01/08/2850483.html 

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    from celery.schedules import crontab
     
    app.conf.beat_schedule = {
        # Executes every Monday morning at 7:30 a.m.
        'add-every-monday-morning': {
            'task': 'tasks.add',
            'schedule': crontab(hour=7, minute=30, day_of_week=1),
            'args': (16, 16),
        },
    }

    上面的这条意思是每周1的早上7.30执行tasks.add任务

    还有更多定时配置方式如下：

    Example Meaning
    crontab() Execute every minute.
    crontab(minute=0, hour=0) Execute daily at midnight.
    crontab(minute=0, hour='*/3') Execute every three hours: midnight, 3am, 6am, 9am, noon, 3pm, 6pm, 9pm.
    crontab(minute=0,
    hour='0,3,6,9,12,15,18,21')
    		 Same as previous.
    crontab(minute='*/15') Execute every 15 minutes.
    crontab(day_of_week='sunday') Execute every minute (!) at Sundays.
    crontab(minute='*',
    hour='*',day_of_week='sun')
    		 Same as previous.
    crontab(minute='*/10',
    hour='3,17,22',day_of_week='thu,fri')
    		 Execute every ten minutes, but only between 3-4 am, 5-6 pm, and 10-11 pm on Thursdays or Fridays.
    crontab(minute=0,hour='*/2,*/3') Execute every even hour, and every hour divisible by three. This means: at every hour except: 1am, 5am, 7am, 11am, 1pm, 5pm, 7pm, 11pm
    crontab(minute=0, hour='*/5') Execute hour divisible by 5. This means that it is triggered at 3pm, not 5pm (since 3pm equals the 24-hour clock value of “15”, which is divisible by 5).
    crontab(minute=0, hour='*/3,8-17') Execute every hour divisible by 3, and every hour during office hours (8am-5pm).
    crontab(0, 0,day_of_month='2') Execute on the second day of every month.
    crontab(0, 0,
    day_of_month='2-30/3')
    		 Execute on every even numbered day.
    crontab(0, 0,
    day_of_month='1-7,15-21')
    		 Execute on the first and third weeks of the month.
    crontab(0, 0,day_of_month='11',
    month_of_year='5')
    		 Execute on the eleventh of May every year.
    crontab(0, 0,
    month_of_year='*/3')
    		 Execute on the first month of every quarter.

    上面能满足你绝大多数定时任务需求了，甚至还能根据潮起潮落来配置定时任务， 具体看 http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html#solar-schedules 　　

    　　　

    　　

    五、最佳实践之与django结合 

    django 可以轻松跟celery结合实现异步任务，只需简单配置即可

    创建django工程proj,创建app（demoapp）

    - proj/
  - proj/__init__.py
  - proj/settings.py
  - proj/urls.py
- manage.py
    - demoapp/
        - models.py
     
    在setting文件配置demoapp：
    INSTALLED_APPS = [
        'django.contrib.admin',
        'django.contrib.auth',
        'django.contrib.contenttypes',
        'django.contrib.sessions',
        'django.contrib.messages',
        'django.contrib.staticfiles',
        'demoapp',
    ]
    在demoapp下新建tasks文件
    file: proj/demoapp/task.py
    # Create your tasks here
    from __future__ import absolute_import, unicode_literals
    from celery import shared_task

    @shared_task
    def add(x, y):
        return x + y

    @shared_task
    def mul(x, y):
        return x * y

    @shared_task
    def xsum(numbers):
        return sum(numbers)

    然后创建一个新的文件：proj / proj / celery.py来定义Celery实例：

    file: proj/proj/celery.py　　

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    from __future__ import absolute_import, unicode_literals
    import os
    from celery import Celery
     
    # set the default Django settings module for the 'celery' program.
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
     
    app = Celery('proj',
    　　　　　　　　broker = 'redis://localhost',
    　　　　　　　　backend = 'redis://localhost',　　　　　
    )
     
    # Using a string here means the worker don'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')
     
    # Load task modules from all registered Django app configs.
    app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
     
    @app.task(bind=True)
    def debug_task(self):
        print('Request: {0!r}'.format(self.request))

    然后你需要在你的proj / proj / __ init__.py模块中导入这个应用程序。这确保了当Django启动时加载应用程序，以便@shared_task装饰器（稍后提到）将使用它：

    proj/proj/__init__.py:

    1
    2
    3
    4
    5
    6
    7
    from __future__ import absolute_import, unicode_literals
     
    # This will make sure the app is always imported when
    # Django starts so that shared_task will use this app.
    from .celery import app as celery_app
     
    __all__ = ['celery_app']

    ------下面是对celery.py和__init__.py程序的解释---------------------------------------------------------------------------------------------

    请注意，此示例项目布局适用于较大的项目，对于简单的项目，您可以使用单个包含的模块来定义应用程序和任务，如使用Celery的第一步教程。

    让我们分解第一个模块中发生的事情，首先我们从__future__导入绝对absolute_import，这样我们的celery.py就不会和第三方的celery库冲突：

    1
    from __future__ import absolute_import

    然后我们为celery命令行程序设置默认的DJANGO_SETTINGS_MODULE环境变量

    1
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

    以下这一行不是必须的，但它可以使得您时常将setting模块传递给celery程序。它必须始终在创建应用程序实例之前，我们接下来会这样做：

    1
    app = Celery('proj')

    This is our instance of the library.

     将Django的setting模块添加为Celery的配置文件。这意味着您不必使用多个配置文件，而是直接从Django设置配置Celery;但如果需要，你也可以将它们分开。

    用大写名称空间表示必须以大写而不是小写指定所有Celery配置选项，并以CELERY_开头，因此例如task_always_eager`设置变为CELERY_TASK_ALWAYS_EAGER，并且broker_url设置变为CELERY_BROKER_URL。您可以直接在此处传递对象，但使用字符串更好，因为工作人员不必序列化对象。

    1
    app.config_from_object('django.conf:settings', namespace='CELERY')

    接下来，可重用应用程序的一个常见做法是在单独的tasks.py module中定义所有任务，Celery通过以下办法自动发现这些模块：

    1
    app.autodiscover_tasks()

    ------上面是对celery.py和__init__.py程序的解释---------------------------------------------------------------------------------------------

    在你的django views里调用celery task

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    from django.shortcuts import render,HttpResponse
     
    # Create your views here.
     
    from  demoapp import tasks
     
    def task_test(request):
     
        res = tasks.add.delay(228,24)
        print("start running task")
        print("async task res",res.get() )
     
        return HttpResponse('res %s'%res.get())

    　　

    启动django 服务：

    python3 manage.py runserver 0.0.0.0:8081

    启动Celery Worker来开始监听并执行任务

    1
    $ celery -A tasks worker --loglevel=info

    　　

    通过访问view调用任务：

    http://0.0.0.0:8081/myapp/（访问views的task_test函数的url）

    　

     

    　　　

    六、在django中使用计划任务功能　　

    There’s  the django-celery-beat extension that stores the schedule in the Django database, and presents a convenient admin interface to manage periodic tasks at runtime.

    To install and use this extension:

    1. Use pip to install the package:

      $ pip install django-celery-beat

    2. Add the django_celery_beat module to INSTALLED_APPS in your Django project’ settings.py:

          INSTALLED_APPS = (
        ...,
        'django_celery_beat',
    )

Note that there is no dash in the module name, only underscores.

    3. Apply Django database migrations so that the necessary tables are created:

      $ python manage.py migrate

    4. Start the celery beat service using the django scheduler:

      $ celery -A proj beat -l info -S django

    5. Visit the Django-Admin interface to set up some periodic tasks.

    　　

    在admin页面里，有3张表

    配置完长这样

    此时启动你的celery beat 和worker：celery -A cel  beat -l debug -S django

    会发现每隔2分钟，beat会发起一个任务消息让worker执行scp_task任务

    注意，经测试，每添加或修改一个任务，celery beat都需要重启一次，要不然新的配置不会被celery beat进程读到
  • 相关阅读:
    UML实践
    “学士之路”系统设计
    团队采访
    《软件需求规格说明书》 ---学士之路
    奇怪的bug，不懂Atom在添加markdown-themeable-pdf，在配置好phantomjs的情况下报错
    团队项目-“学士之路”展示
    学习Mybatis的两个必须的jar包分享
    Spring MVC controller的方法返回值
    Spring的Controller映射规则
    servlet基础学习总结
  • 原文地址：https://www.cnblogs.com/yoyo008/p/9282491.html
Copyright © 2011-2022 走看看