相关疑难解决方法(0)

我想使用multiprocessing的Pool.map()功能,同时划分出工作.当我使用以下代码时,它工作正常:

import multiprocessing

def f(x):
    return x*x

def go():
    pool = multiprocessing.Pool(processes=4)        
    print pool.map(f, range(10))


if __name__== '__main__' :
    go()

但是,当我在面向对象的方法中使用它时,它不起作用.它给出的错误信息是:

PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup
__builtin__.instancemethod failed

当以下是我的主程序时会发生这种情况:

import someClass

if __name__== '__main__' :
    sc = someClass.someClass()
    sc.go()

以下是我的someClass课程:

import multiprocessing

class someClass(object):
    def __init__(self):
        pass

    def f(self, x):
        return x*x

    def go(self):
        pool = multiprocessing.Pool(processes=4)       
        print pool.map(self.f, range(10))

任何人都知道问题可能是什么,或者一个简单的方法呢？

我正在创建一个threading.Timer(2,work)运行线程.在每个工作函数内部,在某些条件下,全局计数器必须增加而不会发生冲突,以便在生成的工作线程中访问计数器变量.

我已经尝试过Queue.Queue指定计数器以及threading.Lock().这是实现线程安全的全局增量变量的最佳方法.

以前有人在这里问过问题:Python线程.如何锁定线程？

我有几个屏幕.其中一个(DataScreen)包含8个标签,应显示当前传感器值.传感器由单独的进程读取(从MainScreen启动).过程本身就是一个例子multiprocessing.Process.

我可以参考标签 sensor_labels = self.manager.get_screen('data').l

但是,我无法弄清楚如何在子进程中更改它们.我可以从任何不是单独进程的函数中更改它们,只需执行以下操作:

for item in sensor_labels:
    item.text = 'Update'

不幸的是,将参考传递sensor_labels给工人似乎更加困难.如果我将它们作为参数传递,则两个进程(kivy和worker)似乎共享同一个对象(id是相同的).但是,如果我改变label.text = 'New Text'Kivy没有任何变化.

为什么两个对象的id都相同,但文本没有改变？我如何与另一个进程共享一个Kivy标签对象？

这是我工作的最小例子

#! /usr/bin/env python
""" Reading sensor data
"""
from kivy.config import Config
Config.set('kivy', 'keyboard_mode', 'multi')
from kivy.app import App
from kivy.lang import Builder
from kivy.properties import StringProperty, ObjectProperty, NumericProperty
from kivy.uix.label import Label
from kivy.uix.screenmanager import ScreenManager, Screen
from kivy.uix.stacklayout import StackLayout
from multiprocessing import Process, Queue, Array
# all other modules
import time
import …

嘿大家,我现在在python中使用多处理.而我只是想知道是否存在某种简单的计数器变量,每个进程在完成处理某个任务时可能只是递增(有点像总共完成了多少工作).

我查找了API for Value,不要认为它是可变的.

鉴于big.txt来自norvig.com/big.txt,我们的目标是快速计算双子座(想象一下,我必须重复这次计数100,000次).

根据python中的Fast/Optimize N-gram实现,像这样提取bigrams是最优的:

_bigrams = zip(*[text[i:] for i in range(2)])

如果我正在使用Python3,生成器将不会被评估,直到我实现它list(_bigrams)或其他一些将执行相同的功能.

import io
from collections import Counter

import time
with io.open('big.txt', 'r', encoding='utf8') as fin:
     text = fin.read().lower().replace(u' ', u"\uE000")

while True: 
    _bigrams = zip(*[text[i:] for i in range(2)])
    start = time.time()
    top100 = Counter(_bigrams).most_common(100)
    # Do some manipulation to text and repeat the counting.
    text = manipulate(text, top100)      

但是每次迭代需要大约1秒以上,100,000次迭代会太长.

我也尝试过sklearnCountVectorizer,但是提取,计算和获得top100双字母的时间与原生python相当.

