pythongetpid

A. python中extract_tags()怎麼對多行文本提取特徵詞而不是一行一行計算

[python] view plain
#coding:utf-8
import sys
reload(sys)
sys.setdefaultencoding("utf-8")
from multiprocessing import Pool,Queue,Process
import multiprocessing as mp
import time,random
import os
import codecs
import jieba.analyse
jieba.analyse.set_stop_words("yy_stop_words.txt")

def extract_keyword(input_string):
#print("Do task by process {proc}".format(proc=os.getpid()))
tags = jieba.analyse.extract_tags(input_string, topK=100)
#print("key words:{kw}".format(kw=" ".join(tags)))
return tags

#def parallel_extract_keyword(input_string,out_file):
def parallel_extract_keyword(input_string):
#print("Do task by process {proc}".format(proc=os.getpid()))
tags = jieba.analyse.extract_tags(input_string, topK=100)
#time.sleep(random.random())
#print("key words:{kw}".format(kw=" ".join(tags)))
#o_f = open(out_file,'w')
#o_f.write(" ".join(tags)+"\n")
return tags
if __name__ == "__main__":

data_file = sys.argv[1]
with codecs.open(data_file) as f:
lines = f.readlines()
f.close()

out_put = data_file.split('.')[0] +"_tags.txt"
t0 = time.time()
for line in lines:
parallel_extract_keyword(line)
#parallel_extract_keyword(line,out_put)
#extract_keyword(line)
print("串列處理花費時間{t}".format(t=time.time()-t0))

pool = Pool(processes=int(mp.cpu_count()*0.7))
t1 = time.time()
#for line in lines:
#pool.apply_async(parallel_extract_keyword,(line,out_put))
#保存處理的結果，可以方便輸出到文件
res = pool.map(parallel_extract_keyword,lines)
#print("Print keywords:")
#for tag in res:
#print(" ".join(tag))

pool.close()
pool.join()
print("並行處理花費時間{t}s".format(t=time.time()-t1))

運行：
python data_process_by_multiprocess.py message.txt
message.txt是每行是一個文檔，共581行，7M的數據

運行時間：

不使用sleep來掛起進程,也就是把time.sleep(random.random())注釋掉，運行可以大大節省時間。

B. python怎麼讓進程暫停

您的意思是要將進程掛起(Suspend) 而非 阻塞(Block)
如果用sleep() 進程將阻塞
假設進程下有兩個線程 那麼這兩個線程會繼續運行
要使進程掛起 可以考慮使用psutil
import psutil
p = psutil.Process(pid)
p.suspend() #掛起進程
p.resume() #恢復進程

為了證明效果 我寫了一個簡單的進程Process
其下有兩個線程 讀者Reader 和 寫者Writer（簡單的讀者寫者問題）

Process:
import threading

from time import ctime, sleep
import ThreadInReadAndWriteProblem
import multiprocessing
import os

class Process(multiprocessing.Process):

def __init__(self):
multiprocessing.Process.__init__(self) #手動實現父類
pid = os.getpid()

def run(self):
print '當前運行進程PID : %s ' %self.pid #子線程的id與父進程的pid相同 屬於 同一個進程

for i in range(0,5):
r = ThreadInReadAndWriteProblem.Reader()
w = ThreadInReadAndWriteProblem.Writer()
w.start()
r.start()

print '進程阻塞'
sleep(10) #總共運行時間10秒

Reader&Writer
import threading
from time import ctime, sleep
import os

mutex = threading.Lock() #互斥鎖
mutex_readercount = threading.Lock() #計數時的互斥 計算當前正在讀的數目
readerCount = 0 number = 0

#不滿足條件的 進入阻塞狀態

class Reader(threading.Thread): #讀者
def __init__(self):
threading.Thread.__init__(self) #繼承父類構造函數

def run(self):
global mutex
global readerCount
#print '線程PID: %s ' %os.getpid()
while True:
mutex_readercount.acquire()
readerCount +=1
if readerCount == 1:
print '讀者進程等待中，編號%s' %(self.name)
mutex.acquire() == False # 第一個需要申請

mutex_readercount.release()
print '開始讀 , 讀者編號 %s ,現在時間是 %s' %(self.name,ctime())
sleep(2)
print '完成讀 , 讀者編號 %s , 現在時間是 %s' %(self.name,ctime())

mutex_readercount.acquire()
readerCount -= 1
if readerCount == 0: #所有讀者均完成
print '最後一個讀者完成讀 '
mutex.release()
mutex_readercount.release()

class Writer(threading.Thread): #寫者
def __init__(self):
threading.Thread.__init__(self)

