Django RQ出现pickle data was truncated错误的排查与解决求助

我来帮你排查这个问题，你现在遇到的_pickle.UnpicklingError: pickle data was truncated错误，结合你的代码和日志来看，主要有两个核心原因，咱们一步步拆解：

错误原因分析 #

1. 不可靠的等待逻辑
   你在视图里用time.sleep(0.2)来等待异步任务完成，这完全是碰运气——异步任务的执行时间受数据库查询速度、worker负载影响，0.2秒可能根本不够任务完成。当你提前去取a.result时，RQ还没把完整的pickle结果写入存储，自然会出现“数据截断”的错误。

2. QuerySet的懒加载特性不适合直接序列化
   你在任务里直接返回了ProductQuerySet，但QuerySet本身是懒加载的查询对象，不是实际的数据集合。当RQ对它做pickle序列化时，保存的只是查询的结构（比如过滤条件），而不是查询出来的产品数据。等到视图里反序列化时，不仅需要重新连接数据库执行查询（失去了异步的意义），还可能因为上下文丢失（比如worker和web进程的数据库连接隔离）导致反序列化失败。

解决方案 #

咱们从两个关键点入手修改：一是把QuerySet转换成可安全序列化的实际数据，二是用合理的方式处理异步任务的结果，而不是靠sleep赌运气。

第一步：修改异步任务，返回序列化后的数据 #

把任务里的QuerySet转换成字典列表或者JSON字符串，确保RQ存储的是实际的产品数据，而不是懒加载对象：

# tasks.py
from django.core import serializers
from django_rq import job

@job("high", timeout=360)
def long_runnig_task(s):
    # 先执行查询，触发懒加载获取实际数据
    products = Product.objects.filter(stockrecords__partner__id=s)
    # 用Django的序列化器转换成Python字典列表，方便后续反序列化
    return serializers.serialize('python', products)

第二步：修改视图逻辑，用异步方式获取结果 #

因为ListView是同步视图，不能直接阻塞等待异步任务完成，咱们改用“前端轮询任务状态”的方案，先返回加载页面，等任务完成后再刷新显示结果：

# views.py
from django.views.generic import ListView
from django_rq import get_queue, get_job
from .models import Product

class PartnerCatalogueView(ListView):
    """ Browse all products in the catalogue based on vendor """
    template_name = 'catalogue/vendorku.html'
    model = Product
    context_object_name = 'prod'

    def get_queryset(self, **kwargs):
        u = self.kwargs['pk']
        job_key = f'product_job_{u}'
        # 从session获取之前创建的任务ID
        existing_job_id = self.request.session.get(job_key)
        
        if existing_job_id:
            job = get_job(existing_job_id)
            if job and job.is_finished:
                # 任务完成，把序列化的数据转换成Product实例（或者直接用字典）
                serialized_prods = job.result
                products = [Product(**item['fields']) for item in serialized_prods]
                # 清理session里的任务ID
                del self.request.session[job_key]
                return products
            elif job and job.is_failed:
                # 处理任务失败的情况，返回空列表或错误提示
                return Product.objects.none()
        
        # 没有已存在的有效任务，创建新任务
        queue = get_queue('high', autocommit=True, async=True, default_timeout=360)
        new_job = queue.enqueue(long_runnig_task, u)
        self.request.session[job_key] = new_job.id
        
        # 返回空列表，让模板显示加载状态
        return Product.objects.none()

# 新增一个检查任务状态的视图，供前端轮询
from django.http import JsonResponse

def check_product_job_status(request, pk):
    job_key = f'product_job_{pk}'
    job_id = request.session.get(job_key)
    if not job_id:
        return JsonResponse({'finished': False, 'failed': True})
    
    job = get_job(job_id)
    if not job:
        return JsonResponse({'finished': False, 'failed': True})
    
    return JsonResponse({
        'finished': job.is_finished,
        'failed': job.is_failed
    })

第三步：更新模板，添加前端轮询逻辑 #

在vendorku.html里添加加载提示和AJAX轮询代码，自动检测任务完成后刷新页面：

{% extends "base.html" %}

{% block content %}
    {% if prod %}
        <!-- 正常显示产品列表 -->
        <div class="product-list">
            {% for product in prod %}
                <div class="product-item">
                    <h3>{{ product.name }}</h3>
                    <!-- 其他产品信息 -->
                </div>
            {% endfor %}
        </div>
    {% else %}
        <!-- 加载状态提示 -->
        <div id="loading-container" class="text-center">
            <p>正在加载产品，请稍候...</p>
        </div>

        <script>
            // 轮询任务状态的函数
            function checkJob() {
                const pk = "{{ kwargs.pk }}";
                fetch(`{% url 'check_product_job_status' pk=pk %}`)
                    .then(response => response.json())
                    .then(data => {
                        if (data.finished) {
                            // 任务完成，刷新页面显示结果
                            window.location.reload();
                        } else if (data.failed) {
                            document.getElementById('loading-container').innerHTML = '<p>加载失败，请刷新页面重试</p>';
                        } else {
                            // 未完成，1秒后再次查询
                            setTimeout(checkJob, 1000);
                        }
                    })
                    .catch(error => {
                        console.error('检查任务状态失败:', error);
                        setTimeout(checkJob, 2000);
                    });
            }

            // 页面加载后开始轮询
            setTimeout(checkJob, 1000);
        </script>
    {% endif %}
{% endblock %}

第四步：添加路由 #

在你的应用urls.py里新增检查任务状态的路由：

from django.urls import path
from .views import PartnerCatalogueView, check_product_job_status

urlpatterns = [
    path('partner-catalogue/<int:pk>/', PartnerCatalogueView.as_view(), name='partner_catalogue'),
    path('partner-catalogue/<int:pk>/check-job/', check_product_job_status, name='check_product_job_status'),
]

验证修改后的效果 #

1. 启动Django RQ worker：python manage.py rqworker high
2. 访问PartnerCatalogueView页面，会先显示加载提示
3. 等worker完成任务后，页面会自动刷新，显示产品列表
4. 查看worker日志，会看到任务返回的是序列化后的字典列表，而不是QuerySet对象

这样修改后，既解决了pickle截断的问题，又合理利用了异步任务的优势，避免了同步视图阻塞的问题。

内容的提问来源于stack exchange，提问作者takeiteasy