python之路 -- 并发编程之进程

进程的特征：

--动态性：进程的实质是程序在多道程序系统中的一次执行过程，进程是动态产生，动态消亡的。

--并发性：任何进程都可以同其他进程一起并发执行

--独立性：进程是一个能独立运行的基本单位，同时也是系统分配资源的独立单位；

--异步性：由于进程间的相互制约，使进程具有执行的间断性，即进程按各自独立的、不可预知的速度向前推进

--结构特征：进程由程序、数据和进程控制块三部分组成。

--多个不同的进程可以包含相同的程序：一个程序在不同的数据集里就构成不同的进程，能得到不同的结果；但是执行过程中，程序不能发生改变。

进程和程序的区别：

--程序是指令和数据的有序集合，其本身没有任何运行的含义，是一个静态的概念。

--而进程是程序在处理机上的一次执行过程，它是一个动态的概念。

--程序可以作为一种软件资料长期存在，而进程是有一定生命期的。

--程序是永久的，进程是暂时的。

注意：同一个程序执行两次，就会在操作系统中出现两个进程，所以我们可以同时运行一个软件，分别做不同的事情也不会混乱。

在python程序中的进程操作:

from multiprocessing import Process
import os
import time

def func(arg1):
    print("开启子进程的函数")
    print("主进程的子进程：",os.getpid())  # 查看当前进程的进程号
    print("子进程的父进程：", os.getppid())
    time.sleep(2)
    print("%s是个pig"%arg1)

if __name__ == '__main__':
    p = Process(target=func,args=("yl",))    
    #创建一个进程,传入参数的时候必须传元组的形式
    # p是一个进程对象,还没有启动进程
    p.start()   #开启了一个子进程
    print('主进程 :',os.getpid()) 
    # 此处的进程号等于func函数中的子进程的父进程的进程号
    print('主进程的父进程 :',os.getppid()) # 查看当前进程的父进程
    print("我是主进程")

# 输出结果：
主进程 : 80268
主进程的父进程 : 50360
我是主进程
开启子进程的函数
主进程的子进程： 77768
子进程的父进程： 80268
yl是个pig

# 开启了子进程的主进程 :
    # 主进程自己的代码如果长,等待自己的代码执行结束,
    # 子进程的执行时间长,主进程会在主进程代码执行完毕之后等待子进程执行完毕之后 主进程才结束

3.p.join()的用法

import time
from multiprocessing import Process

def func(arg1,arg2):
    print('*'*arg1)
    time.sleep(2)
    print('*'*arg2)

if __name__ == '__main__':
    p = Process(target=func,args=(10,20))
    p.start()
    print('hahahaha')
    p.join()     # 是感知一个子进程的结束,将异步的程序改为同步
    # 相当于主进程在此处被阻塞，等待子进程运行结束之后再运行此处后面的代码
    print('====== : 运行完了')

# 输出结果：
hahahaha
**********
********************
====== : 运行完了

4.多进程之间的数据隔离问题

import os
from multiprocessing import Process

def func():
    global n   # 声明了一个全局变量
    n = 0       # 重新定义了一个n
    print('pid : %s'%os.getpid(),n)

if __name__ == '__main__':
    n = 100
    p = Process(target=func)
    p.start()
    p.join()
    print(os.getpid(),n)    
    #此处的n仍然为，主进程的n，未受到子进程创建的全局变量的影响

# 执行结果：
# pid : 884 0
# 11932 100

#多进程之间的数据是隔离的，互不影响

 5.使用面向对象中类的知识来创建进程

# 可以传入参数的
from multiprocessing import Process

class MyProcess(Process):   #继承Process类
    def __init__(self,arg1,arg2):
        super().__init__()  # 调用父类的__init__方法（必须写，否则报错）
        self.arg1 = arg1
        self.arg2 = arg2

    # 必须有run方法，子进程执行的就是run方法中的代码
    def run(self):
        print(self.pid)     #打印进程的进程号
        print(self.name)    #打印进程的进程名，等同于os.getpid()
        print(self.arg1)
        print(self.arg2)

if __name__ == '__main__':
    p1 = MyProcess(1,2)
    p1.start()
    p2 = MyProcess(3,4)
    p2.start()
使用面向对象中类的知识来创建进程

 6.开启多个子进程

