应用于对象时的多处理池 apply_async 函数会抛出无法腌制 <type 'thread.lock'> 对象

Question

我的__init__ 函数尝试调用调度程序类的函数verify_func()。

def__init__:
def _pickle_method(method):
    func_name = method.im_func.__name__
    obj = method.im_self
    cls = method.im_class
    return _unpickle_method, (func_name, obj, cls)

def _unpickle_method(func_name, obj, cls):
    for cls in cls.mro():
        try:
            func = cls.__dict__[func_name]
        except KeyError:
            pass
    return func.__get__(obj, cls)


if __name__=='__main__':
  while True:
    #Create a scheduler object
    scheduler_obj=Scheduler()
    try:
      #Connect to the database get all the new requests to be verified
      db = Database(scheduler_obj.username_testschema, scheduler_obj.password_testschema, scheduler_obj.mother_host_testschema,
                    scheduler_obj.mother_port_testschema, scheduler_obj.mother_sid_testschema, 0)
      result = db.query("SELECT JOB_ID FROM %s.table_1 WHERE JOB_STATUS='%s'" % (scheduler_obj.username_testschema, 'initiated'))
      copy_reg.pickle(types.MethodType, _pickle_method, _unpickle_method)
      pool = mp.Pool(len(result))
      for row in result:
        verify_id = row[0]
        print mp.current_process()
        try:
          pool.apply_async(scheduler_obj.verify_func, (verify_id))
        except Exception as e:
                    scheduler_obj.logger.exception("MP exception : %s", e)

      pool.close()
      pool.join()

    except Exception as e:
      scheduler_obj.logger.exception(e)
    finally:
      time.sleep(10)

这个verify_func 尝试使用 cx_oracle 对数据库运行一些查询。

def verify_func(self,verify_id):
    print "Verifying with Current Id : ", verify_id
    try:
      db = Database(self.username_testschema, self.password_testschema, self.mother_host_testschema,
                    self.mother_port_testschema, self.mother_sid_testschema, 0)
      result = db.query("select * from %s.table_2 where job_id=%d" % (self.username_testschema,verify_id))
      column_name = [d[0] for d in db.cursor.description]
      final_dictionary = [dict(zip(column_name, row)) for row in result]

      #On getting the new job_ids whose status is .. try to run checks for each one.
      for row in final_dictionary:
        print row

它是否连接并在数据库上运行查询，这会引发此错误，因为它没有说明它到底在哪里中断。

class Database(object):
 def __init__:
    self.connection = cx_Oracle.connect(self.connect_string, mode=self.mode) 

如果需要更多解释，请告诉我。

Answer 1

是的，我做了一个 scheduler_object.dict，我看到我有一个记录器实例变量，它是不可腌制的。谢谢@Dano。

因此，最好的调试方法是找出类变量/方法可能是什么，因为如果您遇到此错误。 此外，您可以覆盖记录器类的 getstate 、 setstate 方法以使其可腌制，如此处所示。 Can't pickle loggers?