将字符串中的字节发送到 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

缩短(主要是最多 1 百万个字符的字符串(我想发送到的示例数据：

'RIFF$x80rx00WAVEfmt x10x00x00x00x01x00x02x00Dxacx00x00x10xb1x02x00x04x00x10x00datax00x80rx00zxefzxefxd5xecxd5xecxc1xeexc1xeexe6xf3xe6xf3xb4xfaxb4xfax92x03x92x03xabx0exabx0exf9x18xf9x18Dx1eDx1evx1dvx1d@x19@x19x86x13x86x13wnwnxf0xfdxf0xfdxf9xf2xf9xf2xd0xedxd0xedKxedKxed{xed{xed%xee%xeePxf3Pxf3xebxfcxebxfc!x05!x05xa4x08xa4x08xe5txe5tx84x0bx84x0bFx0bFx0bxfbx04xfbx040xfa0xfax86xf1x86xf1TxeeT'

负责调用 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)