from multiprocessing import Process
import time
import os

def func(arg1,arg2):
    print('*'*arg1)
    time.sleep(2)
    print('*'*arg2)

if __name__ == '__main__':
    p = Process(target=func,args=(5,10))
    p.start()
    p1 = Process(target=func,args=(5,10))
    p1.start()
    p2 = Process(target=func,args=(5,10))
    p2.start()
    p3 = Process(target=func,args=(5,10))
    p3.start()

# 开启了4个子进程，4个子进程同时运行,运行顺序不能确定先后顺序
#
# if __name__ == '__main__':
#     for i in range(10):
#         p = Process(target=func, args=(5*i, 10*i))
#         p.start()
# # 开启的多个子进程的运行顺序没有先后（但是是按照顺序给操作系统发的指令开启进程）
#         # p.join()    # 多个进程变成同步，执行效率低
#
#     # p.join()    
# 此处的p相当于p9，总是在i等于9时开的之进程执行完成之后执行 print('执行完了'),
# 因为所开的子进程不按顺序执行，所以不能保证 print('执行完了') 最后执行
#     print('执行完了')


# if __name__ == '__main__':
#     p_lst = []
#     for i in range(10):
#         p = Process(target=func,args=(5*i,10*i))
#         p_lst.append(p)
#         p.start()
#
#     [p.join() for p in p_lst]   
# （列表推导式）之前的所有进程必须在这里都执行完才能执行下面的代码
#     print('执行完了')
# 这种方法可以保证 print('执行完了')，是最后执行。

7.守护进程

# 子进程 -- > 守护进程
# 主进程代码执行结束守护进程立即结束
import time
from multiprocessing import Process

def func():
    while True:
        time.sleep(0.2)
        print('我还活着')

def func2():
    print('in func2 start')
    time.sleep(3)
    print('in func2 finished')

if __name__ == '__main__':
    p = Process(target=func)
    p.daemon = True   # 设置子进程为守护进程，必须在start前设置
    p.start()
    p2 = Process(target=func2)
    p2.start()
    p2.terminate()     # 结束一个子进程，不会立即关闭,所以is_alive立刻查看的结果可能还是存活
    time.sleep(1)。
    print(p2.is_alive())  # 检验一个进程是否还活着
    # time.sleep(1)
    print(p2.name)
    # i = 0
    # while i<5:
    #     print('我是socket server')
    #     time.sleep(1)
    #     i+=1

# 输出的结果：
我还活着
我还活着
我还活着
False
Process-2
    
# 守护进程会随着主进程的代码执行完毕 而 结束
# 在主进程内结束一个子进程 p.terminate()
    #  结束一个进程不是在执行方法之后立即生效,需要一个操作系统响应的过程
# 检验一个进程是否活着的状态 p.is_alive()
# p.name p.pid 这个进程的名字和进程号

import os
import time
from multiprocessing import Process

class Myprocess(Process):
    def __init__(self,person):
        super().__init__()
        self.person = person
    def run(self):
        print(os.getpid(),self.name)
        print('%s正在和女主播聊天' %self.person)

if __name__ == '__main__':
    p=Myprocess('哪吒')
    p.daemon=True 
    #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
    p.start()
    time.sleep(1) # 在sleep时查看进程id对应的进程
    print('主')

# 输出的结果：
92480 Myprocess-1
哪吒正在和女主播聊天
主

8.进程锁

import json
import time
from multiprocessing import Process,Lock
# from multiprocessing import Lock

def show(i):
    with open('ticket') as f:
        dic = json.load(f)
    print('余票: %s'%dic['ticket'])

def buy_ticket(i,lock):
    lock.acquire() #拿钥匙进门
    with open('ticket') as f:
        dic = json.load(f)
        time.sleep(0.1)
    if dic['ticket'] > 0 :
        dic['ticket'] -= 1
        print('33[32m%s买到票了33[0m'%i)
    else:
        print('33[31m%s没买到票33[0m'%i)
    time.sleep(0.1)
    with open('ticket','w') as f:
        json.dump(dic,f)
    lock.release()      # 还钥匙

if __name__ == '__main__':
    for i in range(10):
        p = Process(target=show,args=(i,))
        p.start()
    lock = Lock()   # 加锁
    for i in range(10):
        p = Process(target=buy_ticket, args=(i,lock))   # 将lock作为一个参数传入
        p.start()

