Celery部署爬虫(二)

celery

书接上文 Celery部署爬虫(一)

Celery架构的启动都要依靠配置参数来触发，Celery简单、灵活、可靠很大程度上是建立在它的配置文件上，对于celery而言，一切皆配置。

那么。如何来编写配置参数呢，这里有几种选择

实例化参数配置

app.conf.CELERY_BROKER_URL = 'redis://localhost:6379/0'
app.conf.CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'

或者

app.conf.update(
    CELERY_BROKER_URL = 'redis://localhost:6379/0',
    CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
)

而最常见的就是模块化配置了，让配置模块化

# config.py
from __future__ import absolute_import
# broker
import datetime
from kombu import Exchange, Queue
from celery.schedules import crontab
BROKER_URL = 'redis://127.0.0.1:6379/0'
# backen
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1
# 导入任务，如tasks.py
CELERY_IMPORTS = ('tasks', )
# 列化任务载荷的默认的序列化方式
CELERY_TASK_SERIALIZER = 'json'
# 结果序列化方式
CELERY_RESULT_SERIALIZER = 'json'

CELERY_ACCEPT_CONTENT = ['json']

CELERY_TIMEZONE='Asia/Shanghai'    # 指定时区，不指定默认为 'UTC'
# CELERY_TIMEZONE='UTC'
# CELERY_ENABLE_UTC = True

加载模块文件

# Celery.py
from __future__ import absolute_import
from celery import Celery
# 指定名称
app = Celery('my_celery')
# 加载配置模块
app.config_from_object('config')

if __name__ == '__main__':
      app.start()

一份比较常用的配置文件

BROKER_URL = 'amqp://username:passwd@host:port/虚拟主机名'
# 指定结果的接受地址
CELERY_RESULT_BACKEND = 'redis://username:passwd@host:port/db'
# 指定任务序列化方式
CELERY_TASK_SERIALIZER = 'msgpack' 
# 指定结果序列化方式
CELERY_RESULT_SERIALIZER = 'msgpack'
# 任务过期时间,celery任务执行结果的超时时间
CELERY_TASK_RESULT_EXPIRES = 60 * 20   
# 指定任务接受的序列化类型.
CELERY_ACCEPT_CONTENT = ["msgpack"]   
# 任务发送完成是否需要确认，这一项对性能有一点影响     
CELERY_ACKS_LATE = True  
# 压缩方案选择，可以是zlib, bzip2，默认是发送没有压缩的数据
CELERY_MESSAGE_COMPRESSION = 'zlib' 
# 规定完成任务的时间
CELERYD_TASK_TIME_LIMIT = 5  # 在5s内完成任务，否则执行该任务的worker将被杀死，任务移交给父进程
# celery worker的并发数，默认是服务器的内核数目,也是命令行-c参数指定的数目
CELERYD_CONCURRENCY = 4 
# celery worker 每次去rabbitmq预取任务的数量
CELERYD_PREFETCH_MULTIPLIER = 4 
# 每个worker执行了多少任务就会死掉，默认是无限的
CELERYD_MAX_TASKS_PER_CHILD = 40 
# 设置默认的队列名称，如果一个消息不符合其他的队列就会放在默认队列里面，如果什么都不设置的话，数据都会发送到默认的队列中
CELERY_DEFAULT_QUEUE = "default" 
# 设置详细的队列
CELERY_QUEUES = {
    "default": { # 这是上面指定的默认队列
        "exchange": "default",
        "exchange_type": "direct",
        "routing_key": "default"
    },
    "topicqueue": { # 这是一个topic队列 凡是topictest开头的routing key都会被放到这个队列
        "routing_key": "topic.#",
        "exchange": "topic_exchange",
        "exchange_type": "topic",
    },
    "task_eeg": { # 设置扇形交换机
        "exchange": "tasks",
        "exchange_type": "fanout",
        "binding_key": "tasks",
    },
}

image

创建异步任务的方法task

任何被 task 修饰的方法都会被创建一个 Task 对象

# task方法参数
name:可以显式指定任务的名字；默认是模块的命名空间中本函数的名字。
serializer：指定本任务的序列化的方法；
bind:一个bool值，设置是否绑定一个task的实例，如果绑定，task实例会作为参数传递到任务方法中，可以访问task实例的所有的属性，即前面反序列化中那些属性
base:定义任务的基类，可以以此来定义回调函数，默认是Task类，我们也可以定义自己的Task类
default_retry_delay：设置该任务重试的延迟时间，当任务执行失败后，会自动重试，单位是秒，默认3分钟；
autoretry_for：设置在特定异常时重试任务，默认False即不重试；
retry_backoff：默认False，设置重试时的延迟时间间隔策略；
retry_backoff_max：设置最大延迟重试时间，默认10分钟，如果失败则不再重试；
retry_jitter：默认True，即引入抖动，避免重试任务集中执行

还记得上一篇 Celery部署爬虫(一) 中调用函数的方法吗？

def delay(url):
    result = get_url.delay(url)
    return result

其实 delay 是 apply_saync 的一种简易的写法。但是 apply_saync 提供更多的参数。

一份比较Task.apply_async的参数集合
apply_async支持执行选项，它会覆盖全局的默认参数和定义该任务时指定的执行选项，本质上还是调用了send_task方法；

add.apply_async(args=[30,40], kwargs={'z':5})

