在Flask中使用Celery的最佳实践

主要是笔者在项目实践中总结的关于Celery使用的一些最佳实践，欢迎补充

写在前面

本最佳实践是基于作者有限的经验，欢迎大家共同讨论，可以持续维护此最佳实践。另本文中所使用的环境为Mac&Ubuntu环境，软件版本如下：

  • Celery (4.1.0)
  • Flask (0.12.1)
  • RabbitMQ(3.6.9)
  • librabbitmq (1.6.1)

介绍

简单来说Celery是一个异步的任务队列，当我们需要将一些任务(比如一些需要长时间操作的任务)异步操作的时候，这时候Celery就可以帮到我们，另外Celery还支持定时任务(类似Crontab)。详细的介绍可以参考官网

使用RabbitMQ作为Broker

RabbitMQ是官方推荐使用的Broker，它实际是一个消息中间件，负责消息的路由分发，安装RabbitMQ如下： 

1
2
3
# install on Ubuntu
apt-get update
apt-get install rabbitmq-server -yq

需要注意的是，线上环境我们需要创建新的账号，并将guest账号删除，操作如下： 

1
2
3
4
rabbitmqctl add_user myuser mypassword  # 新增用户
rabbitmqctl add_vhost myvhost  # 新增vhost，以使用不同的命名空间
rabbitmqctl set_permissions -p myvhost myuser ".*" ".*" ".*"  # 设置权限
rabbitmqctl  delete_user guest  # 安全原因，删除guest

注意：vhost是一个虚拟空间，用于区分不同类型的消息 然后，在Celery的配置中配置broker URL 

1
CELERY_BROKER_URL = 'amqp://myuser:mypassword@localhost:5672/myvhost'

注意：当使用amqp协议头时，如果安装有librabbitmq则使用librabbitmq，否则使用pyamqp

Celery的日志输出

在task中想要输出日志，最好的方法是通过如下方式 

1
2
3
4
5
6
7
from celery.utils.log import get_task_logger

lg = get_task_logger(__name__)

@celery.task
def log_test():
    lg.debug("in log_test()")

但是仅如此会发现所有的日志最后都跑到shell窗口的stdout当中，原来必须得在启动celery的时候使用-f option来指定输出文件，如下： 

1
celery -A main.celery worker -l debug -f log/celery/celery_task.log &

-A：指定celery实例 worker: 启动worker进程 -l：指定log level，这里指定log level为debug level -f：指定输出的日志文件

使用Redis作为backend

当使用Redis作为存储后端的时候，我们可以通过设置DB number来使得Celery的结果存储与其它数据存储隔离开来，比如在笔者的项目中，redis还用作缓存的存储后端，因此为了区分，Celery在使用Redis的时候使用的DB number是1（默认是0），关于Redis DB number可以参考这里. 因此我们的backend设置如下： 

1
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1' # 最后的数字1代表DB number

查看Celery任务的结果可以通过Redis-cli连接Redis数据库进行查看 

1
2
3
> redis-cli
> select 1 # 这里选择DB 1， 也可以在使用redis-cli -n 1来进入指定的DB
> get key # 获取指定key对应的结果

redis-cli.jpg

调试代码

我认为此处是非常重要的一个技巧，即在调试代码的时候，我们可以将delay或者apply_async先去掉，直接调用worker的函数进行同步调试，调试成功后再加上delay或者apply_async method

Celery可能会遇到的坑

Celery4.x版本使用librabbitmq的问题

Celery 4.x版本在使用librabbitmq时，会出现类似这样的错误 

1
Received and deleted unknown message.  Wrong destination?!?

完整错误如图： celeryerror

解决这个问题有两个方式：

  1. 推荐方式，更改配置项task_protocol为1。 Github上Robert Kopaczewski详细解释了这个问题，原文如下：

Apparently librabbitmq issue is related to new default protocol in celery 4.x. You can switch to previous protocol version by either putting CELERY_TASK_PROTOCOL = 1 in your settings if you're using Django or settings app.conf.task_protocol = 1 in celeryconf.py.

  1. 另一种方式是不使用librabbitmq, 通过pip uninstall librabbitmq, 并且更改broker配置的协议头为'pyamqp',如下，也可以解决这个问题。 
    1
    BROKER_URL = 'pyamqp://guest:guest@localhost:5672/%2F'

由于librabbitmq的性能优势，我们还是推荐方式1来解决该问题。

RabbitMQ远程连接问题

如果RabbitMQ与Celery不在同一台机器上，除在Celery配置的时候要将BROKER_URL设置为正确的IP地址外，还需要将Rabbitmq的配置文件/usr/local/etc/rabbitmq/rabbitmq-env.conf中的NODE_IP_ADDRESS更改为0.0.0.0 