# 输出结果：
余票: 2
余票: 2
余票: 2
余票: 2
余票: 2
0买到票了
2买到票了
1没买到票
3没买到票
4没买到票

9.信号量

# ktv小实例
# 一个KTV只能同时容纳4个人，当不满4人时可以进入。当有人唱完歌出来后可以有人再进去

from multiprocessing import Process
import random,time
from multiprocessing import Semaphore

def ktv(i,sem):
    sem.acquire()   #每个人进入KTV时获取钥匙
    print("%s 走进来了"%i)
    # 此处采用random和time模块模拟唱歌所花费的时间(3-8秒)
    time.sleep(random.randint(3,8))
    print('%s唱完歌，走出来了'%i)
    sem.release()   #归还钥匙

# 开启20个进程模拟20个人进入KTV
if __name__ == "__main__":
    # 设置信号量，相当于设置了KTV门有4把钥匙
    sem_number = Semaphore(4)
    # 利用for循环设置20个进程
    for i in range(20):
        p = Process(target = ktv,args=(i,sem_number))   
        # 将实例化的信号量当做一个参数传入
        p.start()

10.事件

# 一个信号可以使所有的进程都进入阻塞状态
# 也可以控制所有的进程解除阻塞
# 一个事件被创建之后,默认是阻塞状态
from multiprocessing import Event
e = Event()  # 创建了一个事件
print(e.is_set())   # 查看一个事件的状态,默认被设置成阻塞
e.set()      # 将这个事件的状态改为True
print(e.is_set())
e.wait()     # 是依据e.is_set()的值来决定是否阻塞的
print(123456)
e.clear()    # 将这个事件的状态改为False
print(e.is_set())
e.wait()     # 等待 事件的信号被变成True
print('*'*10)

# 红绿灯事件练习

from multiprocessing import Process
import random,time
from multiprocessing import Event

def cars(e,car_number):
    if not e.is_set():
        print('car%i在等待' % car_number)
        e.wait()  # 阻塞 直到得到一个 事件状态变成 True 的信号

    print('33[0;32;40mcar%i通过33[0m' % car_number)

def light(e):
    while True:
        if e.is_set():
            e.clear()
            print("33[31m红灯亮了33[0m")
        else:
            e.set()
            print("33[32m绿灯亮了33[0m")

        time.sleep(2)

if __name__ == "__main__":
    e = Event()
    traffic_light = Process(target=light,args=(e,))
    traffic_light.start()
    # 用20个进程模拟将要路过红绿灯的20辆车
    for i in range(20):
        car = Process(target=cars,args=(e,i))
        car.start()
        time.sleep(random.randint(0,1))

 11.队列

# 队列：先进先出

from multiprocessing import Queue

q = Queue(5) # 创建一个容量为5的队列
q.put('a')        # 在队列中放值，如果队列满了就发生阻塞（即不往下执行）处于等待状态，直到队列中放入了值才继续执行
print(q.get())  # 从队列中取值，如果队列空了就发生阻塞（即不往下执行）处于等待状态，直到队列中有值被取走才继续执行
print(q.empty())    # 判断队列是否为空
print(q.full())     # 判断队列是否是满的

print(q.get_nowait())   #从队列中取值，当队列中没有值的时候会报错
q.put_nowait()        # 把值放入队列中，当队列满的时候会报错

from multiprocessing import Queue,Process
def produce(q):
    q.put('hello')

def consume(q):
    print(q.get())

if __name__ == '__main__':
    q = Queue()
    p = Process(target=produce,args=(q,))
    p.start()
    c = Process(target=consume, args=(q,))
    c.start()

# 生产者消费者模型
from multiprocessing import Process
from multiprocessing import Queue
import time,random

def produce(name,what,queue):
    for i  in range(5):
        time.sleep(random.randint(1,3))
        w = "%s生产了%s,%s"%(name,what,i)
        print(w)
        queue.put(w)

def consuer(q,name):
    while True:
        food = q.get()
        if food is None:
            print("%s消费完了，得到了一个空"%name)
            break
        print('33[31m%s消费了%s33[0m'%(name,food))
        time.sleep(random.randint(1,3))

