进程（Process）

进程是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。在早期面向进程设计的计算机结构中，进程是程序的基本执行实体；在当代面向线程设计的计算机结构中，进程是线程的容器。程序是指令、数据及其组织形式的描述，进程是程序的实体。

创建子进程

1. Linux创建子进程的原理:

1). 父进程和子进程, 如果父进程结束， 子进程也随之结束;
2). 先有父进程， 再有子进程， 通过fork函数实现;

2. fork函数的返回值:调用该方法一次， 返回两次;

• 产生的子进程返回一个0
• 父进程返回子进程的pid;

3. Window也能使用fork函数么?

Windows没有fork函数， Mac有fork函数(Unix -> Linux, Unix-> Mac),
封装了一个模块multiprocessing

4. 常用方法:

• os.fork()
• os.getpid(): 获取当前进程的pid;
• os.getppid(): parent process id, 获取当前进程的父进程的id号；

import  multiprocessing

import  os
import  time
print("当前进程(pid=%d)正在运行..." %(os.getpid()))
print("当前进程的父进程(pid=%d)正在运行..." %(os.getppid()))
print("正在创建子进程......")
pid = os.fork()
pid2 = os.fork()
print("第1个:", pid)
print("第2个: ", pid2)

if pid == 0:
    print("这是创建的子进程， 子进程的id为%s, 父进程的id为%s"
          %(os.getpid(), os.getppid()))
else:
    print("当前是父进程[%s]的返回值%s" %(os.getpid(), pid))
time.sleep(100)

通过实例化Process对象创建多进程

import  multiprocessing
import  threading
def job():
    print("当前子进程的名称为%s" %(multiprocessing.current_process()))

# 创建一个进程对象
p1 = multiprocessing.Process(target=job)
# 运行多进程， 执行任务
p1.start()

# 创建一个进程对象
p2 = multiprocessing.Process(target=job)
# 运行多进程， 执行任务
p2.start()

# 等待所有的子进程执行结束， 再执行主进程的内容
p1.join()
p2.join()
print("任务执行结束.......")

通过继承的方式创建子进程

import  multiprocessing
class JobProcess(multiprocessing.Process):
    # 重写Process的构造方法， 获取新的属性
    def __init__(self, queue):
        super(JobProcess, self).__init__()
        self.queue = queue

    # 重写run方法， 将执行的任务放在里面即可
    def run(self):
        print("当前子进程的名称为%s" % (multiprocessing.current_process()))

processes = []
# 启动10个子进程， 来处理需要执行的任务;
for i in range(10):
    # 实例化对象；
    p = JobProcess(queue=3)
    processes.append(p)
    # 启动多进程， 处理需要执行的任务;
    p.start()

# 等待所有的子进程执行结束， 再继续执行主进程
[process.join() for process in processes]

# 执行主进程
print("任务执行结束.......")

多进程案例

import  multiprocessing
import threading
from mytimeit import timeit
class JobProcess(multiprocessing.Process):
    # 重写Process的构造方法， 获取新的属性
    def __init__(self, li):
        super(JobProcess, self).__init__()
        self.li = li
    # 重写run方法， 将执行的任务放在里面即可
    def run(self):
        for item in self.li:
            sum(item)
        # print(multiprocessing.current_process())
class JobThread(threading.Thread):
    def __init__(self, li):
        super(JobThread, self).__init__()
        self.li = li
    def run(self):
        sum(self.li)
        # print(threading.current_thread())

@timeit
def use_multi_process():
    # 1). 开启的进程数有瓶颈， 取决于CPU的个数;
    # 2). 如果处理的数据比较小， 不建议使用多进程， 因为进程的创建和销毁需要时间，
    #     开启的进程数越多，不一定效率越高;;
    # 3). 如果处理的数据量足够大， 0<开启的进程数<cpu个数, 开启的进程数越多， 效率越高;
    lists = [[1, 2, 3, 4, 5], [2, 3, 4, 5, 6],
          [2, 3, 4, 5, 6, 7, 8], [2, 3, 4, 5, 6]] *1000
    processes = [])

    # 对于处理的数据进行分片， 每个进程处理一部分任务;
    for i in range(0,len(lists), 1000):
        # print(i, i+1000)
        p = JobProcess(lists[i:i+1000])
        processes.append(p)
        p.start()
    # 等待所有的子进程执行结束
    [process.join()for process in processes]
    print("多进程执行结束.......")

