0704 Process继承实现多进程、Pool进程池，进程间通过队列通信，Pool实现多进程实现复制文件

通过继承的方式，实现Process多进程

from multiprocessing import Process
import time

class MyNewProcess(Process):
    def run(self):
        for i in range(10):
            print("----run----")
            time.sleep(1)


if __name__ == "__main__":
    p = MyNewProcess()

    p.start()        # Process 中的 start 方法会主动调用 run 方法
for i in range(10): print("---main---") time.sleep(1)

Pool 进程池实现多进程

import time
from multiprocessing import Pool

def worker():
    for i in range(10):
        print("From worker %s"%i)
        time.sleep(0.5)

def foo():
    for i in range(10):
        print("From foo %s"%i)
        time.sleep(0.5)

def bar():
    for i in range(10):
        print("From bar %s"%i)
        time.sleep(0.5)

if __name__ == "__main__":
    pool = Pool(3)            # 创建三个 进程
    pool.apply_async(worker)
    pool.apply_async(foo)
    pool.apply_async(bar)

    pool.close()                  # 关闭进程池，禁止添加任务
    pool.join()          # 等待子进程结束后，主进程才往下走
    print("Is done...")



# process and Pool 最后都是调用 fork
# 通常情况下，主进程一半用来等低啊，，，，真正的任务子进程中执行

Queue队列的简单使用

from multiprocessing import Queue

q = Queue(3)    # 初始化一个Queue对象，最多可以put三条信息，如果不写3，那么久无限制

q.put("Message01")         # 添加信息的方法
q.put("Message02")
print(q.full())          # 查看 队列 是否满了的方法

q.put("Message03")
print(q.full())

# 因为队列已经满了，所以下面的消息会出现异常，第一个 try 会等待2秒后再抛出异常，
# 第二个 try 会立刻抛出异常
try:
    q.put("Message04", True, 2)
except:
    print("消息队列已满，现有消息数量：%s"%q.qsize())

try:
    q.put_nowait("Message04")
except:
    print("消息队列已满，现有消息数量：%s"%q.qsize())

# 推荐使用的方式，先判断队列是否已满，再写入
if not q.full():
    q.put_nowait("Message04")

# 读取消息的时候，先判断消息队列是否为空，再读取
if not q.empty():
    for i in range(q.qsize()):
        print(q.get_nowait())

Process配合Queue实现进程间通信

from multiprocessing import Process, Queue
import 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.start()
    pw.join()

    # 启动读取子进程，并等待结束
    pr.start()
    pr.join()

进程池与队列合作实现进程间通信

# 修改 import 中的 Queue 为 Manager
from multiprocessing import Manager, Pool
import os


def reader(q):
    print("reader启动(%s),父进程为(%s)"%(os.getpid(), os.getppid()))
    for i in range(q.qsize()):
        print("reader从Queue获取到消息：%s"%q.get(True))

def writer(q):
    print("writer启动(%s),父进程为(%s)"%(os.getpid(), os.getppid()))
    for i in "Always":
        q.put(i)


if __name__ == "__main__":
    print("(%s) start"%os.getpid())
    q = Manager().Queue()  # 使用Manager中的Queue来初始化
    po = Pool()
    # 使用阻塞模式创建进程，这样就不需要咋reader中使用死循环了，可以让writer完全执行后，再reader
    po.apply(writer, (q,))
    po.apply(reader, (q,))

    po.close()
    po.join()
    print("(%s) End" % os.getpid())

利用Pool进程池实现简单的文件复制

import os
import time
from multiprocessing import Pool

def copyFile(oldPath, newPath, fileName):
    print("%s 准备复制中。。。"%fileName)
    with open("%s\%s"%(oldPath, fileName), 'r') as fr, open("%s\%s"%(newPath, fileName), 'w') as fw:
        for line in fr:
            fw.write(line)
    time.sleep(1)


if __name__ == "__main__":
    oldPath = r"file"
    newPath = r"file-副本"

    os.mkdir(newPath)

    pool = Pool(5)

    fileList = os.listdir(oldPath)

    for fileName in fileList:
        pool.apply_async(copyFile, (oldPath, newPath, fileName))

    pool.close()

    pool.join()

    print("文件复制完成....")