进程间的通信IPC机制/生产者消费模型/线程初识(含线程互斥锁)

进程间的通信IPC机制

• 队列:先进先出
• 堆栈:先进后出

# 队列
from multiprocessing import Queue
​
q = Queue(3)  # 括号内可以传参数,表示的是这个队列的最大存储数为3
​
# 往队列中添加数据
q.put(1)
q.put(2)
print(q.empty())  # False,判断队列中的数据是否为空
print(q.full())  # 返回False,判断队列是否满了
q.put(3)
# q.put(4) 当队列已满时,再插入数据,不会报错,会原地等待,知道队列中出现数据可以取走(阻塞态)
print(q.full())  # 返回True,判断队列是否满了
print(q.empty())  # True,判断队列中的数据是否为空
print(q.get_nowait())  # 取值,没有值就报错
​
# 从队列中取数据
print(q.get())  # 返回1
print(q.get())  # 返回2
print(q.get())  # 返回3
print(q.get())  # 当队列中的数据被取完之后,再次获取时,程序会阻塞,直到队列中被放入值

注意:full,get_nowait,empty都不适用于多进程的情况

# 进程间的通信IPC机制

# 子进程放数据,主进程获取数据,两个子进程相互放,存取数据
from multiprocessing import Process,Queue
​
def producer(q):
    q.put('hello GF~')
    
def consumer(q):
    print(q.get())
​
if __name__ == '__main__':
    q = Queue()
    p = Process(target=producer,args=(q,))
    c = Process(target=consumer,args=(q,))
    p.start()
    c.start()
    # hello GF~

---

生产者消费模型

• 生产者:生产/制造数据的
• 消费者:消费/处理数据的

例子:做包子的与买包子的

• 买包子的远比卖包子的多,做包子的远比买包子的少

PS:供需不平衡

from multiprocessing import Process,Queue,JoinableQueue
import time
​
def producer(name,food,q):
    for i in range(10):
        data = '%s生产了%s'%(name,food)
        time.sleep(random.random())
        q.put(data)
        print(data)
    
def consumer(name,q):
    while True:
        data = q.get()
        if data == None: break
        print('%s吃了%s%s'%(name,data,i))
        time.sleep(random.random())
        q.task_done()  # 告诉队列你已经从队列中取出了一个数据,并且处理完毕了
​
if __name__ == '__main__':
    q = JoinableQueue()
    p = Process(target=producer,args=('大厨egon','馒头',q))
    p1 = Process(target=producer,args=('跟班tank','生蚝',q))
    c = Process(target=consumer,args=('某某某',q))
    c1 = Process(target=consumer,args=('某某',q))
    p.start()
    p1.start()
    c.daemon = True
    c1.daemon = True
    c.start()
    c1.start()
    p.join()
    p1.join()
    q.join()  # 等待队列中数据全部取出
    # q.put(None)
    # q.put(None)
    # 有几个消费者就写几个None

---

线程理论

什么是线程?

• 进程:资源单位
• 线程:执行单位

将内存比作工厂,那么进程就相当于是工厂内车间,而线程就相当于车间内的流水线

PS:每个进程都自带一个线程,线程才是真正的执行单位,进程只是在进程运行过程中提供代码运行所需要的资源

为什么要有线程?

• 开进程
  • 1.申请内存空间,耗资源
  • 2."拷贝代码" 耗费资源
• 开线程
  • 一个进程内可以有多个线程,并且线程与线程之间数据是共享的

PS:开启线程的开销要远远小于开启进程的开销

---

创建线程的两种方式

第一种:

from threading import Thread
import time
​
def task(name):
    print('%s is running'%name)
    time.sleep(3)
    print('%s is over'%name)
    
# 开线程不需要再__main__代码块内,但是习惯性的还是写在__main__代码块内
t = Thread(target=task,args=('egon',))
t.start()  # 告诉操作系统开辟一个线程,线程开销远小于进程
# 小的代码执行完,线程就已经开启了
print('主')
'''
egon is running
主
(三秒后出现)
egon is over
'''

第二种:

from threading import Thread
import time
​
class MyThread(Thread):
    def __init__(self,name):
        super().__init__()
        self.name = name
    def run(self):
        print('%s is running'%self.name)
        time.sleep(3)
        print('%s is over'%self.name)
​
t = MyThread('egon')
t.start()
print('主')
'''
egon is running
主
(三秒后出现)
egon is over
'''