@timeit
def use_multi_thread():
    lists = [[1, 2, 3, 4, 5], [2, 3, 4, 5, 6],
          [2, 3, 4, 5, 6, 7, 8], [2, 3, 4, 5, 6]] *1000
    processes = []
    for li in lists:
        p = JobThread(li)
        processes.append(p)
        p.start()
    # 等待所有的子进程执行结束
    [process.join()for process in processes]
    print("多线程执行结束.......")

if __name__ == "__main__":
    use_multi_process()
    use_multi_thread()

守护进程

import  multiprocessing
import time
def job():
    name = multiprocessing.current_process()
    print("开始运行")
    time.sleep(3)
    print("结束进程")
if __name__ == '__main__':
    # 启动一个子进程
    p1 = multiprocessing.Process(target=job, name='use deamon')
    # True/False
    p1.daemon = False
    p1.start()


    # join等待所有子进程执行结束， 再执行主进程
    p1.join(1)
    # 主进程执行
    print("程序执行结束")

终止进程

import time
import  multiprocessing

def job():
    print("start.....")
    time.sleep(1)
    print('end.......')
if __name__ == '__main__':
    p = multiprocessing.Process(target=job)
    print("Before:", p.is_alive())
    p.start()  # 启动子进程
    print("During:", p.is_alive())
    p.terminate()   # 终止子进程
    print('terminate:', p.is_alive())
    p.join()        #等待子进程彻底终止
    print("joined:", p.is_alive())

进程间通信

import  multiprocessing
from multiprocessing import Queue
import time

class Producer(multiprocessing.Process):
    # 往队列里面写内容
    def __init__(self, queue):
        super(Producer, self).__init__()
        self.queue = queue

    def run(self):
        for i in range(10):
            self.queue.put(i)
            time.sleep(0.1)
            print("传递消息， 内容为:%s" %(i))

class Consumer(multiprocessing.Process):
    # 读取队列里面的内容
    def __init__(self, queue):
        super(Consumer, self).__init__()
        self.queue = queue

    def run(self):
        # 判断队列是否为空， 如果是， 跳出循环， 不会再去从队列获取数据;
        # while not self.queue.empty():
        while True:
            time.sleep(0.1)
            print("读取进程传递的消息:%s" %(self.queue.get()))


if __name__ == "__main__":
    q = Queue()
    p1 = Producer(q)
    c1 = Consumer(q)

    p1.start()
    c1.start()


    p1.join()
    c1.terminate()
    c1.join()
    print('all done')

分布式进程

why 分布式进程?

任务需要处理的数据特别大, 希望多台主机共同处理任务;

how 实现分布式进程?

multiprocessing.managers子模块里面可以实现将进程分布到多台机器上;

Master: 管理端， 分配任务给其他主机;
Worker1: 被管理端， 处理master给予的任务;
Worker2:被管理端， 处理master给予的任务;

特别重要的类:

BaseManager： 提供了不同机器进程之间共享数据的一种方法;
(重要的点: ip:port)

Master

import random
from queue import Queue
from multiprocessing.managers import  BaseManager
# 1. 创建需要的队列
# task_queue存储的是任务需要传递的参数
task_queue = Queue()
# result_queue存储的是任务执行结果
result_queue = Queue()
# 2. 将队列注册到网络上
# 需要将两个队列注册到网络上， 使得其他主机可以访问;
BaseManager.register('get_task_queue',  callable=lambda : task_queue)
BaseManager.register('get_result_queue',  callable=lambda : result_queue)

# 绑定端口为为4000， 暗号/密钥为westos;
# 172.25.254.250  172.25.0.250
manager = BaseManager(address=('', 4000), authkey=b'westos')

# 3. 启动manager, 开始共享队列;
manager.start()

