测试 Celery 工作的 Redis 命令

Question

我正在试验使用 Redis 作为代理的 Celery worker。

这是我对 Celery worker 的测试代码：

from celery import Celery
app = Celery('tasks', broker='redis://xxxxx.net:6379/0')

@app.task
def nextexec(payload):
    print(payload)

使用redis-cli，我在运行以下命令插入一个值到celery队列（由Celery自动创建）

RPUSH celery somekey 'somevalue'

但是我的 worker 在执行查询时系统地崩溃，我得到一个 Unrecoverable error: JSONDecodeError。它似乎收到 None 而不是应该解码的 JSON 字符串。

知道我需要执行什么 Redis 查询或者我需要对这个（但简单的）工作脚本进行哪些更改吗？

Answer 1

Celery 使用 serializers 在客户端和 worker 之间传输数据。每条消息都需要序列化，并且有一个 content_type header 描述用于对其进行编码的序列化方法。

这是使用 json.

序列化的示例消息

{'body': 'W1sxXSwge30sIHsiY2FsbGJhY2tzIjogbnVsbCwgImVycmJhY2tzIjogbnVsbCwgImNoYWluIjogbnVsbCwgImNob3JkIjogbnVsbH1d',
 'content-encoding': 'utf-8',
 'content-type': 'application/json',
 'headers': {'argsrepr': '(1,)',
  'eta': None,
  'expires': None,
  'group': None,
  'id': '5ce9a8d8-41d7-47a4-9074-beedabd88dcc',
  'kwargsrepr': '{}',
  'lang': 'py',
  'origin': 'gen5339@pavilion',
  'parent_id': None,
  'retries': 0,
  'root_id': '5ce9a8d8-41d7-47a4-9074-beedabd88dcc',
  'task': 't.wait',
  'timelimit': [None, None]},
 'properties': {'body_encoding': 'base64',
  'correlation_id': '5ce9a8d8-41d7-47a4-9074-beedabd88dcc',
  'delivery_info': {'exchange': '', 'routing_key': 'celery'},
  'delivery_mode': 2,
  'delivery_tag': '0177eb65-344e-4b1c-ab5f-8e2f5d75b8d3',
  'priority': 0,
  'reply_to': 'a5c611b8-18b3-3bbb-b598-c3757f06c4fd'}}

Celery worker 需要接收指定格式的消息。您不能将某些值推送给 broker(redis) 并期望 celery 执行它。

使用 python 来 queue 完成您的任务。

from mymodule import nextexec
payload ='some payload'
nextexec.delay(payload)

测试 Celery 工作的 Redis 命令

Redis command to test a Celery work

redis

celery

python-3.x