多线程环境和模块，如 pickle 或 json

Question

我正在使用“导入线程”和 python 3.4。 简单的情况，我有一个主父线程和一个子线程。我需要将我的字典保存到子线程的文件中。在线程函数中我有变量：

def thread_function(...)
    def save_to_file():
        this_thread_data.my_dict or nonlocal this_thread_data.my_dict 
        ... json or pickle


    this_thread_data = local()
    this_thread_data.my_dict = {...}
    ...

当我使用 pickle 时出现错误

_pickle.PicklingError: Can't pickle <class '_thread.lock'>: attribute lookup lock on _thread failed

当我使用 json 时出现错误

TypeError: <threading.Event object at 0x7f49115a9588> is not JSON serializable

pickle 或 json 会在多线程环境中工作还是我需要使用其他东西来代替？

谢谢。

Answer 1

除非您跳出标准库，否则 Python 线程（和多处理）和酸洗会受到破坏和限制。

如果你使用一个名为pathos.multiprocesssing 的multiprocessing 的fork，你可以直接在multiprocessing 的map 函数中使用类和类方法。这是因为dill 被用来代替pickle 或cPickle，而dill 几乎可以在python 中序列化任何东西。 pathos.multiprocessing 提供了线程模块的接口，就像标准的 python 模块一样。

pathos.multiprocessing 还提供了一个异步映射函数……它可以map 具有多个参数的函数（例如map(math.pow, [1,2,3], [4,5,6])）

见： What can multiprocessing and dill do together?

和： http://matthewrocklin.com/blog/work/2013/12/05/Parallelism-and-Serialization/

>>> from pathos.multiprocessing import ProcessingPool as Pool
>>> #from pathos.multiprocessing import ThreadingPool as Pool
>>> 
>>> p = Pool(4)
>>> 
>>> def add(x,y):
...   return x+y
... 
>>> x = [0,1,2,3]
>>> y = [4,5,6,7]
>>> 
>>> p.map(add, x, y)
[4, 6, 8, 10]
>>> 
>>> class Test(object):
...   def plus(self, x, y): 
...     return x+y
... 
>>> t = Test()
>>> 
>>> p.map(Test.plus, [t]*4, x, y)
[4, 6, 8, 10]
>>> 
>>> p.map(t.plus, x, y)
[4, 6, 8, 10]

字典中有不寻常的东西，没关系……

>>> d = {'1':add, '2':t, '3':Test, '4':range(10), '5':1}
>>>                        
>>> def items(x):
...   return x[0],x[1]
... 
>>> p.map(items, d.items())
[('1', <function add at 0x103b7e2a8>), ('3', <class '__main__.Test'>), ('2', <__main__.Test object at 0x103b7ad90>), ('5', 1), ('4', [0, 1, 2, 3, 4, 5, 6, 7, 8, 9])]

顺便说一句，如果你想腌制线程锁，你也可以这样做。

>>> import dill as pickle
>>> import threading
>>> lock = threading.Lock()
>>> 
>>> pickle.loads(pickle.dumps(lock))
<thread.lock object at 0x10c534650>

看起来您想构建某种闭包，自动将函数调用存储到文件或至少存储到序列化字符串。如果那是你想要的， 你可以试试klepto，它给你一个装饰器，你可以应用到你的函数中 您可以缓存到内存或磁盘或数据库。 Klepto 可以使用pickle或者json， 但它被dill 增强了，所以它可以在python 中序列化几乎任何东西——所以不要 担心你的字典里有什么……只需序列化它。

from klepto import lru_cache as memoize
from klepto.keymaps import picklemap
dumps = picklemap(serializer='dill')

class Adder(object):
    """A simple class with a memoized method"""

    @memoize(keymap=dumps, ignore=('self','**'))
    def __call__(self, x, *args, **kwds):
        debug = kwds.get('debug', False)
        if debug:
            print ('debug:', x, args, kwds)
        return sum((x,)+args)
    add = __call__

add = Adder()
assert add(2,0) == 2
assert add(2,0,z=4) == 2          # cached (ignore z)
assert add(2,0,debug=False) == 2  # cached (ignore debug)
assert add(1,2,debug=False) == 3
assert add(1,2,debug=True) == 3   # cached (ignore debug)
assert add(4) == 4
assert add(x=4) == 4              # cached

Klepto 使您能够在重新启动代码时获得所有缓存的结果。 在这种情况下，你会选择一些文件或数据库后端，然后确保你对存档执行 add.dump()... 然后重新启动 python 或其他，然后执行 add.load() 以加载存档结果。

在此处获取代码： https://github.com/uqfoundation

Answer 2

在多线程环境中使用 pickle 和 json 可以正常工作（但可能不是线程安全的，因此请确保您正在酸洗的数据当时不能更改，例如通过使用锁）。问题是您将被限制在可以实际保存到磁盘的数据类型中。

正如您所发现的，并非所有对象都是可序列化的。最简单的方法是确保您的字典仅包含与 pickle 或 json 序列化程序兼容的值。例如，您似乎在字典中存储了一个使 pickle 失败的锁定对象。您可能想创建一个只包含可以腌制的值的新字典，然后腌制它。

或者，如果你想创建一个自定义对象来存储你的数据，你可以告诉 pickle 究竟如何腌制它。这是更高级的，在您的情况下可能没有必要，但您可以在此处找到更多文档：https://docs.python.org/3.4/library/pickle.html#pickling-class-instances

Answer 3

有更好的方法在线程之间共享数据。如果您愿意使用进程而不是线程，我会推荐 python 'multiprocessing' 模块，特别是 'Manager' 类：https://docs.python.org/2/library/multiprocessing.html#managers。这是一个玩具示例：

from multiprocessing import Manager, Process

def on_separate_process(alist):
   print alist

manager = Manager()
alist = manager.list([1,2,3])

p = Process(target=on_separate_process, args=[alist])
p.start()

打印 [1,2,3]