if __name__ == "__main__":
    q = Queue(20)
    p1 = Process(target=produce,args=("alex","包子",q))
    p1.start()
    p2 = Process(target=produce, args=("武sir", "烧饼", q))
    p2.start()
    c1 = Process(target=consuer,args=(q,"dog"))
    c1.start()
    c2 = Process(target=consuer,args=(q,"cat"))
    c2.start()
    p1.join()
    p2.join()
    q.put(None)
    q.put(None)
#  在消费者这一端:
    # 每次获取一个数据
    # 处理一个数据
    # 发送一个记号 : 标志一个数据被处理成功

# 在生产者这一端:
    # 每一次生产一个数据,
    # 且每一次生产的数据都放在队列中
    # 在队列中刻上一个记号
    # 当生产者全部生产完毕之后,
    # join信号 : 已经停止生产数据了
                # 且要等待之前被刻上的记号都被消费完
                # 当数据都被处理完时,join阻塞结束

# consumer 中把所有的任务消耗完
# producer 端 的 join感知到,停止阻塞
# 所有的producer进程结束
# 主进程中的p.join结束
# 主进程中代码结束
# 守护进程(消费者的进程)结束

 12.管道

# 管道基础
from multiprocessing import Pipe

conn1,conn2 = Pipe()
conn1.send("吃了么？")  # 不需要转换成bytes类型
print(conn2.recv())

# 输出结果：吃了么？



# 管道结合进程使用
from multiprocessing import Pipe,Process

def func(conn):
    conn.send('hello, world!')

if __name__ == '__main__':
    conn1,conn2 = Pipe()
    p = Process(target=func,args=(conn1,))
    p.start()
    print(conn2.recv())

# 处理多进程发送消息结束之后的阻塞方法一

from multiprocessing import Process,Pipe

def func(conn):
    while True:
        mes = conn.recv()
        if mes is None:
            break
        else:
            print(mes)

if __name__ == "__main__":
    conn1,conn2 = Pipe()
    p = Process(target=func,args = (conn2,))
    p.start()
    for i in range(20):
        conn1.send('Hello World！')
    conn1.send(None)

# 处理多进程发送消息结束之后的阻塞方法二
from multiprocessing import Pipe,Process

def func(conn1,conn2):
    conn2.close()
    while True:
        try :
            msg = conn1.recv()
            print(msg)
        except EOFError:
            conn1.close()
            break

if __name__ == '__main__':
    conn1, conn2 = Pipe()
    Process(target=func,args = (conn1,conn2)).start()
    conn1.close()
    for i in range(20):
        conn2.send('吃了么')
    conn2.close()

# 只有当管道都关闭之后才会引发EOFError异常

# 管道实现生产者消费者模型,会出现'数据安全问题'
from multiprocessing import Lock,Pipe,Process
def producer(con,pro,name,food):
    con.close()
    for i in range(100):
        f = '%s生产%s%s'%(name,food,i)
        print(f)
        pro.send(f)
    pro.send(None)
    pro.send(None)
    pro.send(None)
    pro.close()

def consumer(con,pro,name,lock):
    pro.close()
    while True:
            lock.acquire()
            food = con.recv()
            lock.release()
            if food is None:
                con.close()
                break
            print('%s吃了%s' % (name, food))
if __name__ == '__main__':
    con,pro = Pipe()
    lock= Lock()
    p = Process(target=producer,args=(con,pro,'egon','泔水'))
    c1 = Process(target=consumer, args=(con, pro, 'alex',lock))
    c2 = Process(target=consumer, args=(con, pro, 'bossjin',lock))
    c3 = Process(target=consumer, args=(con, pro, 'wusir',lock))
    c1.start()
    c2.start()
    c3.start()
    p.start()
    con.close()
    pro.close()

# pipe 数据不安全性
# 通过加锁来控制操作管道的行为 来避免进程之间争抢数据造成的数据不安全现象
# 队列 进程之间数据安全的
# 队列的底层就是 管道 + 锁 实现的

# 加锁之后不会出现数据安全问题
from multiprocessing import Process,Pipe,Lock
def consumer(produce, consume,name,lock):
    produce.close()
    while True:
        lock.acquire()
        baozi=consume.recv()
        lock.release()
        if baozi:
            print('%s 收到包子:%s' %(name,baozi))
        else:
            consume.close()
            break