def run(self):
global mutex
global writerCount
#print '線程PID： %s' %os.getpid()
while True:
print '寫者進程等待中 編號： %s' %(self.name)
mutex.acquire()
print '開始寫 編號：%s 現在時間是： %s ' %(self.name,ctime())
sleep(5)
print '結束寫 編號: %s 現在時間是 %s' %(self.name,ctime())
mutex.release()

測試程序
import ThreadInReadAndWriteProblem
import
import psutil
import Scheler
from time import ctime, sleep

def main():
p = .Process()
p.start()

sleep(3)

stop(p.pid)
print '進程掛起 %s' %ctime()
sleep(5)

wake(p.pid)
print '喚醒進程 %s' %ctime()

def stop(pid):
print '進程暫停 進程編號 %s ' %(pid)
p = psutil.Process(pid)
p.suspend()

def wake(pid):
print '進程恢復 進程編號 %s ' %(pid)
p = psutil.Process(pid)
p.resume()

if __name__ == '__main__':
main()

結果：
當前運行進程PID : 3096
寫者進程等待中 編號： Thread-2
開始寫 編號：Thread-2 現在時間是： Mon Nov 30 21:12:12 2015
讀者進程等待中，編號Thread-1
寫者進程等待中 編號： Thread-4
進程阻塞
寫者進程等待中 編號： Thread-6
寫者進程等待中 編號： Thread-8
寫者進程等待中 編號： Thread-10
進程暫停 進程編號 3096
進程掛起 Mon Nov 30 21:12:15 2015
進程恢復 進程編號 3096
喚醒進程 Mon Nov 30 21:12:20 2015
結束寫 編號: Thread-2 現在時間是 Mon Nov 30 21:12:20 2015
寫者進程等待中 編號： Thread-2
開始讀 , 讀者編號 Thread-1 ,現在時間是 Mon Nov 30 21:12:20 2015

開始讀 , 讀者編號 Thread-3 ,現在時間是 Mon Nov 30 21:12:20 2015
開始讀 , 讀者編號 Thread-5 ,現在時間是 Mon Nov 30 21:12:20 2015
開始讀 , 讀者編號 Thread-7 ,現在時間是 Mon Nov 30 21:12:20 2015
開始讀 , 讀者編號 Thread-9 ,現在時間是 Mon Nov 30 21:12:20 2015
完成讀 , 讀者編號 Thread-1 , 現在時間是 Mon Nov 30 21:12:22 2015
完成讀 , 讀者編號 Thread-3 , 現在時間是 Mon Nov 30 21:12:22 2015
完成讀 , 讀者編號 Thread-5 , 現在時間是 Mon Nov 30 21:12:22 2015
完成讀 , 讀者編號 Thread-7 , 現在時間是 Mon Nov 30 21:12:22 2015

C. python 多進程

基於官方文檔：
https://docs.python.org/zh-cn/3/library/multiprocessing.html
日樂購，剛才看到的一個博客，寫的都不太對，還是基於官方的比較穩妥
我就是喜歡抄官方的，哈哈

通常我們使用Process實例化一個進程，並調用 他的 start() 方法啟動它。
這種方法和 Thread 是一樣的。

上圖中，我寫了 p.join() 所以主進程是 等待 子進程執行完後，才執行 print("運行結束")
否則就是反過來了（這個不一定，看你的語句了，順序其實是隨機的）例如：

主進加個 sleep

所以不加join() ,其實子進程和主進程是各干各的，誰也不等誰。都執行完後，文件運行就結束了

上面我們用了 os.getpid() 和 os.getppid() 獲取 當前進程，和父進程的id
下面就講一下，這兩個函數的用法：
os.getpid()
返回當前進程的id
os.getppid()
返回父進程的id。 父進程退出後，unix 返回初始化進程（1）中的一個
windows返回相同的id (可能被其他進程使用了)
這也就解釋了，為啥我上面 的程序運行多次， 第一次列印的parentid 都是 14212 了。
而子進程的父級 process id 是調用他的那個進程的 id ： 1940

視頻筆記：
多進程：使用大致方法：

參考： 進程通信（pipe和queue）

pool.map （函數可以有return 也可以共享內存或queue） 結果直接是個列表

poll.apply_async() （同map,只不過是一個進程，返回結果用 xx.get() 獲得）

報錯：

參考 ： https://blog.csdn.net/xiemanR/article/details/71700531

把 pool = Pool() 放到 if name == " main ": 下面初始化搞定。
結果：

這個肯定有解釋的

測試多進程計算效果：
進程池運行：