# 其他参数
task_id:为任务分配唯一id，默认是uuid;
countdown : 设置该任务等待一段时间再执行，单位为s；
eta : 定义任务的开始时间；eta=time.time()+10;
expires : 设置任务时间，任务在过期时间后还没有执行则被丢弃；
retry : 如果任务失败后, 是否重试;使用true或false，默认为true
shadow：重新指定任务的名字str，覆盖其在日志中使用的任务名称；
retry_policy : {},重试策略.如下：
    max_retries : 最大重试次数, 默认为 3 次.
    interval_start : 重试等待的时间间隔秒数, 默认为 0 , 表示直接重试不等待.
    interval_step : 每次重试让重试间隔增加的秒数, 可以是数字或浮点数, 默认为 0.2
    interval_max : 重试间隔最大的秒数, 即 通过 interval_step 增大到多少秒之后, 就不在增加了, 可以是数字或者浮点数, 默认为 0.2 .

routing_key:自定义路由键；
queue：指定发送到哪个队列；
exchange：指定发送到哪个交换机；
priority：任务队列的优先级，0到255之间，对于rabbitmq来说0是最高优先级；
serializer：任务序列化方法；通常不设置；
compression：压缩方案，通常有zlib, bzip2
headers：为任务添加额外的消息；
link：任务成功执行后的回调方法；是一个signature对象；可以用作关联任务；
link_error: 任务失败后的回调方法，是一个signature对象；

# 实例如下
add.apply_async((2, 2), retry=True, retry_policy={
    'max_retries': 3,
    'interval_start': 0,
    'interval_step': 0.2,
    'interval_max': 0.2,
})

image

获取任务结果和状态

由于celery发送的都是去其他进程执行的任务，如果需要在客户端监控任务的状态，有如下方法：

r = task.apply_async()
r.ready()     # 查看任务状态，返回布尔值,  任务执行完成, 返回 True, 否则返回 False.
r.wait()      # 会阻塞等待任务完成, 返回任务执行结果，很少使用；
r.get(timeout=1)       # 获取任务执行结果，可以设置等待时间，如果超时但任务未完成返回None；
r.result      # 任务执行结果，未完成返回None；
r.state       # PENDING, START, SUCCESS，任务当前的状态
r.status      # PENDING, START, SUCCESS，任务当前的状态
r.successful  # 任务成功返回true
r.traceback  # 如果任务抛出了一个异常，可以获取原始的回溯信息

image

关于定时任务

在配置文件中加入 CELERYBEAT_SCHEDULE 选项可以轻松实现定时任务

只需在 task 处声明定时函数即可

CCELERYBEAT_SCHEDULE = {
    'add-every-30-seconds': {
         'task': 'tasks.max',
         'schedule': datetime.timedelta(seconds=30),# 每 30 秒执行一次
         'args': (5, 8)                             # 任务函数参数
    },
    'multiply-at-some-time': {
        'task': 'tasks.add',
        'schedule': crontab(hour=9, minute=50),   # 每天早上 9 点 50 分执行一次
        'args': (3, 7)                            # 任务函数参数
    }
}

也可以根据实例来调度 需要引入 Celery 模块

from Celery import app
app.conf.beat_schedule = {
    'func2':{
        'task':'Task_func.func2',
        'schedule': 10.0,
        'args': ('wd',),
    },
    'func3':{
        'task':'Task_func.func3',
        'schedule': 10.0,
        'args': (1000000,3),
    },

}

定时任务是在 Celery 框架中调用起来显得辣么简单，这也是 Celery 能够胜任众多场景一个比较重要的地方吧！

Chain链式法则

在 Celery 的日常使用中，链式法则也是经常出没

看看一些比较高级的用法

# coding=utf-8
from celery import group,chain,chord
from period_tasks import add,mul,sum
# group
# res = group(add.s(34,2),add.s(1,2))()  # 任务 [1+2,1+2]
# while True:
#     if res.ready():
#         print('res:{}'.format(res.get()))
#         break


# chain
# 链式任务中，默认上一个任务的返回结果作为参数传递给子任务
# res = chain(add.s(1,2),add.s(3),mul.s(3))()  # 任务((1+2)+3)*3
# while True:
#     if res.ready():
#         print('res:{}'.format(res.get()))
#         break

'''
还可以使用|表示链式任务，上面任务也可以表示为：
res = (add.s(1,2) | add.s(3) | (mul.s(3)))()
res.get()
'''

# chord：任务分割，分为header和body两部分，hearder 任务执行完在执行body,其中 hearder 返回结果作为参数传递给body
# res = chord(header=[add.s(1,2),mul.s(3,4)],body=sum.s())()  # 任务(1+2)+(3*4)
# while True:
#     if res.ready():
#         print('res:{}'.format(res.get()))
#         break

res = add.chunks(zip(range(10),range(10)),4)()  # 4 代表每组的任务的个数
while True:
    if res.ready():
        print('res:{}'.format(res.get()))
        break

既然是链式法则，那用起来也就那个鸟样了，并没有啥稀奇的地方。

image

这大概就是一些基本的用法，配合一些爬虫框架比如 Scrapy 或者是自己自定义的脚本就可以开始任务了，使用诸如 celery -A haha worker -l info -P gevent -c 10 的命令即可开启celery服务，运行 python run_task.py 这样的命令来开始消费

image

然而，下一篇就不是如此的小打小闹了
将一举拿下Celery

欢迎转载，但要声明出处，不然我顺着网线过去就是一拳。
个人技术博客：http://www.gzky.live