multiprocessing.Pool无限循环(并行处理不停止)

Question

这是我的用户定义的函数

yms = ['201901', '201902', '201903', '201904', '201905', .... , '201812']

def minc(ym):
    print('MINC %s %s\n' %(ym, str(datetime.datetime.now())))
    print('value %s is in PID : %s \n' %(ym, os.getpid()))
    t = datetime.datetime.now()
    minc1 = pd.read_sql("""
    select substring(MINC_IN_YM, 1, 4) as YEAR, substring(MINC_IN_YM, 5, 2) as MONTH,
    count(MINC_INSP_NO) as NROWS,
    sum(MINC_OKQTY) as TOTAL_QUANTITY,
    sum(MINC_AV_PRICE*MINC_OKQTY) as TOTAL_DOLLARS
    from dwadm.W_MINC
    where MINC_INC_INF in ('RN', 'CN')
    and MINC_ACCID in ('A', 'G', 'V')
    and MINC_IN_YM = '%s'
    and substring(MINC_BRNCD, 1, 1) not in ('S', 'C')
    GROUP BY YEAR, MONTH, MINC_BRNCD, MINC_BRNCD_WHS, MINC_VNDCD, MINC_PTNO
    """ % ym, conn)
    print('MINC ends %s1 %s\n' %(ym, str(datetime.datetime.now()), str(datetime.datetime.nows())
    
    return minc1

我想用函数'minc()‘进行多进程处理，下面的函数是'parallelized()’。

def parallelized():
    if __name__ == '__main__' :
      pool = Pool(processes = 8)
      df = pool.map(minc, yms)
      pool.close()
      pool.join()

parallelized()

或

if __name__ == '__main__':
    freeze_support()
    t1 = time.time()
    pool = Pool(8)
    pool.map(minc, yms)
    pool.close()
    pool.join()

但是当我运行这些代码时，这些代码不会停止。他们在无止境地奔跑..。我不知道如何解决这个问题。当我使用用户定义的函数'minc()‘而使用'sum()’或任何其他基本函数时，我可以意识到并行处理执行得很好。是否有其他解决方案可以很好地执行用户定义函数的并行处理？

请救救我！

--它是执行并行化()函数时的输出.

MINC 201902 2020-11-03 08:15:23.469221

MINC 201901 2020-11-03 08:15:23.469228

MINC 201903 2020-11-03 08:15:23.469482

MINC 201904 2020-11-03 08:15:23.469703

MINC 201905 2020-11-03 08:15:23.469915

MINC 201906 2020-11-03 08:15:23.470106

MINC 201907 2020-11-03 08:15:23.470283

值201902为PID : 1222。

MINC 201908 2020-11-03 08:15:23.470459

值201901为PID : 1221。

值201903为PID : 1223。

值201904为PID : 1224。

值201905为PID : 1225。

值201906为PID : 1226。

值201907为PID : 1227。

值201908为PID : 1228。

MINC 201909 2020-11-03 08:15:23.489395

值201909为PID : 1222。

Answer 1

这是代码的简化版本，您使用multiprocessing.Pool的方式似乎是正确的。如果子进程中的代码出现故障，程序就会挂起。您确定您的pd.read_sql代码状态良好吗？

import datetime
from multiprocessing.pool import Pool
from os import getpid

yms = range(1, 10000)


def minc(ym):
    print('MINC %s %s %s\n' % (ym, getpid(), str(datetime.datetime.now())))

    return ym


def parallelized():
    pool = Pool(processes = 8)
    df = pool.map(minc, yms)
    print("FINISHED")
    assert list(yms) == df

    # you don't really have to close/join but it is still a good idea
    # https://stackoverflow.com/a/38271957/598057
    pool.close()
    pool.join()

parallelized()