网络编程生产者消费者模型 GiL

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

守护线程

所谓守护线程，是指在程序运行的时候在后台提供一种通用服务的线程，比如垃圾回收线程就是一个很称职的守护者，并且这种线程并不属于程序中不可或缺的部分。因此，当所有的非守护线程结束时，程序也就终止了，同时会杀死进程中的所有守护线程。反过来说，只要任何非守护线程还在运行，程序就不会终止。

from threading import Thread
import time ,os,random
def task(args):
print("%s is runnning%s"%(args,os.getpid()))
time.sleep(random.randint(1,3))
print("%s is done %s"%(args,os.getpid()))
if __name__ == '__main__':
t1=Thread(target=task,args=("t1",))
t2=Thread(target=task,args=("t2",))
t1.daemon=True
t1.start()
t2.start()
print("主")

结果：

t1 is runnning48492
t2 is runnning48492
主
t2 is done 48492
t1 is done 48492

互斥锁：当一个全局的共有资源，被多个线程同时进行调用会发生到意想不到的效果。比如我在使用打印机打印文件，还没有打完，这时别人也在用打印机打印，如果不加控制，打印的东西肯定是错乱乱的。

进程互斥锁.

互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。

范例一不加互斥锁的情况下：

from multiprocessing import Process,Lock#互斥锁
import os,timedef task():
print("%s print 1"%os.getpid())
time.sleep(3)
print("%s print 2"%(os.getpid()))
print("%s print 3"%(os.getpid()))
if __name__ == '__main__':
p1=Process(target=task)
p2=Process(target=task)
p3=Process(target=task)
p1.start()
p2.start()
p3.start()

结果：

26248 print 1
49232 print 1
49380 print 1
26248 print 2
26248 print 3
49232 print 2
49232 print 3
49380 print 2
49380 print 3

范例二：加互斥锁的情况：

from multiprocessing import Process,Lock#互斥锁
import os,timedef task(mutex):
mutex.acquire()
print("%s print 1"%os.getpid())
time.sleep(3)
print("%s print 2"%(os.getpid()))
print("%s print 3"%(os.getpid()))
mutex.release()
if __name__ == '__main__':
mutex=Lock()
p1=Process(target=task,args=(mutex,))
p2=Process(target=task,args=(mutex,))
p3=Process(target=task,args=(mutex,))
p1.start()
p2.start()
p3.start()

结果：

6664 print 1#是不是这样显得很整洁，一个进程抢到后，直到执行完别的进程才能抢到
6664 print 2
6664 print 3
52764 print 1
52764 print 2
52764 print 3
33084 print 1
33084 print 2
33084 print 3

线程互斥锁，主要是保护数据安全，可以使子线程不可修改全局变量

范例要求，开启100个进程，然后每个进程减一，最后n的结果为0.

不加互斥锁
from threading import Thread,Lock
import time
n=100
def task():
global n
temp=n
time.sleep(0.1)
n=temp-1
if __name__ == '__main__':
t_1=[]
for i in range(10):
t=Thread(target=task,args=())
t_1.append(t)
t.start()
for t in t_1:
t.join()
print(n)

结果：

加互斥锁：

from threading import Thread,Lock
import time
n=100
def task(mutex):
global n
with mutex: #替代mutex。acquire 和mutex.release
temp=n
time.sleep(0.1)
n=temp-1
if __name__ == '__main__':
mutex=Lock()
t_1=[]
for i in range(10):
t=Thread(target=task,args=(mutex,))
t_1.append(t)
t.start()
for t in t_1:
t.join()

print(n)

结果：

范例：要求设计一个旅客买票系统，多个人可以显示剩余票数，但当一个人抢票成功后，票务系统里会自动减一，假设有一张票，一个人抢到后，别人都抢不到了。

from multiprocessing import Process ,Lock
import time,os,random,jsondef search():
with open("db1",encoding="utf-8") as read_f:
dic=json.load(read_f)
print("%s 还剩%s票"%(os.getpid(),dic["count"]))
def get():
with open("db1",encoding="utf-8") as read_f:
dic=json.load(read_f)
if dic["count"]>0:
dic["count"]-=1
time.sleep(random.randint(1,3))
with open("db1","w",encoding="utf-8") as write_f:
json.dump(dic,write_f)
print("%s抢票成功"%os.getpid())def task(mutex):
search()
with mutex:
get()if __name__ == '__main__':
mutex=Lock()
for i in range(10):
p=Process(target=task,args=(mutex,))
p.start()

结果：

66892 还剩1票
6808 还剩1票
46144 还剩1票
27152 还剩1票
35884 还剩1票
47308 还剩1票
15256 还剩1票
44636 还剩1票
48744 还剩1票
62000 还剩1票
66892抢票成功

信号量： 为了防止出现因多个程序同时访问一个共享资源而引发的一系列问题，我们需要一种方法，它可以通过生成并使用令牌来授权，在任一时刻只能有一个执行线程访问代码的临界区域。临界区域是指执行数据更新的代码需要独占式地执行。而信号量就可以提供这样的一种访问机制，让一个临界区同一时间只有一个线程在访问它，也就是说信号量是用来调协进程对共享资源的访问的。其中共享内存的使用就要用到信号量。

进程池和信号量的区别：进程池至始至终只有4个人干活，而信号量开始分成100个进程，这100个进程去抢4个锁，至始至终有100个人干活。（面试中会用到）

信号量和互斥锁的区别：互斥锁同一时间只能有一个人抢到锁，信号了在同一时间有多个人抢到所。

from multiprocessing import Process,Semaphore
import time,random,os
def task(sm):
with sm:
print("%s 上厕所"%os.getpid())
time.sleep(random.randint(1,3))
if __name__ == '__main__':
sm=Semaphore(3)
for i in range(10):
p=Process(target=task,args=(sm,))
p.start()