1
NODE_IP_ADDRESS=0.0.0.0

Celery import问题

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you're using relative imports?

Please see
http://docs.celeryq.org/en/latest/internals/protocol.html
for more information.

The full contents of the message body was:
'\x8e\xa7expires\xc0\xa3utc\xc3\xa4args\x91\x85\xa3tid\xb85971a43d47f84bb278f77fc2\xa3sen\xa2A1\xa2tt\xa2ar\xa2co\xc4\x00\xa1t\xa4like\xa5chord\xc0\xa9callbacks\xc0\xa8errbacks\xc0\xa7taskset\xc0\xa2id\xc4$c133dbf8-2c89-4311-b7cf-c377041058ec\xa7retries\x00\xa4task\xd9$tasks.messageTasks.send_like_message\xa5group\xc0\xa9timelimit\x92\xc0\xc0\xa3eta\xc0\xa6kwargs\x80' (239b)
Traceback (most recent call last):
  File "/Users/liufeng/.pyenv/versions/2.7.13/envs/kaopu_backend/lib/python2.7/site-packages/celery/worker/consumer/consumer.py", line 561, in on_task_received
    strategy = strategies[type_]
KeyError: u'tasks.messageTasks.send_like_message'

出现这条错误是由于我们的tasks跟celery并不是在同一个文件中，即不是同一个module，当我们通过如下命令启动task worker时，实际只加载了app module，而没有加载tasks相关的module 

1
2
celery -A app.celery worker -l info
`

要解决这个问题，必须为celery配置文件添加import参数，如下 

1
app.config['imports'] = ['tasks.messageTasks']

Celery unregistered task问题

在开发过程中遇到了这样一个问题 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
[2017-08-31 15:38:19,605: ERROR/MainProcess] Received unregistered task of type u'app.tasks.messageTasks.send_follow_message'.
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you're using relative imports?

Please see
http://docs.celeryq.org/en/latest/internals/protocol.html
for more information.

The full contents of the message body was:
'\x8e\xa7expires\xc0\xa3utc\xc3\xa4args\x91\x86\xa6sender\xa5Jenny\xa9target_id\xb859a5313847f84be534ad7d46\xabtarget_type\xa4user\xa7content\xc4\x00\xa8receiver\xb859a5313847f84be534ad7d46\xa4type\xa6follow\xa5chord\xc0\xa9callbacks\xc0\xa8errbacks\xc0\xa7taskset\xc0\xa2id\xc4$a4d40c14-1976-41a6-a753-d2a495929920\xa7retries\x00\xa4task\xd9*app.tasks.messageTasks.send_follow_message\xa5group\xc0\xa9timelimit\x92\xc0\xc0\xa3eta\xc0\xa6kwargs\x80' (312b)
Traceback (most recent call last):
  File "/Users/liufeng/.pyenv/versions/2.7.13/envs/kaopu_backend/lib/python2.7/site-packages/celery/worker/consumer/consumer.py", line 561, in on_task_received
    strategy = strategies[type_]
KeyError: u'app.tasks.messageTasks.send_follow_message'

解决这个问题，最开始是根据提示，将所有涉及到task的module全部加上from __future__ import absolute_import 之后运行之后还是不行，后来发现是由于之前启动时使用的是app module， 但是我的代码已经改成了main.py，所以重新启动了celery，最后问题解决

使用镜像迁移系统也依然需要重新添加rabbitmq的用户

问题最开始是发现无法点赞，也无法Follow用户，通过http消息发现出现502错误，于是登录到服务器检查，发现应用服务本身没有任何报错，于是又去查看Celery的日志，结果发现出现如下错误： 

1
[2017-11-13 16:32:01,243: ERROR/MainProcess] consumer: Cannot connect to amqp://celeryuser:**@loc      alhost:5672/celeryvhost: Couldn't log in: a socket error occurred.

经过一番搜索发现网上的评论主要是说URL不对的情况下会出现这种情况，但是我的URL没有改过啊，那又会是什么问题呢？继续看，发现有人提到了权限问题，于是又是一番检查，发现RabbitMQ中并没有原先设置的用户（我使用的是原系统的镜像，原以为用户也是已经设置好的） 

1
2
# 查看有哪些用户
rabbitmqctl  list_users

然后就简单了，按照步骤创建用户，vhost，再赋予权限，删除guest，然后就终于都连好了

另外，发现从镜像复制系统后，RabbitMQ并不能正常工作，必须杀掉原先的进程，重新启动

更改task的代码后，重启Celery

需要注意的是，在更改task的代码后，必须重新启动Celery，否则代码改动无法生效，可能导致一些意外的问题