celery+Rabbit MQ实战记录

悠悠 2022-04-24 09:16 260阅读 0赞

基于以前的一篇文章，[celery+Rabbit MQ的安装和使用][celery_Rabbit MQ]，

本文更加详细的介绍如何安装和使用celey, Rabbit MQ。  
并记录在使用celery时遇到的一些问题。

# 1.安装 Rabbit MQ #

在 OS X上，直接执行如下命令：

$ brew install rabbitmq

其他操作系统下的安装可以参考[安装 RabbitMQ][RabbitMQ]

**启动RabbitMQ**

$ sudo rabbitmq-server

你也可以添加 -detached 属性来让它在后台运行（注意：只有一个破折号）：

$ sudo rabbitmq-server -detached

**查看RabbitMQ状态**

$ sudo rabbitmqctl status

**停止RabbitMQ**

永远不要用 kill 停止 RabbitMQ 服务器，而是应该用 rabbitmqctl 命令：

$ sudo rabbitmqctl stop

**添加用户**

默认用户guest,密码guest，只允许本地访问，如需远程访问，需要设置.

$ sudo rabbitmqctl add_user test 123456
    Adding user "test" ...

添加虚拟主机，并赋予用户test权限

$ sudo rabbitmqctl add_vhost myvhost
    Adding vhost "myvhost" ...

$ sudo rabbitmqctl set_permissions -p myvhost test ".*" ".*" ".*"
    Setting permissions for user "test" in vhost "myvhost" ...

# 2.安装celery #

## 2.1 创建虚拟环境,并安装celery ##

$ mkdir celery_demo
    $ cd celery_demo
    $ virtualenv -p python3 venv3

$ ./venv3/bin/pip install celery

## 2.2 配置celery ##

创建配置文件 [celeryconfig.py][]，里面包含`BROKER_URL`、`CELERYD_LOG_FORMAT`、`CELERY_ROUTES`.

# celeryconfig.py
    
    RABBIT_MQ = {
        'HOST': '127.0.0.1',
        'PORT': 5672,
        'USER': 'test',
        'PASSWORD': '123456'
    }
    
    # broker
    BROKER_URL = 'amqp://%s:%s@%s:%s/myvhost' % (RABBIT_MQ['USER'], RABBIT_MQ['PASSWORD'], RABBIT_MQ['HOST'], RABBIT_MQ['PORT'])
    
    # celery日志格式
    CELERYD_LOG_FORMAT = '[%(asctime)s] [%(levelname)s] %(message)s'
    
    CELERY_ROUTES = {
            'demo_task.add': {'queue': 'sunday'},
    }

其中，参数定义如下：

*  `BROKER_URL`指定了broker信息，即消息队列的地址。
 *  `CELERYD_LOG_FORMAT` 指定了日志格式。
 *  `CELERY_ROUTES` 指定了路由信息，即调用`demo_task.add`后，消息具体放入哪个队列，这里是队列名称为`sunday`。

## 2.3 启动消费者 ##

**消费者**

消费者代码如下：

# demo_task.py
    from celery import Celery
    
    
    app = Celery("orange", backend='amqp')
    app.config_from_object("celeryconfig")
    
    @app.task
    def add(x, y):
            return x + y

首先，创建`Celery`实例，从文件中读取配置。  
接着，定义task。

其中，在创建`Celery`实例时，参数`backend`指定了结果存储后端，用于追踪task执行状态和结果。这里使用`amqp`，即使用RabbitMQ保存结果。默认情况下，`backend`参数是关闭的。

**使用`celery worker`启动消费者**

./venv3/bin/celery worker  -A demo_task -Q sunday --loglevel=info  -f app.log
    
    
    celery@admindeMacBook-Pro-2.local v4.3.0 (rhubarb)
    
    Darwin-18.2.0-x86_64-i386-64bit 2019-04-20 17:49:40
    
    [config]
    .> app:         orange:0x109ca0400
    .> transport:   amqp://test:**@127.0.0.1:5672/myvhost
    .> results:     disabled://
    .> concurrency: 8 (prefork)
    .> task events: OFF (enable -E to monitor tasks in this worker)
    
    [queues]
    .> sunday           exchange=sunday(direct) key=sunday
    
    
    [tasks]
      . demo_task.add

其中，参数定义如下：

*  参数`-A` 是app name,即定义celery的文件。
 *  参数`-Q`指定了队列的名称，如果不指定，默认为`celery`。
 *  参数`-f` 指定了日志打印文件。

可以通过以下命令查看更多帮助信息：

*  `celery help`查看celery的选项
 *  `celery worker --help`查看worker的选项

## 2.4 启动生产者 ##

**生产者**

代码如下api.py所示：

# api.py
    from demo_task import add
    
    
    print("start...")
    
    result = add.apply_async((1, 2))
    
    print("result:", result)
    print(result.ready())
    
    print("end...")

代码中，将`1`、`2`参数放入消息队列。

celery配置`backend`参数后, 调用任务时，会返回 `AsyncResult` 实例。  
`AsyncResult` 的`ready()` 方法可以查看任务是否完成处理。

**[执行api.py][api.py]**

执行结果分两种情况：worker启动和没有启动。

当celery worker启动的时候，结果如下：

./venv3/bin/python api.py
    start...
    result: 60dd0ab6-5fa2-4190-954e-584eb519384f
    True
    end...

可以看到，task很快得到执行，结果状态为True。

当celery worker没有启动的时候，结果如下：

./venv3/bin/python api.py
    start...
    result: 7280466f-73cd-44ec-85e7-5ad4f079a797
    False
    end...

