将字符串中的字节发送到 Redis，以便将其用作 Celery 任务的参数

Question

我想使用 Redis 作为代理将字符串中的字节发送到 Celery 的任务，但我收到如下所示的错误：

[2017-06-17 21:27:13,826] ERROR in app: Exception on /endpoint_method [POST]
Traceback (most recent call last):
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/flask/app.py", line 1982, in wsgi_app
    response = self.full_dispatch_request()
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/flask/app.py", line 1614, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/flask/app.py", line 1517, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/flask/app.py", line 1612, in full_dispatch_request
    rv = self.dispatch_request()
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/flask/app.py", line 1598, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/Users/developer/my_project/application.py", line 23, in endpoint_method
    task = my_task.execute.delay(request.data)
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/celery/app/task.py", line 412, in delay
    return self.apply_async(args, kwargs)
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/celery/app/task.py", line 535, in apply_async
    **options
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/celery/app/base.py", line 737, in send_task
    amqp.send_task_message(P, name, message, **options)
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/celery/app/amqp.py", line 558, in send_task_message
    **properties
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/kombu/messaging.py", line 169, in publish
    compression, headers)
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/kombu/messaging.py", line 252, in _prepare
    body) = dumps(body, serializer=serializer)
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/kombu/serialization.py", line 221, in dumps
    payload = encoder(data)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/contextlib.py", line 35, in __exit__
    self.gen.throw(type, value, traceback)
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/kombu/serialization.py", line 54, in _reraise_errors
    reraise(wrapper, wrapper(exc), sys.exc_info()[2])
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/kombu/serialization.py", line 50, in _reraise_errors
    yield
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/kombu/serialization.py", line 221, in dumps
    payload = encoder(data)
  File "/Users/developer/my_project/venv/lib/python2.7/site-packages/kombu/utils/json.py", line 72, in dumps
    **dict(default_kwargs, **kwargs))
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/json/__init__.py", line 250, in dumps
    sort_keys=sort_keys, **kw).encode(obj)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/json/encoder.py", line 207, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/json/encoder.py", line 270, in iterencode
    return _iterencode(o, 0)
EncodeError: 'utf8' codec can't decode byte 0x80 in position 5: invalid start byte

我要发送到的缩短的（主要是最多 100 万个字符串）示例数据：

'RIFF$\x80\r\x00WAVEfmt \x10\x00\x00\x00\x01\x00\x02\x00D\xac\x00\x00\x10\xb1\x02\x00\x04\x00\x10\x00data\x00\x80\r\x00z\xefz\xef\xd5\xec\xd5\xec\xc1\xee\xc1\xee\xe6\xf3\xe6\xf3\xb4\xfa\xb4\xfa\x92\x03\x92\x03\xab\x0e\xab\x0e\xf9\x18\xf9\x18D\x1eD\x1ev\x1dv\x1d@\x19@\x19\x86\x13\x86\x13w\nw\n\xf0\xfd\xf0\xfd\xf9\xf2\xf9\xf2\xd0\xed\xd0\xedK\xedK\xed{\xed{\xed%\xee%\xeeP\xf3P\xf3\xeb\xfc\xeb\xfc!\x05!\x05\xa4\x08\xa4\x08\xe5\t\xe5\t\x84\x0b\x84\x0bF\x0bF\x0b\xfb\x04\xfb\x040\xfa0\xfa\x86\xf1\x86\xf1T\xeeT'

负责调用Celery任务的部分代码：

task = my_task.execute.delay(request.data)

这是 tasks/my_task.py 文件的样子：

from __future__ import absolute_import, unicode_literals
from my_module.celery import app

@app.task
def execute(request_data):
    return <some operations on request_data>

和celery.py:

from __future__ import absolute_import, unicode_literals
from celery import Celery

app = Celery(
    'my_module',
    broker='redis://127.0.0.1:6379/1',
    backend='redis://127.0.0.1:6379/2',
    include=[
        'my_module.tasks.my_task'
    ]
)

app.conf.update(
    result_expires=3600,
)

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

也许有一些聪明的方法可以在将字符串发送到 Redis 之前对该字符串进行编码，并在运行任务中对其进行解码？

Answer 1

最后，我想出了一个解决方案，使用 binascii 方法 binascii.b2a_base64(data) 和 binascii.a2b_base64(string)。这是我必须更改才能使其正常工作的内容：

task = my_task.execute.delay(binascii.b2a_base64(request.data))

在 tasks/my_task.py 中，我必须将数据转换回来才能对其进行正确的操作：

binascii.a2b_base64(request_data)

将字符串中的字节发送到 Redis，以便将其用作 Celery 任务的参数

Sending bytes in a string to Redis in order to using it as parameter for Celery's task

python

redis

celery

celery-task

binascii