【发布时间】：2018-07-22 12:29:47
【问题描述】：
我正在尝试将对我们 API 的 POST 请求调用卸载到 Celery，因为我们很快将每秒向我们的 API 发送多达 10 个请求，每个请求将在我们的数据库中创建超过 100 个对象.我想我会将它们添加到队列中，然后让 Redis + Celery 处理，然后从那里开始工作。
不过我遇到了一些问题。
首先，我的芹菜设置：
########## CELERY
import djcelery
djcelery.setup_loader()
INSTALLED_APPS += ['expert.taskapp.celery.CeleryConfig']
CELERY_ACCEPT_CONTENT = ['json', 'pickle']
CELERY_TASK_SERIALIZER = 'pickle'
CELERY_RESULT_SERIALIZER = 'pickle'
BROKER_URL = 'redis://127.0.0.1:6379'
CELERY_BROKER_URL = env('CELERY_BROKER_URL', default='redis://127.0.0.1:6379')
if CELERY_BROKER_URL == 'django://':
CELERY_RESULT_BACKEND = 'redis://'
else:
CELERY_RESULT_BACKEND = CELERY_BROKER_URL
########## END CELERY
为我的 Django REST 框架使用类 basec 视图，这是我目前的视图：
from celery import shared_task
from celery.decorators import task
from .tasks import create
class DataCreateAPIView(CreateAPIView):
def create(self, request, *args, **kwargs):
create.delay(request)
所以我的想法是让创建视图处理所有事情，直到我进入过程的创建部分，我立即将创建任务卸载到 Celery。
在我的 tasks.py 文件中，我有这个：
from celery import shared_task
from celery.decorators import task
from expert.models import Chamber, Parameter, Sensor, Data
@task(name='POST request Data point.')
def create(self, request, *args, **kwargs):
queryset = Data.objects.all()
queryset = DataCreateSerializer.setup_eager_loading(queryset)
# serializer_class = DataCreateSerializer
try:
sensor = Sensor.objects.get(serial_number=request.data["data_source"])
request.data["data_source"] = sensor.id
except Sensor.DoesNotExist:
print("Sensor serial number " + str(request.data["data_source"]) + " not registered.")
return Response(status=status.HTTP_404_NOT_FOUND)
dataDict = dict(request.data)
for param in dataDict['parameters']:
Parameter.objects.get_or_create(parameter_name=param, parameter_position="None")
final_data = []
for data in dataDict['data_array']:
zipped = zip(dataDict['parameters'], data['values'])
for parameter, value in zipped:
# parameter = Parameter.objects.get_or_create(parameter_name=parameter, parameter_position="None")[0]
parameter = Parameter.objects.get(parameter_name=parameter)
final_data.append({
"sensor": sensor.id,
"parameter": parameter.id,
"time": data['time'],
"parameter_value": value
})
serializer = DataCreateSerializer(data=final_data, many=True)
if serializer.is_valid():
serializer.save()
return Response(status=status.HTTP_200_OK)
return Response(serializer.errors)
在我接受大请求的地方，稍微修改一下以适应我们的架构，然后编写数据库。
现在，如果我只是采用相同的“创建”函数并将其直接放在 CreateAPIView 上，这一切都可以正常工作。
当我尝试使用 Celery 执行此操作时，当我初始化 celery 工人时，我的任务列表中显示了该任务，但我无法获得到达 celery 的请求。在此之前我因以下错误而中断：
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/django/core/handlers/exception.py" in inner
42. response = get_response(request)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/django/core/handlers/base.py" in _get_response
187. response = self.process_exception_by_middleware(e, request)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/django/core/handlers/base.py" in _get_response
185. response = wrapped_callback(request, *callback_args, **callback_kwargs)
File "/usr/lib/python3.5/contextlib.py" in inner
30. return func(*args, **kwds)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/django/views/decorators/csrf.py" in wrapped_view
58. return view_func(*args, **kwargs)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/django/views/generic/base.py" in view
68. return self.dispatch(request, *args, **kwargs)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/rest_framework/views.py" in dispatch
489. response = self.handle_exception(exc)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/rest_framework/views.py" in handle_exception
449. self.raise_uncaught_exception(exc)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/rest_framework/views.py" in dispatch
486. response = handler(request, *args, **kwargs)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/rest_framework/generics.py" in post
192. return self.create(request, *args, **kwargs)
File "/home/luke/Projects/expert/impedans_expert/impedans_expert/expert/api/views.py" in create
249. create.delay(request)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/celery/app/task.py" in delay
461. return self.apply_async(args, kwargs)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/celery/app/task.py" in apply_async
573. **dict(self._get_exec_options(), **options)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/celery/app/base.py" in send_task
354. reply_to=reply_to or self.oid, **options
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/celery/app/amqp.py" in publish_task
310. **kwargs
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/kombu/messaging.py" in publish
165. compression, headers)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/kombu/messaging.py" in _prepare
241. body) = dumps(body, serializer=serializer)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/kombu/serialization.py" in dumps
164. payload = encoder(data)
File "/usr/lib/python3.5/contextlib.py" in __exit__
77. self.gen.throw(type, value, traceback)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/kombu/serialization.py" in _reraise_errors
59. reraise(wrapper, wrapper(exc), sys.exc_info()[2])
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/kombu/five.py" in reraise
131. raise value.with_traceback(tb)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/kombu/serialization.py" in _reraise_errors
55. yield
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/kombu/serialization.py" in dumps
164. payload = encoder(data)
File "/home/luke/.virtualenvs/expert/lib/python3.5/site-packages/kombu/serialization.py" in pickle_dumps
356. return dumper(obj, protocol=pickle_protocol)
Exception Type: EncodeError at /expert/api/data/create/
Exception Value: cannot serialize '_io.BufferedReader' object
老实说，我必须知道如何从那里开始。我试过用谷歌搜索异常，但即使我看到一页内容看起来很相似，我还是无法理解该线程。
我将非常感谢您在这件事上的任何帮助。
【问题讨论】：
-
你是如何解决这个问题的？` kombu.exceptions.EncodeError: Object of type Request is not JSON serializable` :D
标签： python django rest celery django-celery