關於Flask使用Celery的實踐經驗分享

　　  最近大Boss反饋Celery經常出現問題，幾經實踐終於把問題解決了！於是乎有了這篇博客的誕生，算是一個實踐經驗的分享吧！

　　 軟件版本如下：

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如下：

# install on Ubuntu apt-get update apt-get install rabbitmq-server -yq

需要注意的是，線上環境我們需要創建新的賬號，並將guest賬號刪除，操作如下：

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：

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

注意：當使用amqp協議頭時，如果安裝有librabbitmq則使用librabbitmq，否則使用pyamqp

Celery的日志輸出

在task中想要輸出日志，最好的方法是通過如下方式：

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來指定輸出文件，如下：

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設置如下：

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

查看Celery任務的結果可以通過Redis-cli連接Redis數據庫進行查看：

> redis-cli > select 1 # 這里選擇DB 1， 也可以在使用redis-cli -n 1來進入指定的DB > get key # 獲取指定key對應的結果

Celery可能會遇到的坑

Celery4.x版本使用librabbitmq的問題

Celery 4.x版本在使用librabbitmq時，會出現類似這樣的錯誤

Received and deleted unknown message.  Wrong destination?!?

解決這個問題有兩個方式：

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.

2. 另一種方式是不使用librabbitmq, 通過pip uninstall librabbitmq, 並且更改broker配置的協議頭為'pyamqp',如下，也可以解決這個問題。

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

NODE_IP_ADDRESS=0.0.0.0

Celery import問題

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

celery -A app.celery worker -l info 

要解決這個問題，必須為celery配置文件添加import參數，如下

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

Celery unregistered task問題

在開發過程中遇到了這樣一個問題：

[2018-03-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的日志，結果發現出現如下錯誤：

[2018-03-31 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中並沒有原先設置的用戶（我使用的是原系統的鏡像，原以為用戶也是已經設置好的）

# 查看有哪些用戶 rabbitmqctl list_users 

然后就簡單了，按照步驟創建用戶，vhost，再賦予權限，刪除guest，然后就終於都連好了

另外，發現從鏡像復制系統后，RabbitMQ並不能正常工作，必須殺掉原先的進程，重新啟動

更改task的代碼后，重啟Celery

需要注意的是，在更改task的代碼后，必須重新啟動Celery，否則代碼改動無法生效，可能導致一些意外的問題

以上就是筆者使用過程中的一些坑，經驗有限，如果有錯漏還請指正！不要看格式啦，因為之間都是習慣寫印象筆記，所以你們就將就着看吧！