标签: python-multiprocessing

我想创建一个Thread或一个在While True循环中永远运行的进程.

我需要以队列的形式向工作人员发送和接收数据,可以是multiprocessing.Queue()或collections.deque().我更喜欢使用collections.deque(),因为它明显更快.

我还需要能够最终杀死这个工作者(因为它运行了一段时间的True循环.这里有一些测试代码,我把它们放在一起试图理解线程,进程,队列和deque之间的区别.

import time
from multiprocessing import Process, Queue
from threading import Thread
from collections import deque

class ThreadingTest(Thread):

    def __init__(self, q):
        super(ThreadingTest, self).__init__()
        self.q = q
        self.toRun = False

    def run(self):
        print("Started Thread")
        self.toRun = True
        while self.toRun:
            if type(self.q) == type(deque()):
                if self.q:
                    i = self.q.popleft()
                    print("Thread deque: " + str(i))
            elif type(self.q) == type(Queue()):
                if not self.q.empty():
                    i = self.q.get_nowait()
                    print("Thread Queue: " + str(i))

    def stop(self):
        print("Trying to stop Thread")
        self.toRun = False
        while …

有关更多设置，请参阅此问题。我想Toy并行创建大量 class 实例。然后我想将它们写入 xml 树。

import itertools
import pandas as pd
import lxml.etree as et
import numpy as np
import sys
import multiprocessing as mp


def make_toys(df):
    l = []
    for index, row in df.iterrows():
        toys = [Toy(row) for _ in range(row['number'])]
        l += [x for x in toys if x is not None]
    return l


class Toy(object):
    def __new__(cls, *args, **kwargs):
        if np.random.uniform() <= 1:
            return super(Toy, cls).__new__(cls, *args, **kwargs)

    def __init__(self, row):
        self.id = …

在multiprocessing.Pool我试图以相同的顺序显示我的打印.

from multiprocessing import Pool
import time
def func(arg):
    time.sleep(0.001)
    print(arg, end=" ")

proc_pool = Pool(4)
proc_pool.map(func, range(30))

输出是:0 1 8 9 10 11 14 15 6 7 16 17 4 5 12 13 18 19 2 3或类似的.它不在订单中0 1 2 3 ...

我知道imap可以提供更好的订购......但它仍然不是我想要的.我可以覆盖打印功能,将其保存到一个变量中并立即打印出来 - 但我希望它们一完成就显示出来 - 而不是一切都完成.

我试图了解此Python 2.7.5示例脚本的输出：

import time
from multiprocessing import Pool

print(time.strftime('%Y-%m-%d %H:%M', time.localtime(time.time())))
props2=[
            '170339',
            '170357',
            '170345',
            '170346',
            '171232',
            '170363',
            ]
def go(x):
     print(x)

if __name__ == '__main__':
    pool = Pool(processes=3)
    pool.map(go, props2)

print(time.strftime('%Y-%m-%d %H:%M', time.localtime(time.time())))  

这产生输出：

2015-08-06 10:13

2015-08-06 10:13

2015-08-06 10:13

170339

170357

170345

170346

171232

170363

2015-08-06 10:13

2015-08-06 10:13

2015-08-06 10:13

我的问题是：

A）为什么时间在开始和结束时打印三遍？我希望它能打印开始时间，然后打印结束时间。

B）真正的问题-如何使它多次运行一个命令，而一次运行所有其他命令？

以下代码我用于并行csv处理:

#!/usr/bin/env python

import csv
from time import sleep
from multiprocessing import Pool
from multiprocessing import cpu_count
from multiprocessing import current_process
from pprint import pprint as pp

def init_worker(x):
  sleep(.5)
  print "(%s,%s)" % (x[0],x[1])
  x.append(int(x[0])**2)
  return x

def parallel_csv_processing(inputFile, outputFile, header=["Default", "header", "please", "change"], separator=",", skipRows = 0, cpuCount = 1):
  # OPEN FH FOR READING INPUT FILE
  inputFH   = open(inputFile,  "rt")
  csvReader = csv.reader(inputFH, delimiter=separator)

  # SKIP HEADERS
  for skip in xrange(skipRows):
    csvReader.next()

  # PARALLELIZE COMPUTING INTENSIVE OPERATIONS - …

我正在尝试加速使用巨大矩阵的算法.我将它并行化以对行进行操作,并将数据矩阵放在共享内存中,这样系统就不会被堵塞.但是,我没有像我希望的那样顺利地工作,它现在抛出一个关于文件的奇怪错误,我不理解,因为我甚至没有打开文件中的文件.

模拟大致正在进行的程序,1000次迭代代表算法中发生的事情.

import multiprocessing
import ctypes
import numpy as np

shared_array_base = multiprocessing.Array(ctypes.c_double, 10*10)
shared_array = np.ctypeslib.as_array(shared_array_base.get_obj())
shared_array = shared_array.reshape(10, 10)

def my_func(i, shared_array):
    shared_array[i,:] = i

def pool_init(_shared_array, _constans):
    global shared_array, constans
    shared_array = _shared_array
    constans = _constans

def pool_my_func(i):
    my_func(i, shared_array)

if __name__ == '__main__':
    for i in np.arange(1000):
        pool = multiprocessing.Pool(8, pool_init, (shared_array, 4))
        pool.map(pool_my_func, range(10))
    print(shared_array)

这引发了这个错误(我在OSX上):

Traceback (most recent call last):
  File "weird.py", line 24, in <module>
    pool = multiprocessing.Pool(8, pool_init, (shared_array, 4))
  File "//anaconda/lib/python3.4/multiprocessing/context.py", …

我有以下代码,我需要一次读取多个传感器.我已经设置了线程和多处理来为我完成这项任务.当线程和多重处理代码在主类之外时,它可以正常工作,但是类不能使用它所检索的数据.当我把mutlithreading代码insdie这个类时,我遇到了一个EOFError: Ran out of input错误.

这是代码:

import os
import multiprocessing
from multiprocessing import Process, Pool
import threading
import queue
import tkinter as tk
from tkinter import *
from tkinter import ttk
import time
import minimalmodbus
import serial
minimalmodbus.CLOSE_PORT_AFTER_EACH_CALL = True
THREAD_LOCK = threading.Lock()

class Application(tk.Frame):
    def __init__(self, master=None):
        super().__init__(master)

        self.pack()

        self.first_gas_labelframe = LabelFrame(self, text="Gas 1", width=100)
        self.first_gas_labelframe.grid(row=0, column=0)

        self.value_label = Label(self.first_gas_labelframe, text="Value")
        self.value_label.grid(row=0, column=0)

        self.unit_label = Label(self.first_gas_labelframe, text="Unit")
        self.unit_label.grid(row=1, column=0)

        self.temp_label = Label(self.first_gas_labelframe, text="Temp")
        self.temp_label.grid(row=2, column=0)

        self.temp_label6 = …

假设我有两个过程：一个读者和一个作家。作者如何检测读者何时完成了价值写作？

该multiprocessing模块有一个带有close方法的队列，该方法似乎是为此目的定制的。但是，如何检测队列何时关闭？

这似乎不起作用，因为作者从未退出过：

import multiprocessing as mp

def getter(q):
    while True:
        try:
            print(q.get())
        except Exception:
            break

def putter(q):
    q.put(1)
    q.put(2)
    q.close()

q = mp.Queue()
writer = mp.Process(target=putter, args=(q, ))
reader = mp.Process(target=getter, args=(q, ))
reader.start()
writer.start()

writer.join()
reader.join()

读写器对应该使用标记值来表示写入结束吗？那么使用该close方法有什么意义呢？

编辑：虽然这个问题询问有关Queue模块（现在称为queue），但我专门询问的是mp.Queue什么以及该.close方法的正确用法是什么。

我目前正在尝试使用python多处理.我使用的库是multiprocess(NOT multiprocessing).

我有以下代码,它创建了许多计算作业,并通过映射操作运行它:

pool = multiprocess.Pool(4)
all_responses = pool.map_async(wrapper_singlerun, range(10000))
pool.join()
pool.close()

但是,每当我运行此代码片段时,我都会收到以下错误:

    pool.join()
  File "/Users/davidal/miniconda3/lib/python3.6/site-packages/multiprocess/pool.py", line 509, in join
    assert self._state in (CLOSE, TERMINATE)
AssertionError

你知道为什么会发生这个错误吗？我pool.map_async以前用过,但想到我需要一个pool rendez-vous命令.否则,我的PC创建了类似forkbomb的东西,它创建了太多的线程(至少,这是我认为它做的......)

任何想法都表示赞赏!

我是python的新开发者.我的代码是以下代码:

import warnings
import requests
import multiprocessing

from colorama import init
init(autoreset=True)

from requests.packages.urllib3.exceptions import InsecureRequestWarning
warnings.simplefilter("ignore", UserWarning)
warnings.simplefilter('ignore', InsecureRequestWarning)

from bs4 import BeautifulSoup as BS

headers = {'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/39.0.2171.95 Safari/537.36'}


class Worker(multiprocessing.Process):

    def run(self):
        with open('ips.txt', 'r') as urls:
            for url in urls.readlines():
                req = url.strip()
                try:
                    page = requests.get(req, headers=headers, verify=False, allow_redirects=False, stream=True,
                                        timeout=10)
                    soup = BS(page.text)
                    # string = string.encode('ascii', 'ignore')
                    print('\033[32m' + req + ' …