python多进程
from multiprocessing import Process
import os
def processFunc(name):
print("child process is %s, pid is %s" %(name, os.getpid() ) )
return
if __name__ == '__main__':
print("Parent process is %s." %(os.getpid() ))
p = Process(target = processFunc, args = ('test', ))
print('Child will start ')
p.start()
p.join()
print("Child stop")
进程池
from multiprocessing import Pool
import os , time, random
def long_time_task(name):
print('run task name is %s' %(name))
start = time.time()
time.sleep(random.random()*3)
end = time.time()
print('Task %s runs %0.2f seconds.' %(name, (end - start )) )
if __name__ == '__main__':
print('Parent pid is %s' %(os.getpid() ))
p = Pool(4)
for i in range(5):
p.apply_async(long_time_task, args = (str(i) ,) )
print("Waiting all processes!!!")
p.close()
p.join()
print("All subprocess done")
启动进程，并调用命令行
import subprocess
print('$ nslookup www.python.org')
r = subprocess.call(['nslookup', 'www.python.org'])
print('Exit code:', r)
队列Queue可实现两个进程间通信
from multiprocessing import Process, Queue
import os, time, random
def write(q):
print('Process to Write pid is %s' %(os.getpid() ) )
for i in ['A','B','C']:
q.put(i)
time.sleep(random.random())
def read(q):
print('Process to Read pid is %s' %(os.getpid() ) )
while(True):
value = q.get(True)
print('Get %s from queue ' %(value))
if __name__ == '__main__':
q = Queue()
pw = Process(target=write, args = (q,))
pr = Process(target = read , args = (q,) )
pw.start()
pr.start()
pw.join()
pr.terminate()
python多线程
import threading , time
def loop():
print('thread %s is running ...' % threading.current_thread().name)
n = 0
while n < 5:
n = n+ 1
print('thread %s >>> %s' %(threading.current_thread().name, n))
time.sleep(1)
print('thread %s ended. ' %(threading.current_thread().name ) )
if __name__ == '__main__':
print('Thread %s is running...' % threading.current_thread().name)
t = threading.Thread(target = loop, name = 'LoopThread')
t.start()
t.join()
print('Thread %s ended.' % threading.current_thread().name)
多线程访问全局变量，记得加锁
import time, threading
# 假定这是你的银行存款:
balance = 0
lock = threading.Lock()
def change_it(n):
# 先存后取，结果应该为0:
global balance
balance = balance + n
balance = balance - n
def run_thread(n):
for i in range(100000):
lock.acquire()
try:
change_it(n)
finally:
lock.release()
t1 = threading.Thread(target=run_thread, args=(5,))
t2 = threading.Thread(target=run_thread, args=(8,))
t1.start()
t2.start()
t1.join()
t2.join()
print(balance)
避免枷锁带来的效率衰退，可使用线程本地变量
import threading
# 创建全局ThreadLocal对象:
local_school = threading.local()
def process_student():
# 获取当前线程关联的student:
std = local_school.student
print('Hello, %s in thread %s' %(std, threading.current_thread().name ))
def process_thread(name):
# 绑定ThreadLocal的student:
local_school.student = name
process_student()
if __name__ == '__main__':
t1 = threading.Thread(target = process_thread, args=('Alice',), name = 'Thread-A')
t2 = threading.Thread(target= process_thread, args=('Bob',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()
分布式进程，用于不同机器通信，采用BaseManager，在masterprocess.py中实现如下
import random, time, queue
from multiprocessing.managers import BaseManager
task_queue = queue.Queue()
result_queue = queue.Queue()
def taskqueuefunc():
global task_queue
return task_queue
def resultqueuefunc():
global result_queue
return result_queue
class QueueManager(BaseManager):
pass
def ServerStart():
QueueManager.register('get_task_queue', callable = taskqueuefunc)
QueueManager.register('get_result_queue', callable = resultqueuefunc)
manager = QueueManager(address=('127.0.0.1', 5000), authkey=b'abc')
manager.start()
task = manager.get_task_queue()
result = manager.get_result_queue()
for i in range(10):
n = random.randint(0,10000)
print('Put task %d...' %n)
task.put(n)
# 从result队列读取结果:
print('Try get results...')
for i in range(10):
r = result.get(timeout=10)
print('Result: %s' % r)
# 关闭:
manager.shutdown()
print('master exit.')
if __name__ == '__main__':
ServerStart()
在另一个文件workprocess.py中实现另一个进程处理数据
import time,sys,queue
from multiprocessing.managers import BaseManager
class QueueManager(BaseManager):
pass
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')
server_addr = '127.0.0.1'
print('Connect to server %s...' % server_addr)
m = QueueManager(address=(server_addr,5000),authkey=b'abc')
m.connect()
task = m.get_task_queue()
result = m.get_result_queue()
for i in range(10):
try:
n = task.get(timeout=1)
print('run task %d %d...' % (n,n))
r = '%d %d = %d' % (n,n,n*n)
time.sleep(1)
result.put(r)
except queue.Empty:
print('task queue is empty.')
print('worker exit.')
先启动masterprocess.py，然后启动workprocess.py，可以看到效果
谢谢关注我的公众号