celery 任务队列初步探索

比眉伴天荒 2022-05-31 02:53 229阅读 0赞

# Celery 简介 #

什么是Celery：  
celery（任务队列）是一种在线程或机器间分发任务的机制。  
celery中文文档：docs.jinkan.org/docs/celery/index.html  
celery官网：docs.celeryproject.org  
celery的主要功能就是任务分发，可以简单的看作是一个由生产者、消费者以及broker（中间人）组成的分布式工作小组。生产者就是产生任务的对像，消费者就是处理任务的职程。他们之间由中间人来联系，产生的任务被存储在broker中分发给消费者处理。

**开始第一个celery小demo：**

# consumer.py
    from celery import Celery
    
    app = Celery('calculate',broker='amqp://guest@localhost//')
    
    @app.task
    def calculate(a,b):
            return a+b

这是一个简单的消费者，作用是将生产者产生的两个参数进行求和计算，这里的broker指的是中间人选用 rabbitmq 。

> broker有多种选择，常用的主要有 rabbitmq 以及 redis （redis://localhost）。这些是需要自己手动在计算机中配置安装的。

启动职程：celery -A consumer worker –loglevel=info  
![这里写图片描述][SouthEast]

#product.py
    from consumer import add
    for i in range(1000):
        for j in range(1000):
            add.delay(i,j)

这是一个产生任务的生产者，总共产生 1000^2 条任务，当 product.py 运行时，产生的任务序列将会被推送到 broker 队列中分配给所有的消费者处理。  
![这里写图片描述][SouthEast 1]  
**保存处理结果：**  
当想要保持追踪任务的状态，celery中可以加入一个 backend 设置，这需要在某个地方来存储或者是发送这些状态，如：orm，Memcached，redis，rabbitmq或者mongo等

> app = Celery(‘consumer’,backend=’redis://localhost’,broker=’amqp://’)

**配置：**  
例如修改默认的序列化配置  
app.conf.CELERY\_TASK\_SERIALIZER = ‘json’  
或者是一下修改多条配置  
`python app.conf.update( CELERY_TASK_SERIALIZER = ‘json’， CELERY_ACCEPT_CONTENT = [‘json’]，#接受数据格式 CELERY_RESULT_SERIALIZER = 'json'， CELERY_TIMEZONE='Europe/Oslo'， CELERY_ENABLE_UTC=True， )`

[SouthEast]: /images/20220531/e2d6f3e5c67a410399677aefc3036739.png
[SouthEast 1]: /images/20220531/7864e2f2cd3a4b4386c65d081c61bd18.png