zoukankan      html  css  js  c++  java
  • Celery 分布式任务队列

    一、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, 仅需配置一行就可以

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

    rabbitMQ 需要安装

    使用Redis做broker也可以

    安装redis组件

    $ pip install -U "celery[redis]"

    配置

     Redis database:

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

    Where the URL is in the format of:

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

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

    app.conf.result_backend = 'redis://localhost:6379/0'
    1. 3 使用Celery 

    安装celery模块

     pip install celery

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

    创建一个任务文件tasks.py

    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来开始监听并执行任务

     celery -A tasks worker --loglevel=info

    调用任务

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

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

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

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

    二、在项目中如何使用celery 

    可以把celery配置成一个应用

    目录格式如下

    proj/__init__.py
        /celery.py
        /tasks.py

    proj/celery.py内容

    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 

    $ celery -A proj worker -l inf

    后台启动worker

    $ celery multi start w1 -A proj -l info

    $ celery multi restart w1 -A proj -l info

    or stop it:

    $ celery multi stop w1 -A proj -l info
    $ celery multi stopwait w1 -A proj -l info

    三、Celery 定时任务

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

     
    写一个脚本 叫periodic_task.py
    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 会添加一条定时任务

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

    app.conf.beat_schedule = {
        'add-every-30-seconds': {
            'task': 'tasks.add',
            'schedule': 20.0,
            'args': (16, 16)
        },
    }
    app.conf.timezone = 'UTC'

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

    启动任务调度器 celery beat

    $ celery -A periodic_task beat

    输出like below

    celery beat v4.0.2 (latentcall) is starting.
    __    -    ... __   -        _
    LocalTime -> 2017-03-08 16: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来执行任务

    $ celery -A periodic_task worker

    此时观察worker的输出,每隔一小会,就会执行一次定时任务

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

    更复杂的定时配置  

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

    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),
        },
    }

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

    四、最佳实践之与django结合 

    描述:实现运维平台的异步执行与定时任务,以下简单描述了安装过程及使用。
     
    安装django和celery
    pip install django
    pip install celery
    pip install django-celery
    新建一个项目名为news
    root@bogon:~# django-admin startproject news
    查看目录树
    root@bogon:~# tree news/
    news/
    ├── manage.py
    └── news
    ├── __init__.py
    ├── settings.py
    ├── urls.py
    └── wsgi.py
    定义一个celery实例 news/news/celery.py
    from __future__ import absolute_import
      
    import os
      
    from celery import Celery
      
    # set the default Django settings module for the 'celery' program.
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'news.settings')
      
    from django.conf import settings
      
    app = Celery('news')
      
    # Using a string here means the worker will not have to
    # pickle the object when using Windows.
    app.config_from_object('django.conf:settings')
    app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
      
      
    @app.task(bind=True)
    def debug_task(self):
      print('Request: {0!r}'.format(self.request))
    在django中加载应用news/news/__init__.py
    from __future__ import absolute_import
      
    # 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
     
    celery基本配置
    import djcelery
    djcelery.setup_loader()
      
    BROKER_URL = 'django://'
    # 以下为mq配置
    # BROKER_URL = "amqp://guest:guest@localhost:5672//"
      
    INSTALLED_APPS = (
      'djcelery',
      'kombu.transport.django',
    celery队列配置以及详细说明
    # 以下是标准的redis配置
    BROKER_URL = 'redis://localhost:6379'
    CELERY_RESULT_BACKEND = 'redis://localhost:6379'
    CELERY_ACCEPT_CONTENT = ['application/json']
    CELERY_TASK_SERIALIZER = 'json'
    CELERY_RESULT_SERIALIZER = 'json'
    CELERY_TIMEZONE = 'Asia/Shanghai'
      
      
    # rabbitmq 配置
    # 官网优化的地方也推荐使用c的librabbitmq
    CELERY_RESULT_BACKEND = "amqp"
    # celery任务执行结果的超时时间
    CELERY_TASK_RESULT_EXPIRES = 1200
    # celery worker的并发数
    CELERYD_CONCURRENCY = 50
    # celery worker 每次去rabbitmq取任务的数量
    CELERYD_PREFETCH_MULTIPLIER = 4
    # 每个worker执行了多少任务就会死掉
    CELERYD_MAX_TASKS_PER_CHILD = 40
    # 这是使用了django-celery默认的数据库调度模型,任务执行周期都被存在你指定的orm数据库中
    CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
    # 我做的rabbitmq和celery监控很完善所以这个任务超时时间比较短只有半小时
    CELERYD_TASK_TIME_LIMIT = 1800
    # 默认的队列,如果一个消息不符合其他的队列就会放在默认队列里面
    CELERY_DEFAULT_QUEUE = "default_dongwm"
      
    ## 以下是队列的一些配置
    CELERY_QUEUES = {
      "default_dongwm": { # 这是上面指定的默认队列
        "exchange": "default_dongwm",
        "exchange_type": "direct",
        "routing_key": "default_dongwm"
      },
      "topicqueue": { # 这是一个topic队列 凡是topictest开头的routing key都会被放到这个队列
        "routing_key": "topictest.#",
        "exchange": "topic_exchange",
        "exchange_type": "topic",
      },
      "test2": { # test和test2是2个fanout队列,注意他们的exchange相同
        "exchange": "broadcast_tasks",
        "exchange_type": "fanout",
        "binding_key": "broadcast_tasks",
      },
      "test": {
        "exchange": "broadcast_tasks",
        "exchange_type": "fanout",
        "binding_key": "broadcast_tasks2",
      },
    }
    <br>class MyRouter(object):
      def route_for_task(self, task, args=None, kwargs=None):
        if task.startswith('topictest'):
          return {
            'queue': 'topicqueue',
            }
        # 我的dongwm.tasks文件里面有2个任务都是test开头
        elif task.startswith('webui.tasks.test'):
          return {
            "exchange": "broadcast_tasks",
          }
        # 剩下的其实就会被放到默认队列
        else:
          return None
    # CELERY_ROUTES本来也可以用一个大的含有多个字典的字典,但是不如直接对它做一个名称统配
    CELERY_ROUTES = (MyRouter(), )
    创建一个应用和异步任务
    root@bogon:~# django-admin startapp webui
    root@bogon:~# cat webui/tasks.py
    from __future__ import absolute_import
      
    from celery import shared_task
      
    @shared_task
    def add(x, y):
      return x + y
     
    创建一个定时任务(上边的)
     

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

    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,会发现每隔2分钟,beat会发起一个任务消息让worker执行scp_task任务

    注意,经测试,每添加或修改一个任务,celery beat都需要重启一次,要不然新的配置不会被celery beat进程读到

  • 相关阅读:
    flask虚拟环境
    db.Column
    flask_cors跨域请求
    app.config.from_object
    jquery链式原理.html
    swiper轮播
    jquery引用
    animate.html
    设置和获取html里面的内容.html
    jquery获取dom属性方法
  • 原文地址:https://www.cnblogs.com/lst1010/p/6582468.html
Copyright © 2011-2022 走看看