可以看到，结果状态为False。

**直接获取结果**

可以使用`get()`函数等待任务完成，但这很少使用，因为它把异步调用变成了同步调用。

对于`get()`的使用，按照是否设置超时参数，分为两种：不使用超时参数timeout和使用超时参数timeout。

(1)不使用超时参数timeout

默认timeout为`None`,不会超时，而是阻塞住。

通过tcpdump抓包，可以发现，`get`调用后，每隔30s，RabbitMQ(server端)向生产者API(client端)发送心跳包，共发了5次。

最后，过了3分钟，RabbitMQ(server端)关闭连接，生产者API(client端)也关闭连接，报错：

start...
    result: aae4fb20-823c-4674-8bff-28429f14d5d7
    False
    Traceback (most recent call last):
      File "api.py", line 13, in <module>
        result.get()
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/result.py", line 226, in get
        on_message=on_message,
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/base.py", line 496, in wait_for_pending
        no_ack=no_ack,
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 146, in wait_for
        on_interval=on_interval)
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 223, in consume
        conn, consumer, timeout, on_interval)[task_id]
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 204, in drain_events
        wait(timeout=1)
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/kombu/connection.py", line 315, in drain_events
        return self.transport.drain_events(self.connection, **kwargs)
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/kombu/transport/pyamqp.py", line 103, in drain_events
        return connection.drain_events(**kwargs)
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/amqp/connection.py", line 500, in drain_events
        while not self.blocking_read(timeout):
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/amqp/connection.py", line 505, in blocking_read
        frame = self.transport.read_frame()
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/amqp/transport.py", line 256, in read_frame
        frame_header = read(7, True)
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/amqp/transport.py", line 448, in _read
        raise IOError('Server unexpectedly closed connection')
    OSError: Server unexpectedly closed connection

(2)使用超时参数timeout

如果使用`timeout`参数,`get(timeout)`调用发起后， 当超过`timeout`指定的时间仍然没有获得结果，会超时报错。

例如，调用`get(timeout=1)`：

... ...
    
    # sync
    print(result.get(timeout=1))
    
    print("end...")

如果worker没有启动或者worker处理超时，`get`会报超时错误：

start...
    result: 190ad871-aae2-48ac-a18e-962cd2d537b7
    False
    Traceback (most recent call last):
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 146, in wait_for
        on_interval=on_interval)
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 223, in consume
        conn, consumer, timeout, on_interval)[task_id]
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 202, in drain_events
        raise socket.timeout()
    socket.timeout
    
    During handling of the above exception, another exception occurred:
    
    Traceback (most recent call last):
      File "api.py", line 14, in <module>
        print(result.get(timeout=1))
      File "/workspace//celery_demo/venv3/lib/python3.6/site-packages/celery/result.py", line 226, in get
        on_message=on_message,
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/base.py", line 496, in wait_for_pending
        no_ack=no_ack,
      File "/workspace/celery_demo/venv3/lib/python3.6/site-packages/celery/backends/amqp.py", line 148, in wait_for
        raise TimeoutError('The operation timed out.')
    celery.exceptions.TimeoutError: The operation timed out.

## 2.5 查看结果 ##

查看app.log中的日志

[2019-04-20 20:53:59,327] [INFO] Connected to amqp://test:**@127.0.0.1:5672/myvhost
    [2019-04-20 20:53:59,345] [INFO] mingle: searching for neighbors
    [2019-04-20 20:54:00,379] [INFO] mingle: all alone
    [2019-04-20 20:54:00,441] [INFO] celery@admindeMacBook-Pro-2.local ready.
    [2019-04-20 20:55:46,498] [INFO] Received task: demo_task.add[1c8d47bd-449d-4bd9-b4db-819777081d23]
    [2019-04-20 20:55:46,561] [INFO] Task demo_task.add[1c8d47bd-449d-4bd9-b4db-819777081d23] succeeded in 0.060540573904290795s: 3

# 3.任务过期 #

调用`apply_async`时使用参数`expires`，则表示任务有超时时间，超过这个时间后，task不会得到执行。

result = add.apply_async((1, 2), expires=10)

当worker没有启动或者其他异常情况下，会出现任务超时，不被执行。

查看日志，可以看到task过期，不会被执行。

[2019-04-20 21:32:45,828] [INFO] Received task: demo_task.add[12691c27-3f2f-4c96-9b4e-54636e20d0eb]   expires:[2019-04-20 13:31:51.030013+00:00]
    [2019-04-20 21:32:45,829] [INFO] Discarding revoked task: demo_task.add[12691c27-3f2f-4c96-9b4e-54636e20d0eb]

# 3.参考 #

[Celery 初步][Celery]

[calling API][]

[Configuration and defaults][]

[expiration][]

[celery_Rabbit MQ]: https://blog.csdn.net/lanyang123456/article/details/77115868
[RabbitMQ]: http://docs.jinkan.org/docs/celery/getting-started/brokers/InstallingRabbitMQ
[celeryconfig.py]: http://celeryconfig.py
[api.py]: http://xn--api-m35fr68n.py
[Celery]: http://docs.jinkan.org/docs/celery/getting-started/first-steps-with-celery.html#first-steps
[calling API]: http://docs.jinkan.org/docs/celery/userguide/calling.html#guide-calling
[Configuration and defaults]: http://docs.jinkan.org/docs/celery/configuration.html#message-routing
[expiration]: http://docs.jinkan.org/docs/celery/userguide/calling.html#expiration