然后我尝试了一些multiprocessing,使用Python多处理和共享计数器的轻微修改和 …

我花了一些时间试图理解多处理,虽然它的细节避开了我未经训练的头脑.我已经能够得到一个池来返回一个简单的整数,但如果该函数不仅仅返回一个结果,就像我可以找到的所有例子一样(即使在文档中,这是一个不起眼的例子,我也不太明白.

这是一个我正在努力工作的例子.但是,我不能按预期工作,我确信这有一个简单的原因.我可能需要使用队列或共享内存或管理器,但是当我阅读文档时,我似乎无法将我的大脑包围在它实际意味着什么以及它做什么.到目前为止,我所能了解的是泳池功能.

另外,我正在使用一个类,因为我需要避免使用全局变量,就像在这个问题的答案中一样.

import random

class thisClass:
    def __init__(self):
        self.i = 0

def countSixes(myClassObject):
    newNum = random.randrange(0,10)
    #print(newNum) #this proves the function is being run if enabled
    if newNum == 6:
        myClassObject.i += 1

if __name__ == '__main__':
    import multiprocessing
    pool = multiprocessing.Pool(1) #use one core for now

    counter = thisClass()

    myList = []
    [myList.append(x) for x in range(1000)]

    #it must be (args,) instead of just i, apparently
    async_results = [pool.apply_async(countSixes, (counter,)) for i in myList] …

我想使用multiprocessing.Value+multiprocessing.Lock在不同的进程之间共享一个计数器。例如：

import itertools as it
import multiprocessing

def func(x, val, lock):
    for i in range(x):
        i ** 2
    with lock:
        val.value += 1
        print('counter incremented to:', val.value)

if __name__ == '__main__':
    v = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock()

    with multiprocessing.Pool() as pool:
        pool.starmap(func, ((i, v, lock) for i in range(25)))
    print(counter.value())

这将引发以下异常：

RuntimeError：同步对象只能通过继承在进程之间共享

我最困惑的是，一个相关的（虽然不是完全类似的）模式适用于multiprocessing.Process()：

if __name__ == '__main__':
    v = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock()

    procs = [multiprocessing.Process(target=func, args=(i, v, lock))
             for i in …

我知道使用传统的多处理库我可以声明一个值并在进程之间共享状态。

https://docs.python.org/3/library/multiprocessing.html?highlight=multiprocessing#sharing-state-between-processes

使用较新的concurrent.futures库时，如何在进程之间共享状态？

import concurrent.futures

def get_user_object(batch):
    # do some work
    counter = counter + 1
    print(counter)

def do_multithreading(batches):
    with concurrent.futures.ThreadPoolExecutor(max_workers=25) as executor:
        threadingResult = executor.map(get_user_object, batches)

def run():
    data_pools = get_data()
    start = time.time()
    with concurrent.futures.ProcessPoolExecutor(max_workers=PROCESSES) as executor:
        processResult = executor.map(do_multithreading, data_pools)
    end = time.time()
    print("TIME TAKEN:", end - start)

if __name__ == '__main__':
    run()

我想保持这个计数器的同步值。

在以前的库中，我可能使用过multiprocessing.Value和Lock.

我正在尝试使用一起购买的商品的次数来制作一个非常简单的商品推荐系统，

所以首先我创建了一个像 Counter 这样的 item2item 字典

# people purchased A with B 4 times, A with C 3 times.
item2item = {'A': {'B': 4, 'C': 3}, 'B': {'A': 4, 'C': 2}, 'C':{'A': 3, 'B': 2}}
# recommend user who purchased A and C
samples_list = [['A', 'C'], ...]    

因此，对于samples = ['A', 'C']，我建议最多item2item['A'] + item2item['C']。

但是，合并对于大矩阵很重，所以我尝试使用多处理，如下所示

from operator import add
from functools import reduce
from concurrent.futures import ProcessPoolExecutor
from collections import Counter

with ProcessPoolExecutor(max_workers=10) as pool:
    for samples in samples_list:
        # …