python 27 - 多处理时布尔检查失败

Question

我有一个脚本，它从 MySQL 表中检索活动“作业”列表，然后使用多处理库为每个活动作业实例化一次我的主脚本。我的多处理脚本具有检查给定作业是否已被另一个线程占用的功能。它通过检查 DB 表中的特定列是否为/不是 NULL 来做到这一点。 DB 查询返回单个项目元组：

def check_if_job_claimed():
    #...
    job_claimed = cursor.fetchone() #Returns (claim_id,) for claimed jobs, and (None,) for unclaimed jobs
    if job_claimed:
        print "This job has already been claimed by another thread."
        return
    else:
        do_stuff_to_claim_the_job

当我在没有多处理部分的情况下运行此功能时，声明检查工作得很好。但是当我尝试并行运行作业时，声明检查将所有 (None,) 元组读取为具有价值并因此具有真实性，因此该函数假定该工作已被声明。

我已尝试调整多处理器使用的并发进程数，但声明检查仍然不起作用...即使我将进程数设置为 1。我也尝试过使用 if 语句看看我能不能让它这样工作：

if job_claimed == True
if job_claimed == (None,)
# etc. 

不过运气不好。

是否有人知道多处理库中的某些内容会阻止我的声明检查函数正确解释 job_claimed 元组？也许我的代码有问题？

编辑

我在调试模式下对 job_claimed 变量运行了一些真实性测试。以下是这些测试的结果：

(pdb) job_claimed
    (None,)
(pdb) len(job_claimed)
    1
(pdb) job_claimed == True
    False
(pdb) job_claimed == False
    False
(pdb) job_claimed[0]
    None
(pdb) job_claimed[0] == True
    False
(pdb) job_claimed[0] == False
    False
(pdb) any(job_claimed)
    False
(pdb) all(job_claimed)
    False
(pdb) job_claimed is not True
    True
(pdb) job_claimed is not False
    True

编辑

根据要求：

with open('Resource_File.txt', 'r') as f:
    creds = eval(f.read())
connection = mysql.connector.connect(user=creds["mysql_user"],password=creds["mysql_pw"],host=creds["mysql_host"],database=creds["mysql_db"],use_pure=False,buffered=True)

def check_if_job_claimed(job_id):
    cursor = connection.cursor()
    thread_id_query = "SELECT Thread_Id FROM jobs WHERE Job_ID=\'{}\';".format(job_id) 
    cursor.execute(thread_id_query)
    job_claimed = cursor.fetchone()
    job_claimed = job_claimed[0]
    if job_claimed:
        print "This job has already been claimed by another thread. Moving on to next job..."
        cursor.close()
        return False
    else:
        thread_id = socket.gethostname()+':'+str(random.randint(0,1000))
        claim_job = "UPDATE jobs SET Thread_Id = \'{}\' WHERE Job_ID = \'{}\';".format(job_id)
        cursor.execute(claim_job)
        connection.commit()
        print "Job is now claimed"
        cursor.close()
        return True

def call_the_queen(dict_of_job_attributes):
    if check_if_job_claimed(dict_of_job_attributes['job_id']):
        instance = OM(dict_of_job_attributes) #<-- Create instance of my target class
        instance.queen_bee()

#multiprocessing code
import multiprocessing as mp
if __name__ == '__main__':
    active_jobs = get_active_jobs()
    pool = mp.Pool(processes = 4)
    pool.map(call_the_queen,active_jobs)
    pool.close()
    pool.join()

Answer 1

任何非空元组（或列表、字符串、可迭代等）都将评估为True。可迭代的内容是否为非 True 无关紧要。要对此进行测试，您可以使用 any(iterable) 或 all(iterable) 来测试可迭代对象中的任何或所有项目是否评估为 True。

但是，根据您的编辑，您的问题可能是由跨多个进程使用全局连接对象引起的。

相反，每个进程都应该创建自己的连接。

def check_if_job_claimed(job_id):
    connection = mysql.connector.connect(user=creds["mysql_user"],password=creds["mysql_pw"],host=creds["mysql_host"],database=creds["mysql_db"],use_pure=False,buffered=True)

您也可以尝试使用connection pooling，但我不确定这是否可以跨进程工作，并且可能需要您切换到线程。

另外，我会将if __name__ == '__main__': 下的所有代码移动到一个函数中。在使用多处理时，您通常希望避免污染全局命名空间，因为当 python 创建一个新进程时，它会尝试将全局状态复制到新进程中。这可能会导致一些奇怪的错误，因为全局变量不再共享状态（因为它们在单独的进程中），或者当对象在新进程中重建时，它要么无法序列化，要么在序列化过程中丢失一些信息。