使用链 EncodeError(RuntimeError('获取对象的 str 时超出最大递归深度)) 如何解决 python Celery 错误

Question

由于签名是动态生成的，因此您如何运行 for 循环中的链式任务。使用以下方法是因为将测试人员任务定义为：

@task
def tester(items):
    ch = []
    for i in items:
        ch.append(test.si(i))
    return chain(ch)()

如果链太大 os 或系统特定，

会引发 EncodeError(RuntimeError('maximum recursion depth exceeded while getting the str of an object',),) 错误。

E.g calling the task as follows

item = range(1,40000) #40,000 raises exception but #3,000 doesn't after setting sys.setrecursionlimit(15000)
tester.delay(item)

提出 EcodeError。过去，当项目的长度为 5000，即范围（1,5000）时，我曾经遇到过这个错误。我通过导入 sys 并在 module 的顶部调用 sys.setrecursionlimit(15000) 来解决这个问题。但这有一个限制，所以我决定稍微重构一下并使用下面的方法。这是在尝试拆分列表并在 chunks.The 之后分块进行问题是它似乎在 2000 年之后不再继续，即测试将 2000 打印到屏幕上。

@task
def test(i):
    print i


@task
def tester(items):
    ch = []
    for i in items:
        ch.append(test.si(i))
    counter = 1
    if len(ch) > 2000:
        ch_length = len(ch) #4k
        while ch_length >= 2000:
            do = ch[0:2000] # 2k
            print "Doing...NO#...{}".format(counter)
            ret = chain(do)() #doing 2k
            print "Ending...NO#...{}".format(counter)
            ch = ch[2000:] #take all left i.e 2k

            ch_length = len(ch) #2k
            if ch_length <= 2000 and ch_length > 0:
                print "DOING LAST {}".format(counter)
                ret = chain(ch)()
                print "ENDING LAST {}".format(counter)
                break
            else:
                break
            counter +=1
    else:
        ret = chain(ch)()
    return ret

根据 celery 文档，链基本上一个接一个地执行其中的任务。我希望 while 循环在继续之前仅在链中完成第一次迭代后继续。

我希望有人对此有经验并能提供帮助。

提前祝大家圣诞快乐！

Answer 1

您似乎遇到了这个问题：https://github.com/celery/celery/issues/1078

同样调用chain(ch)()好像是异步执行的。尝试在其上显式调用 apply()。

@app.task
def tester(items):
    ch = []
    for i in items:
        ch.append(test.si(i))

    PSIZE = 1000
    for cl in range(0, len(ch), PSIZE):
        print("cl: %s" % cl)
        chain(ch[cl:cl + PSIZE]).apply()
        print("cl: %s END" % cl)
    return None

使用链 EncodeError(RuntimeError('获取对象的 str 时超出最大递归深度)) 如何解决 python Celery 错误

How to solve python Celery error when using chain EncodeError(RuntimeError('maximum recursion depth exceeded while getting the str of an object))

python

celery