# 4. 通过网络访问共享的Queue对象,
# manager.register: 注册一个队列, 唯一标识'get_task_queue'
# manager.get_task_queue()调用注册， 调用过程中执行的内容为callable只想的函数;
task = manager.get_task_queue()
result =  manager.get_result_queue()
# 5. 开始往队列里面放执行任务需要的数据;
for i in range(100):
    n = random.randint(1,1000)
    task.put(n)
    print("任务列表加入任务: %d" %(n))

# 6. 从result队列里面读取各个及机器执行的结果;
for j in range(100):
    res = result.get()
    print("队列任务的执行结果:%s" %(res))

# 7. 关闭manager, 取消共享队列;
manager.shutdown()

Worker1

from multiprocessing.managers import BaseManager

# 1. 连接Master端， 获取共享的队列;
# address写的是master端的ip和共享的端口, authkey与master端保持一致;
import time

worker = BaseManager(address=('172.25.254.250', 4000), authkey=b'westos')

# 2. 注册队列， 获取共享端的队列内容
BaseManager.register('get_task_queue')
BaseManager.register('get_result_queue')

# 3. 去连接
worker.connect()

# 4. 通过网络访问共享的Queue对象,
# manager.register: 注册一个队列, 唯一标识'get_task_queue'
# manager.get_task_queue()调用注册， 调用过程中执行的内容为callable只想的函数;
task = worker.get_task_queue()
result =  worker.get_result_queue()


# 5. 读取管理端共享的任务， 依次执行;
for i in range(50):
    n = task.get()
    print("运行任务 %d ** 2:" %(n))
    res = "%d ** 2 = %d" %(n, n**2)
    time.sleep(1)
    # 将运行结果放入reesult队列
    result.put(res)

print("执行结束.....")

Worker2

from multiprocessing.managers import BaseManager

# 1. 连接Master端， 获取共享的队列;
# address写的是master端的ip和共享的端口, authkey与master端保持一致;
import time

worker = BaseManager(address=('172.25.254.250', 4000), authkey=b'westos')


# 2. 注册队列， 获取共享端的队列内容
BaseManager.register('get_task_queue')
BaseManager.register('get_result_queue')

# 3. 去连接
worker.connect()


# 4. 通过网络访问共享的Queue对象,
# manager.register: 注册一个队列, 唯一标识'get_task_queue'
# manager.get_task_queue()调用注册， 调用过程中执行的内容为callable只想的函数;
task = worker.get_task_queue()
result =  worker.get_result_queue()


# 5. 读取管理端共享的任务， 依次执行;
for i in range(50):
    try:
        n = task.get()
        print("运行任务 %d ** 2:" %(n))
        time.sleep(1)
        res = "%d ** 2 = %d" %(n, n**2)
        # 将运行结果放入reesult队列
        result.put(res)
    except Exception as e:
        print("任务队列为空")

print("执行结束.....")

进程池

why进程池?
- 如果启动大量子进程， 会消耗时间用来创建和销毁子进程,
- 使用进程池， 不需要启动大量的进程;

import  multiprocessing
import time
def job(id):
    print('start %d.....' %(id))
    print('end %d.....' %(id))
# 创建一个进程池对象
pool = multiprocessing.Pool(10)
# 给进程池的进程分配任务
for i in range(10):
    pool.apply_async(job,args=(i, ) )
pool.close()
# 等待所有的子进程执行结束
pool.join()
print('success')

进程池之ProcessPoolExecutor

from concurrent.futures import  ThreadPoolExecutor
from concurrent.futures import  ProcessPoolExecutor
def job(id):
    print('start %d.....' %(id))
    print('end %d.....' %(id))
    return  id
# 第2种方式: submit
# pool = ProcessPoolExecutor(max_workers=4)
# # 分配任务给子进程， 并且返回一个Future对象;
# f1 = pool.submit(job, 1)
# # 获取进程是否执行结束；
# f1.done()
# # 获取子进程执行的结果
# f1.result()

# 第三种方式:
pool = ProcessPoolExecutor(max_workers=4)
for res in pool.map(job, range(1,100)):
    print(res)