为什么 multiprocessing.pool.map 会引发 PicklingError（编码）？

Question

为什么下面的代码在使用threads时运行，但在使用multiprocessing时抛出异常？

from multiprocessing import Pool
from multiprocessing.dummy import Pool as ThreadsPool
import urllib2

urls = [
  'http://www.python.org',
  'http://www.python.org/about/',
  'http://www.python.org/doc/',
  'http://www.python.org/download/']

def use_threads():

    pool = ThreadsPool(4)
    results = pool.map(urllib2.urlopen, urls)
    pool.close()
    pool.join()

    print [len(x.read()) for x in results]

def use_procs():

    p_pool = Pool(4)
    p_results = p_pool.map(urllib2.urlopen, urls)
    p_pool.close()
    p_pool.join()

    print 'using procs instead of threads'
    print [len(x.read()) for x in p_results]

if __name__ == '__main__':
    use_procs()

例外是

Traceback (most recent call last):
  File "pools.py", line 39, in <module>
    use_procs()
  File "pools.py", line 31, in use_procs
    p_results = p_pool.map(urllib2.urlopen, urls)
  File "/usr/lib64/python2.7/multiprocessing/pool.py", line 250, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/usr/lib64/python2.7/multiprocessing/pool.py", line 554, in get
    raise self._value
multiprocessing.pool.MaybeEncodingError: Error sending result: '[<addinfourl at 35286624 whose fp = <socket._fileobject object at 0x2198ad0>>]'. Reason: 'PicklingError("Can't pickle <type 'instancemethod'>: attribute lookup __builtin__.instancemethod failed",)'

我知道进程和线程之间的通信方式是不同的。为什么pickle 在网站内容上失败？如何设置编码来解决这个问题？

Answer 1

问题不是编码错误，而是由于酸洗错误，因为urllib2.urlopen() 返回的结果是不可提取的对象（_ssl._SSLSocket 根据我收到的错误消息中显示的稍微不同的原因与您的代码)。为了解决这个问题，您可以通过在打开 url 后读取数据来将返回对象的使用限制为子进程本身，如下所示。然而，这可能意味着需要在进程之间传递更多数据。

# Added.
def get_data(url):

    soc = urllib2.urlopen(url)
    return soc.read()

def use_procs():

    p_pool = Pool(4)
#    p_results = p_pool.map(urllib2.urlopen, urls)
    p_results = p_pool.map(get_data, urls)
    p_pool.close()
    p_pool.join()

    print 'using procs instead of threads'
#    print [len(x.read()) for x in results]
    print [len(x) for x in p_results]

输出：

using procs instead of threads
[49062, 41616, 40086, 101224]

Answer 2

正如我已经提到的 - 因为您试图在进程之间传递套接字对象，所以引发了该错误。您必须将脚本逻辑更改为以下内容：

from multiprocessing.pool import Pool
from multiprocessing.pool import ThreadPool
import urllib2

urls = [
  'http://www.python.org',
  'http://www.python.org/about/',
  'http://www.python.org/doc/',
  'http://www.python.org/download/'
]

def worker(url):
    return urllib2.urlopen(url).read() # string returned

def use_threads():

    pool = ThreadPool(4)
    results = pool.map(worker, urls)
    pool.close()
    pool.join()

    print([len(x) for x in results])

def use_procs():

    p_pool = Pool(4)
    p_results = p_pool.map(worker, urls)
    p_pool.close()
    p_pool.join()

    print('using procs instead of threads')
    print([len(x) for x in p_results])

if __name__ == '__main__':
    use_procs()

顺便说一句：你可以做池工厂并从中挑选池，而不是在use_threads和use_procs中重复代码：

from multiprocessing.pool import Pool
from multiprocessing.pool import ThreadPool
import urllib2

urls = [
  'http://www.python.org',
  'http://www.python.org/about/',
  'http://www.python.org/doc/',
  'http://www.python.org/download/'
]


def worker(url):
    return urllib2.urlopen(url).read()


def pool_factory(key, n):
    if key == 'proc':
        print('using procs instead of threads')
        return Pool(n)
    else:
        return ThreadPool(n)


def main():

    pool = pool_factory('proc', 4)  # change `proc` to anything for using ThreadPool
    results = pool.map(worker, urls)
    pool.close()
    pool.join()
    print([len(x) for x in results])


if __name__ == '__main__':
    main()