#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/17from multiprocessing import Process
import osdef get_id(name):print(name,"Main process:",os.getppid(),"current process;", os.getpid())P1 = Process(target=get_id, args=('andy',))
P2 = Process(target = get_id, args=("Jack", ))if __name__ == "__main__":P2.start()P1.start()print("主进程")

from multiprocessing import Process
import osclass Custom_Process(Process):def __init__(self, name):super().__init__()self.name = namedef run(self):print(self.name, "Main process:", os.getppid(), "current process;", os.getpid())if __name__ == "__main__":P1 = Custom_Process('andy')P2 = Custom_Process("jack")P1.start()P2.start()print("主进程")

super().__init__()

Parent.__init__(self)

server.listen(5)# 设置可以接受的连接数量

#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/16import socket,json, struct, subprocess
from multiprocessing import ProcessBUFF_SIZE = 1024
IP_PORT = ("127.0.0.1", 8081)server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,1)# 重用端口
server.bind(IP_PORT)
server.listen(5)# 设置可以接受的连接数量def communicate(conn, client_addr):while True:# 内层循环为通信循环msg = conn.recv(BUFF_SIZE)if not msg:breakpipes = subprocess.Popen(msg.decode("utf-8"),shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)error = pipes.stderr.read()if error:print("Error:",error)response_msg = errorelse:response_msg = pipes.stdout.read()header = {'data_size':len(response_msg)}# 数据长度header_json = json.dumps(header)#序列化header_json_byte = bytes(header_json,encoding="utf-8")conn.send(struct.pack('i',len(header_json_byte)))#先发送报头长度，仅包含数据长度， 这里的i指int类型conn.send(header_json_byte)# 再发送报头conn.sendall(response_msg)# 正式的信息print("Request from:",client_addr, "Command:",msg)conn.close()
if __name__ == "__main__":while True:# 外层循环为链接循环conn, client_addr = server.accept()p = Process(target=communicate, args=(conn, client_addr))p.start()server.close()

#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/16import socket, json, structBUFF_SIZE = 1024
IP_PORT = ("127.0.0.1", 8081)client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(IP_PORT)while True:msg = input(">>:").strip().encode("utf-8")if not msg:breakclient.send(msg)header = client.recv(4)print("Header:",struct.unpack("i", header))header_length = struct.unpack('i', header)[0]print("Header_length:", header_length)header_json = json.loads(client.recv(header_length).decode("utf-8"))data_size = header_json['data_size']print("Data_size:",data_size)recv_size = 0recv_data = b''while recv_size < data_size:recv_data += client.recv(BUFF_SIZE)recv_size += len(recv_data)print(recv_data.decode("gbk"))client.close()

import os, time
from multiprocessing import Process, Lockdef work(mutex):mutex.acquire()print("%d is working..." % os.getpid())time.sleep(2)print("%d is done!" % os.getpid())mutex.release()if __name__ == "__main__":mutex = Lock()p1 = Process(target=work, args=(mutex,))p2 = Process(target=work, args=(mutex,))p1.start()p2.start()

import json, random, time, os
from multiprocessing import Process , Lockdef search():dic = json.load(open('db.txt',))print("%s查询，车票剩余%s" % (os.getpid(),dic['count']))def get_ticket():dic = json.load(open('db.txt',))if dic['count'] > 0:dic['count'] -= 1time.sleep(random.randint(1,4))json.dump(dic,open('db.txt', 'w'))print('%s 购买成功' % os.getpid())print("车票剩下%s" % dic["count"])else:print("%s抢票失败 " % os.getpid())def task(mutex):search()mutex.acquire()get_ticket()mutex.release()if __name__ == "__main__":mutex = Lock()for i in range(10):p = Process(target=task, args=(mutex,))p.start()

#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/19import os, time
from multiprocessing import Process, Lockdef work(mutex, t):mutex.acquire()print("%s Running at %s\n" % (os.getpid(),time.strftime("%H:%M:%S")))time.sleep(t)mutex.release()print("%s Stop at %s\n" % (os.getpid(),time.strftime("%H:%M:%S")))if __name__ == "__main__":print("Main Process Running at %s\n" % time.strftime("%H:%M:%S"))mutex = Lock()p1 = Process(target=work, args=(mutex,5))p2 = Process(target=work, args=(mutex,3))p1.start()p2.start()p1.join()print("Join1 finish at %s!\n" % time.strftime("%H:%M:%S"))p2.join()print("Join2 finish at %s!\n" % time.strftime("%H:%M:%S"))print("Main Process Stop at %s\n" % time.strftime("%H:%M:%S"))

#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/19from multiprocessing import Process
import timedef work():print("Running...")time.sleep(3)print("Finish!")if __name__ == "__main__":p = Process(target=work,)#p.daemon = Truep.start()print("Main finish!")

#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/19import multiprocessing
import timedef worker(s):s.acquire()print(multiprocessing.current_process().name + "acquire");time.sleep(2)print(multiprocessing.current_process().name + "release\n");s.release()if __name__ == "__main__":s = multiprocessing.Semaphore(2)for i in range(5):p = multiprocessing.Process(target = worker, args=(s,))p.start()

#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/19import random,os
from multiprocessing import Queue,Processdef put_q(q):print("Put...")for i in range(5):n = random.randint(1,5)print(n)q.put(n)def get_q(q):print("\nGet...")while True:if not q.empty():print("%s" % os.getpid(),q.get())else:breakif __name__ == "__main__":q = Queue(8)p1 = Process(target=put_q,args=(q,))p2 = Process(target=get_q,args=(q,))p1.start()p1.join() # 防止进程2先启动，队列为空p2.start()

#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/19import random,os
from multiprocessing import Pipe,Processdef send_p(p):print("send...")for i in range(5):n = random.randint(1,5)print(n)p.send(n)def receive_p(p):print("\nReceive...")while True:print("%s" % os.getpid(),p.recv())if __name__ == "__main__":p = Pipe()p1 = Process(target=send_p,args=(p[0],))p2 = Process(target=receive_p,args=(p[1],))p1.start()p1.join() p2.start()

运行：

8.Pool 进程池

Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，

如果池还没有满，那么就会创建一个新的进程用来执行该请求；

但如果池中的进程数已经达到规定最大值，那么该请求就会等待，

直到池中有进程结束，才会创建新的进程来它。

看例子：

#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/20import time
from multiprocessing import Pool, Processdef work(msg):print(msg, 'is working\n')time.sleep(2)print(msg,'finish!\n')if __name__ == "__main__":pro = Process()pool = Pool(processes=3)for i in range(1,6):msg = "process %s" % ipool.apply_async(work,(msg,))pool.close()pool.join()# 阻塞主进程，等待子进程执行完

运行：

指定进程池只有3个进程，所以第四个进程只有前面结束一个进程时才能开始。

需要说明的是 pool.apply_async()是非阻塞的，pool.apply()则是阻塞的。看区别：

修改：

pool.apply(work,(msg,))

#!/usr/bin/env python
#coding:utf-8
#Created by Andy @ 2017/9/20import time, os
from multiprocessing import Pool, Processdef work(msg):print(msg, 'is working\n')time.sleep(2)print(msg,'finish!\n')return msgdef plus(msg):if msg:msg = msg + '*plus*'print(msg)if __name__ == "__main__":pro = Process()pool = Pool(processes=3)for i in range(1,6):msg = "process %s" % ipool.apply_async(work,(msg,), callback=plus)# 回调函数pool.close()pool.join()