进程之间通信

https://www.cnblogs.com/sticker0726/p/7931573.html

可以通过文件操作来实现确定速度慢，数据不安全，用硬盘资源

必须找到一种好的机制避免这几种缺点：IPC实现进程之间通过共享内存实现进程之间通信。具体的实现方式有两种：管道和队列

队列（先进先出）是基于管道加锁实现的。（以后用队列，管道自己要亲自加锁）

from multiprocessing import Queue #进程队列
import queue #线程队列
q=Queue() #受限于内存大小
q.put()
q.get()
#堆栈(后进先出）
queue.LifoQueue
#优先级队列(按照优先级出来）
q=queue.PriorityQueue(3)
q.put((10),"ddddd")#数字越小优先级越高
q.put((8),"dd")#数字越小优先级越高

生产者和消费者模型

生产者消费者模型：一类负责造数据，一类负责处理数据，

为什么要用生产者消费者模型？

可以实现程序的解耦，生产者不直接接触消费者。而是与队列打交道，可以一直的生产,提高生产者和消费者的效率

应用:

爬虫
celery

两个线程下的生产者和消费者

from multiprocessing import Queue,Process
import time,randomdef procducer(q):
for i in range(10):
res='包子%s'%i
print('已经造好%s'%(res))
time.sleep(random.randint(1,2))
q.put(res)def consumer(q):
while True:
res = q.get()
if res is None: #当队列中取出是NOne时,结束循环
break print('吃%s'%res)
time.sleep(random.randint(1,3))if __name__ == '__main__':
q=Queue()
p=Process(target=procducer,args=(q,))
c=Process(target=consumer,args=(q,))
p.start()
c.start()
p.join() #保证生产者生产完
q.put(None) #把none放入队列是为了让消费者结束程序
print('主进程')

结果:

已经造好包子0
已经造好包子1
吃包子0
已经造好包子2
吃包子1
已经造好包子3
已经造好包子4
吃包子2
已经造好包子5
已经造好包子6
吃包子3
已经造好包子7
吃包子4
已经造好包子8
已经造好包子9
吃包子5
主进程
吃包子6
吃包子7
吃包子8
吃包子9

但是在现实生产中我们不会用这种方式来实现生产者消费者模型而是用rabbitMQ软件来实现这个功能所以这部分只是了解一下点击进入RabbitMQ

GIL锁:全局解释器锁（重点*********）

GIL锁是作用于多线程:

在Cpython解释器中，同一个进程下开启的多线程，由于有GIL的存在,同一时刻只能有一个线程被cpu执行，即使有多核也无法利用多核优势，无法并行（多核仅仅是对于计算性能的提升，对于IO阻塞不起作用）

首先需要明确的一点是GIL并不是Python的特性，python并没有GIL的锁。它是在实现Python解析器(CPython)时所引入的一个概念

GIL本质就是一把互斥锁，既然是互斥锁，所有互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。

为什么要有这把锁的存在呢? 因为Cpython解释中的垃圾回收机制(gc)不是线程安全的,如果没有了GIL锁,同一进程下的线程利用了多核实现了并行,但是在一个进程中只有一个垃圾回收线程,如果并行的线程中有相同的引用,这时候GC无法同时检测两个线程,导致无法判定该引用内存是保留还是删除导致数据不安全,GIL锁并不是保护你的数据而是保护的是垃圾回收线程的数据.

多核就是为了实现并行,既然cpython解释器同一时间只能只能有一个线程执行,那么多线程就没有就无法发挥多核优势了,这不就说明,Cpython的多线程没有用了.

但是我们要知道计算机的任务分为两种:

计算密集型任务:要进行大量的计算消耗CPU资源,比如计算圆周率、对视频进行高清解码等等，全靠CPU的运算能力。这种计算密集型任务虽然也可以用多任务完成，但是任务越多，花在任务切换的时间就越多，CPU执行任务的效率就越低，所以，要最高效地利用CPU，计算密集型任务同时进行的数量应当等于CPU的核心数. ,比如数据分析，
io密集型任务:我们的web开发，io的时间99%的时间都花在请求网络，从数据库上搜索数据上,即使你使用多核进行并行，也只能快那么一点点，况且多核之间的切换还需要花时间。这时候你的多核优势就无法发挥出优势了比如socket,爬虫,web

GIL什么时候会释放呢？

1.时间片的释放，即指定时间释放。在某个线程运行到一段时间后，它就会释放该GIL锁，让其他线程运行

from threading import Threadtotal = 0
def add():
global total
for i in range(1000000):
total += 1def cut():
global total
for i in range(1000000):
total -= 1if __name__ == '__main__':
add_th = Thread(target=add)
cut_th = Thread(target=cut)
add_th.start()
cut_th.start()
add_th.join()
cut_th.join()
print(total)

结果：

850531

我们可以看到每次执行结果都是不一样的，这说明，GIL是按照时间片来释放的，并不是真正的串行，如果是真正的串行的话，最后total的结果就是0

2.遇到io就会切换，即当该线程遇到io时就会释放GIL

from threading import Thread
import timetotal = 0
def add():
global total
time.sleep(3)
total = total + 50
print('add_total',total)def cut():
global total
total = total + 100
print('total',total)if __name__ == '__main__':
add_th = Thread(target=add)
cut_th = Thread(target=cut)
add_th.start()
cut_th.start()
add_th.join()
cut_th.join()
print(total)

结果：

total 100
add_total 150
150

你会发现，cut先打印出结果，这说明GIL遇到io释放了，被cut获得了，就先执行cut的

结论:在python中要用多核优势开启多进程,