結果：

普通計算：

我們同樣傳入 1 2 10 三個參數測試：

其實對比下來開始快了一半的；
我們把循環里的數字去掉一個 0；
單進程：

多進程：

兩次測試 單進程/進程池 分別為 0.669 和 0.772 幾乎成正比的。
問題 二：
視圖：
post 視圖裡面

Music 類：

直接報錯：

寫在 類裡面也 在函數里用 self.pool 調用也不行，也是相同的錯誤。

最後 把 pool = Pool 直接寫在 search 函數裡面，奇跡出現了：

前台也能顯示搜索的音樂結果了

總結一點，進程這個東西，最好 寫在 直接運行的函數裡面，而不是 一個函數跳來跳去。因為最後可能 是在子進程的子進程運行的，這是不許的，會報錯。
還有一點，多進程運行的函數對象，不能是 lambda 函數。也許lambda 虛擬，在內存？？

使用 pool.map 子進程 函數報錯，導致整個 pool 掛了:
參考： https://blog.csdn.net/hedongho/article/details/79139606
主要你要，對函數內部捕獲錯誤，而不能讓異常拋出就可以了。

關於map 傳多個函數參數
我一開始，就是正常思維，多個參數，搞個元祖，讓參數一一對應不就行了：

報錯：

參考：
https://blog.csdn.net/qq_15969343/article/details/84672527
普通的 process 當讓可以穿多個參數，map 卻不知道咋傳的。
apply_async 和map 一樣，不知道咋傳的。

最簡單的方法：
使用 starmap 而不是 map

結果：
子進程結束
1.8399453163146973
成功拿到結果了

關於map 和 starmap 不同的地方看源碼：

關於apply_async() ,我沒找到多參數的方法，大不了用 一個迭代的 starmap 實現。哈哈

關於 上面源碼裡面有 itertools.starmap
itertools 用法參考：
https://docs.python.org/zh-cn/3/library/itertools.html#itertool-functions

有個問題，多進程最好不要使用全部的 cpu , 因為這樣可能影響其他任務，所以 在進程池 添加 process 參數 指定，cpu 個數：

上面就是預留了 一個cpu 干其他事的

後面直接使用 Queue 遇到這個問題：

解決：
Manager().Queue() 代替 Queue()

因為 queue.get() 是堵塞型的，所以可以提前判斷是不是 空的，以免堵塞進程。比如下面這樣：
使用 queue.empty() 空為True

D. python怎麼獲得進程的pid

#-*-encoding:UTF-8-*-
importos
importsys
importstring
importpsutil
importre

defget_pid(name):
process_list=psutil.get_process_list()
regex="pid=(d+),sname='"+name+"'"
printregex
pid=0
forlineinprocess_list:
process_info=str(line)
ini_regex=re.compile(regex)
result=ini_regex.search(process_info)
ifresult!=None:
pid=string.atoi(result.group(1))
printresult.group()
break
defmain(argv):<br>name=argv[1]<br>get_pid(name)

if__name__=="__main__":
main(sys.argv)

E. python多進程中隊列不空時阻塞，求解為什麼

最近接觸一個項目，要在多個虛擬機中運行任務，參考別人之前項目的代碼，採用了多進程來處理，於是上網查了查python中的多進程

一、先說說Queue（隊列對象）

Queue是python中的標准庫，可以直接import 引用，之前學習的時候有聽過著名的「先吃先拉」與「後吃先吐」，其實就是這里說的隊列，隊列的構造的時候可以定義它的容量，別吃撐了，吃多了，就會報錯,構造的時候不寫或者寫個小於1的數則表示無限多

import Queue

q = Queue.Queue(10)

向隊列中放值(put)