def producer(produce, consume,n):
    consume.close()
    for i in range(n):
        produce.send(i)
    produce.send(None)
    produce.send(None)
    produce.close()

if __name__ == '__main__':
    produce,consume=Pipe()
    lock = Lock()
    c1=Process(target=consumer,args=(produce,consume,'c1',lock))
    c2=Process(target=consumer,args=(produce,consume,'c2',lock))
    p1=Process(target=producer,args=(produce,consume,30))
    c1.start()
    c2.start()
    p1.start()
    produce.close()
    consume.close()

 13.进程之间的数据共享

from multiprocessing import Manager,Process

def main(dic):
    dic['count'] -= 1
    print(dic)

if __name__ == '__main__':
    m = Manager()
    dic=m.dict({'count':100})
    p_lst = []
    p = Process(target=main, args=(dic,))
    p.start()
    p.join()

# 输出结果：
# {'count': 99}

from multiprocessing import Manager,Process,Lock
def main(dic,lock):
    dic['count'] -= 1

if __name__ == '__main__':
    m = Manager()
    l = Lock()
    dic=m.dict({'count':100})
    p_lst = []
    for i in range(50):
        p = Process(target=main,args=(dic,l))
        p.start()
        p_lst.append(p)
    for i in p_lst: i.join()
    print('主进程',dic)
    
# 输出结果：
# 主进程 {'count': 50}

 14.进程池

import os
import time
from multiprocessing import Pool
def func(n):
    print('start func%s'%n,os.getpid())
    time.sleep(1)
    print('end func%s'%n,os.getpid())

if __name__ == '__main__':
    p = Pool(5)     # 一般参数为CPU的核数加一
    for i in range(10):
        p.apply(func,args=(i,))   # 同步创建进程，效率很低
        p.apply_async(func,args=(i,))   # 异步创建进程，显著提高效率
    p.close()  # 结束进程池接收任务
    p.join()   # 感知进程池中的任务执行结束

# p.map(funcname,iterable)     默认异步的执行任务,且自带close和join
# p.apply   同步调用的
# p.apply_async 异步调用 和主进程完全异步 需要手动close 和 join

# 进程池的返回值
from multiprocessing import Pool
def func(i):
    return i*i

if __name__ == '__main__':
    p = Pool(5)
    for i in range(10):
        res = p.apply(func,args=(i,))   # apply的结果就是func的返回值
        print(res)

        
import time
from multiprocessing import Pool
def func(i):
    time.sleep(0.5)
    return i*i

if __name__ == '__main__':
    p = Pool(5)
    res_l = []
    for i in range(10):
        res = p.apply_async(func,args=(i,))   
        # apply的结果就是func的返回值
        res_l.append(res)
    for res in res_l:print(res.get())
    # 等着 func的return结果
    
# 返回值 : apply_async返回的对象obj
        #          为了用户能从中获取func的返回值obj.get()
        # get会阻塞直到对应的func执行完毕拿到结果

# 进程池的回调函数
import os
from multiprocessing import Pool
def func1(n):
    print('in func1',os.getpid())
    return n*n

def func2(nn):
    print('in func2',os.getpid())
    print(nn)

if __name__ == '__main__':
    print('主进程 :',os.getpid())
    p = Pool(5)
    for i in range(10):
        p.apply_async(func1,args=(10,),callback=func2)
    # callback回调函数，先执行进程1调用func1，然后将func1的返回值传给func2
    p.close()
    p.join()
    

# 进程池回调函数的爬虫实例
#进程池的回调函数适合于爬虫的处理，用回调函数处理获取的网页，能够显著提高爬虫的效率

import requests
from multiprocessing import Pool

def get(url):
    response = requests.get(url[0])
    if response.status_code == 200:
        return url,response.content.decode('utf-8')

def deal_get(args):
    # url,content = args
    url = args[0]
    content = args[1]
    print(url,len(content))

if __name__ == "__main__":
    p = Pool(5)
    url_list = [
        'http://www.baidu.com',
        "http://www.sohu.com/",
        "https://www.cnblogs.com/",
        "https://www.sogou.com/",
    ]
    for i in url_list:
        p.apply_async(get,args=(i,),callback=deal_get)
    p.close()
    p.join()