多处理redis实例-thread_lock错误答案

【问题标题】：Multiprocessing redis instance -thread_lock error多处理redis实例-thread_lock错误
【发布时间】：2023-03-16 22:13:01
【问题描述】：

我有一个大数据框（百万行），我通过根据特定列进行过滤来创建更多数据框。现在，我想将数据插入到我将执行一些繁重计算的 redis 中。我正在尝试创建一个 redis 实例，并尝试通过多处理将数据插入到 redis 数据库中。我是多处理的新手，在插入数据时遇到错误。我不确定它是否可以完成以及为什么我会收到 thread_lock 错误。谁能解释一下，我该如何继续解决问题。

代码如下：

from multiprocessing import Process, Queue
r=redis.StrictRedis(host='localhost',port='6379',db=0)
q=Queue()
p1=Process(target=insertintoRedis,args=(df1,q,r))
p2=Process(target=insertintoRedis,args=(df2,q,r))
p1.start()
p2.start()
p1.join() 
p2.join() 


def insertintoRedis( df, q ,r):
  for row in df.values:
    r.hset(row[-5],row[0],row[-4])
  return

我在 p1.start() 处收到此错误：

TypeError: cannot pickle '_thread.lock' object

【问题讨论】：

标签： python redis multiprocessing python-multiprocessing

【解决方案1】：

我怀疑问题是r，redis.StrictRedis 的实例，不能被腌制并因此作为参数传递给insertIntoRedis，并且每个进程必须创建自己的redis.StrictRedis 实例。如果您将多次调用insetintoRedis（不幸的是，您的代码过于简化了），那么最好使用进程池，您只需为池中的每个进程创建一个redis.StrictRedis 的实例，可以重复使用。这也可以从您的工作函数insetintoRedis 获取返回值（如果将multiprocessing.Queue 实例传递给此函数的原因是为了获得结果，您将不再需要它——但这又不能从您缺少代码中推断出来的）。

大致思路如下：

from multiprocessing import Pool, Queue
from functools import partial

def init_pool():
    global r
    r = redis.StrictRedis(host='localhost', port='6379', db=0)

# right now limit the pool size to 2 since we only have two tasks:    
q = Queue()
pool = Pool(2, initializer=init_pool)
return_values = pool.map(partial(insertintoRedis, q), [df1, df2])

# note that the order of the arguments has been changed:
def insertintoRedis(q, df):
  for row in df.values:
    r.hset(row[-5],row[0],row[-4])
  return # or return some_value

如果您有大量数据帧并且您希望避免创建map 所需的大列表并且可以使用生成器函数或表达式来代替，那么multiprocessing.Pool.imap 可能是更合适的方法。

【讨论】：