q.put(『yang')

q.put(4)

q.put([『yan','xing'])

在隊列中取值get()

默認的隊列是先進先出的

>>> q.get()
『yang'
>>> q.get()
4
>>> q.get()
[『yan', 『xing']

當一個隊列為空的時候如果再用get取則會堵塞，所以取隊列的時候一般是用到

get_nowait()方法，這種方法在向一個空隊列取值的時候會拋一個Empty異常

所以更常用的方法是先判斷一個隊列是否為空，如果不為空則取值

隊列中常用的方法

Queue.qsize() 返回隊列的大小
Queue.empty() 如果隊列為空，返回True,反之False
Queue.full() 如果隊列滿了，返回True,反之False
Queue.get([block[, timeout]]) 獲取隊列，timeout等待時間
Queue.get_nowait() 相當Queue.get(False)
非阻塞 Queue.put(item) 寫入隊列，timeout等待時間
Queue.put_nowait(item) 相當Queue.put(item, False)

二、multiprocessing中使用子進程概念

from multiprocessing import Process

可以通過Process來構造一個子進程

p = Process(target=fun,args=(args))

再通過p.start()來啟動子進程

再通過p.join()方法來使得子進程運行結束後再執行父進程

from multiprocessing import Process
import os

# 子進程要執行的代碼
def run_proc(name):
print 'Run child process %s (%s)...' % (name, os.getpid())

if __name__=='__main__':
print 'Parent process %s.' % os.getpid()
p = Process(target=run_proc, args=('test',))
print 'Process will start.'
p.start()
p.join()
print 'Process end.'

上面的程序運行後的結果其實是按照上圖中1，2，3分開進行的，先列印1，3秒後列印2，再3秒後列印3

代碼中的p.close()是關掉進程池子，是不再向裡面添加進程了，對Pool對象調用join()方法會等待所有子進程執行完畢，調用join()之前必須先調用close()，調用close()之後就不能繼續添加新的Process了。

當時也可以是實例pool的時候給它定義一個進程的多少

如果上面的代碼中p=Pool(5)那麼所有的子進程就可以同時進行

三、多個子進程間的通信

多個子進程間的通信就要採用第一步中說到的Queue，比如有以下的需求，一個子進程向隊列中寫數據，另外一個進程從隊列中取數據，

#coding:gbk

from multiprocessing import Process, Queue
import os, time, random

# 寫數據進程執行的代碼:
def write(q):
for value in ['A', 'B', 'C']:
print 'Put %s to queue...' % value
q.put(value)
time.sleep(random.random())

# 讀數據進程執行的代碼:
def read(q):
while True:
if not q.empty():
value = q.get(True)
print 'Get %s from queue.' % value
time.sleep(random.random())
else:
break

if __name__=='__main__':
# 父進程創建Queue，並傳給各個子進程：
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
# 啟動子進程pw，寫入:
pw.start()
# 等待pw結束:
pw.join()
# 啟動子進程pr，讀取:
pr.start()
pr.join()
# pr進程里是死循環，無法等待其結束，只能強行終止:
print
print '所有數據都寫入並且讀完'

四、關於上面代碼的幾個有趣的問題

if __name__=='__main__':
# 父進程創建Queue，並傳給各個子進程：
q = Queue()
p = Pool()
pw = p.apply_async(write,args=(q,))
pr = p.apply_async(read,args=(q,))
p.close()
p.join()

print
print '所有數據都寫入並且讀完'

如果main函數寫成上面的樣本，本來我想要的是將會得到一個隊列，將其作為參數傳入進程池子里的每個子進程，但是卻得到

RuntimeError: Queue objects should only be shared between processes through inheritance

的錯誤，查了下，大意是隊列對象不能在父進程與子進程間通信，這個如果想要使用進程池中使用隊列則要使用multiprocess的Manager類

if __name__=='__main__':
manager = multiprocessing.Manager()
# 父進程創建Queue，並傳給各個子進程：
q = manager.Queue()
p = Pool()
pw = p.apply_async(write,args=(q,))
time.sleep(0.5)
pr = p.apply_async(read,args=(q,))
p.close()
p.join()

print
print '所有數據都寫入並且讀完'

這樣這個隊列對象就可以在父進程與子進程間通信，不用池則不需要Manager，以後再擴展multiprocess中的Manager類吧

關於鎖的應用，在不同程序間如果有同時對同一個隊列操作的時候，為了避免錯誤，可以在某個函數操作隊列的時候給它加把鎖，這樣在同一個時間內則只能有一個子進程對隊列進行操作，鎖也要在manager對象中的鎖

#coding:gbk

from multiprocessing import Process,Queue,Pool
import multiprocessing
import os, time, random

# 寫數據進程執行的代碼:
def write(q,lock):
lock.acquire() #加上鎖
for value in ['A', 'B', 'C']:
print 'Put %s to queue...' % value
q.put(value)
lock.release() #釋放鎖

# 讀數據進程執行的代碼:
def read(q):
while True:
if not q.empty():
value = q.get(False)
print 'Get %s from queue.' % value
time.sleep(random.random())
else:
break

if __name__=='__main__':
manager = multiprocessing.Manager()
# 父進程創建Queue，並傳給各個子進程：
q = manager.Queue()
lock = manager.Lock() #初始化一把鎖
p = Pool()
pw = p.apply_async(write,args=(q,lock))
pr = p.apply_async(read,args=(q,))
p.close()
p.join()

print
print '所有數據都寫入並且讀完'