---

线程对象及其他方法

同一个进程内可以有多个线程:

from threading import Thread,current_thread,active_count
import time
import os
​
def task(name):
    print('%s is running'%name)
    print('子current_thread:',current_thread().name)
    print('子',os.getpid())
    time.sleep(3)
    print('%s is over'%name)
​
t = Thread(target=task,args=('egon',))
t1 = Thread(target=task,args=('jason',))
t.start()
t1.start()
print('主')
print('主current_thread:',current_thread().name)
print('主',os.getpid())
'''
egon is running
子current_thread: Thread-1
子 3404
jason is running
主
主current_thread: Thread-2
主current_thread: MainThread
子 3404
主 3404
egon is over
jason is over
'''

记录一个进程内的当前正在活跃的线程数(当t.join()时):

from threading import Thread,current_thread,active_count
import time
import os
​
def task(name,i):
    print('%s is running'%name)
    time.sleep(i)
    print('%s is over'%name)
​
t = Thread(target=task,args=('egon',1))
t1 = Thread(target=task,args=('egon',2))
t.start()
t1.start()
t.join()  # 主线程等待子线程运行完毕
print('当前正在活跃的线程数:',active_count())
print('主')
'''
egon is running
jason is running
egon is over
当前正在活跃的线程数:2-----其中主线程也算一个
主
jason is over
'''

记录一个进程内的当前正在活跃的线程数(当t1.join()时):

from threading import Thread,current_thread,active_count
import time
import os
​
def task(name,i):
    print('%s is running'%name)
    time.sleep(i)
    print('%s is over'%name)
​
t = Thread(target=task,args=('egon',1))
t1 = Thread(target=task,args=('egon',2))
t.start()
t1.start()
t1.join()  # 主线程等待子线程运行完毕
print('当前正在活跃的线程数:',active_count())
print('主')
'''
egon is running
jason is running
egon is over
jason is over
当前正在活跃的线程数:1------只剩下主线程了
主
'''

---

守护线程

未设置守护线程前的情况:

from threading import Thread,current_thread
import time
​
def task(i):
    print(curren_thread().name)
    time.sleep(i)
    print('GG')
​
for i in range(3):
    t = Thread(target=task,args=(i,))
    t,start()
​
print('主')
'''
Thread-1
GG
Thread-2
Thread-3
主
GG
GG
'''
# 主线程运行结束之后需要等待子线程结束后才能结束

设置守护线程后的情况:

from threading import Thread,current_thread
import time
​
def task(i):
    print(current_thread().name)
    time.sleep(i)
    print('GG')
​
t = Thread(target=task,args=(1,))
t.daemon = True
t.start()
print('主')
'''
Thread-1
主
'''

总结:

• 主线程的结束就意味着进程的结束
• 主线程必须等待其他非守护线程子线程的结束才能结束(销毁内存空间)
• 意味着子线程在运行的时候需要使用进程中的资源,而主线程一旦结束了资源也就销毁了

---

关于深入理解守护线程的案例:

from threading import Thread
from multiprocessing import Process
import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")
​
def bar():
    print(456)
    time.sleep(3)
    print("end456")
​
if __name__ == '__main__':
    t1=Thread(target=foo)
    t2=Thread(target=bar)
    t1.daemon=True
    t1.start()
    t2.start()
    print("main-------")
'''
(瞬间出现)
123
456
main-------
(一秒后出现)
end123
(两秒后出现,合计三秒出现)
end456
'''

分析:

• 单纯的主线程结束后并未全部结束,而主线程必须等待所有非守护线程的结束才能结束
• 案例中主线程在等待bar函数的结束,而bar要结束需要3s,且bar函数对应的t2线程并非守护线程,这时候哪怕t1线程为守护线程,且对应的foo函数早已将1s走完,整个主线程也未结束

---

线程间通信

from threading import Thread
​
money = 666
​
def task():
    global money
    money = 999
​
t = Thread(target=task)
t.start()
t.join()
print(money)  # 999

---

线程互斥锁

from threading import Thread,Lock
import time
​
n = 100
​
def task(mutex):
    global n
    mutex.acquire()
    tmp = n
    time.sleep(0.1)
    n = tmp - 1
    mutex.release()
​
t_list = []
mutex = Lock()
for i in range(100):
    t = Thread(target=task,args=(mutex,))
    t,start()
    t_list.append(t)
for t in t_list:
    t.join()
print(n)  # 0