使用Celery Chunks遇TypeError：square_batch()参数需可迭代对象而非int

错误原因 #

你当前的代码中，square_batch.chunks(data,10)会把data（range(100)）里的每个整数单独作为参数传给square_batch任务，但你的任务定义里期望接收一个可迭代对象（比如列表）来遍历计算平方，因此触发TypeError: worker.square_batch() argument after * must be an iterable, not int。

解决方案 #

要实现“拆分100条数据为10组，每组返回一个平方数列表”的预期，需要调整任务调用逻辑，确保每个任务接收的是一组数据而非单个整数：

修改调用代码 #

先将原始数据拆分为指定大小的批量，再传递给chunks：

from itertools import islice

@app.get("/chunk")
def cnk():
    data = range(100)
    
    # 定义批量拆分工具函数
    def chunked_iterable(iterable, size):
        iterator = iter(iterable)
        while True:
            chunk = list(islice(iterator, size))
            if not chunk:
                break
            yield (chunk,)  # 每个批量包装为任务参数元组
    
    # 基于拆分后的批量创建chunks任务
    job = square_batch.chunks(chunked_iterable(data, 10), 10)
    task = job.apply_async()
    return {"task submitted with id": task.id}

任务定义保持不变 #

你的任务定义无需修改，原本的逻辑就是接收一组数据并返回平方列表：

@celery_app.task(base=DefaultTask)
def square_batch(k):
    return [x*x for x in k]

效果验证 #

修改后，Celery会启动10个任务，每个任务接收10个整数的列表，处理后返回对应的平方数列表，最终会得到10个符合预期的结果集合。

内容的提问来源于stack exchange，提问作者Yuvraj Singh Rathore