线程池进程池

目录

• 验证GIL锁的存在方式
• GIL与普通互斥锁的区别
• io密集型和计算密集型
• 死锁现象(哲学家就餐问题)
• Semaphore信号量
• Event事件
• 线程queue
• 线程池
• 进程池
• multiprocess.Pool模块

验证GIL锁的存在方式

from threading import Thread
from multiprocessing import Process

def task():
    while True:
        pass
if __name__ == '__main__':
    for i in range(6):
		# t=Thread(target=task)  # 因为有GIL锁，同一时刻，只有一条线程执行，所以cpu不会满
        t=Process(target=task)   # 由于是多进程，进程中的线程会被cpu调度执行，6个cpu全在工作，就会跑满
        t.start()

GIL与普通互斥锁的区别

GIL锁是不能保证数据的安全,普通互斥锁来保证数据安全
from threading import Thread,Lock
import time

mutex = Lock()
money = 100

def task():
    global money
    mutex.acquire()
    temp = money
    time.sleep(1)
    money = temp-1
    mutex.release()
    

if __name__ == '__main__':
    ll=[]
    for i in range(10):
        t = Thread(target=task)
        t.start()
        # t.join()  # 会怎么样？变成了串行，不能这么做
        ll.append(t)
    for t in ll:
        t.join()
    print(money)
 

io密集型和计算密集型

cpython解释器
在单核情况下:
    开多线程还是多进程,不管干什么都是开多线程
在多核情况下:
    如果是计算密集型,需要开进程,能被多个cpu调度执行
    如果是io密集型,需要开线程,cpu遇到io会切换到其他线程执行

from threading import Thread
from multiprocessing import Process
import time

计算密集型
def task():
    count = 0
    for i in range(100000000):
        count += i


if __name__ == '__main__':
    ctime = time.time()
    ll = []
    for i in range(10):
        t = Thread(target=task)  # 开线程：42.68658709526062
        # t = Process(target=task)   # 开进程：9.04949426651001
        t.start()
        ll.append(t)

    for t in ll:
        t.join()
    print(time.time()-ctime)

# io密集型
def task():
    time.sleep(2)


if __name__ == '__main__':
    ctime = time.time()
    ll = []
    for i in range(400):
        t = Thread(target=task)  # 开线程：2.0559656620025635
        # t = Process(target=task)   # 开进程：9.506720781326294
        t.start()
        ll.append(t)

    for t in ll:
        t.join()
    print(time.time()-ctime)

死锁现象(哲学家就餐问题)

是指两个或两个以上的进程或线程在执行过程中,因争夺资源而造成的一种互相等待的现象,若无外力作用,他们都将无法推进下去.此时称系统处于死锁状态或系统产生了死锁,这些永远在互相等待的进程称为死锁进程,如下就是死锁:
    死锁现象,张三拿到了A锁,等B锁,李四拿到了B锁,等A锁
    
from threading import Thread,Lock
import time
mutexA = Lock()
mutexB = Lock()

def eat_apple(name):
    mutexA.acquire()
    print('%s 获取到了a锁'%name)
    mutexB.acquire()
    print('%s 获取到了b锁' % name)
    print('开始吃苹果，并且吃完了')
    mutexB.release()
    print('%s 释放了b锁' % name)
    mutexA.release()
    print('%s 释放了a锁' % name)
    
def eat_egg(name):
    mutexB.acquire()
    print('%s 获取到了b锁' % name)
    time.sleep(2)
    mutexA.acquire()
    print('%s 获取到了a锁' % name)
    print('开始吃鸡蛋，并且吃完了')
    mutexA.release()
    print('%s 释放了a锁' % name)
    mutexB.release()
    print('%s 释放了b锁' % name)


if __name__ == '__main__':
    ll = ['egon', 'alex', '铁蛋']
    for name in ll:
        t1 = Thread(target=eat_apple, args=(name,))
        t2 = Thread(target=eat_egg, args=(name,))
        t1.start()
        t2.start()        

递归锁(可重入),同一个人可以多次acquire,每acquire一次,内部计数器加1,每relaese一次
内部计数器减1
只有计数器不为0,其他人都不获得这把锁

from threading import Thread,Lock,RLock
import time
# 同一把锁
# mutexA = Lock()
# mutexB = mutexA

# 使用可重入锁解决（同一把锁）
# mutexA = RLock()
# mutexB = mutexA
mutexA = mutexB =RLock()

def eat_apple(name):
    mutexA.acquire()
    print('%s 获取到了a锁' % name)
    mutexB.acquire()
    print('%s 获取到了b锁' % name)
    print('开始吃苹果，并且吃完了')
    mutexB.release()
    print('%s 释放了b锁' % name)
    mutexA.release()
    print('%s 释放了a锁' % name)


def eat_egg(name):
    mutexB.acquire()
    print('%s 获取到了b锁' % name)
    time.sleep(2)
    mutexA.acquire()
    print('%s 获取到了a锁' % name)
    print('开始吃鸡蛋，并且吃完了')
    mutexA.release()
    print('%s 释放了a锁' % name)
    mutexB.release()
    print('%s 释放了b锁' % name)


if __name__ == '__main__':
    ll = ['egon', 'alex', '铁蛋']
    for name in ll:
        t1 = Thread(target=eat_apple, args=(name,))
        t2 = Thread(target=eat_egg, args=(name,))
        t1.start()
        t2.start()

Semaphore信号量

Semaphore:信号量可以立即为多把锁,同时允许多个线程来更改数据
 
from threading import Thread,Semaphore
import timme
import random
sm = Semaphore(3) # 数字表示可以同时有多少个线程操作

def task(name):
    sm.acquire()
    print("%s 正在蹲坑"%name)
    time.sleep(random.randint(1,5))
    sm.release()
    

Event事件

一些线程需要等到其他线程执行完成之后才能执行,类似于发射信号
比如一个线程等待另一个线程执行结束在继续执行
一些线程需要等到其他线程执行完成之后才能执行,类似于发射信号
比如一个线程等待另一个线程执行结束在继续执行
 from threading import Thread, Event
 import time

 event = Event()


 def girl(name):
     print('%s 现在不单身，正在谈恋爱'%name)
     time.sleep(10)
     print('%s 分手了，给屌丝男发了信号'%name)
     event.set()


 def boy(name):
     print('%s 在等着女孩分手'%name)
     event.wait()  # 只要没来信号，就卡在者
     print('女孩分手了，机会来了，冲啊')


 if __name__ == '__main__':
     lyf = Thread(target=girl, args=('刘亦菲',))
     lyf.start()

     for i in range(10):
         b = Thread(target=boy, args=('屌丝男%s号' % i,))
         b.start()

起两个线程，第一个线程读文件的前半部分，读完发一个信号，另一个进程读后半部分，并打印

from threading import Thread,Event
import time
import os

event = Event()
# 获取文件的总大小
size = os.path.getsize('a.txt')


def read_first():
    with open('a.txt','r',encoding='utf-8')as f:
        n = size // 2 # 取文件的一半,整除
        data = f.read(n)
        print(data)
        print('我一半读完了,发个信号')
        event.set()
def read_last():
    event.wait()  # 等着发信号
    with open('a.txt', 'r', encoding='utf-8') as f:
        n = size // 2  # 取文件一半，整除
        # 光标从文件开头开始，移动了n个字节，移动到文件一半
        f.seek(n, 0)
        data = f.read()
        print(data)


if __name__ == '__main__':
    t1=Thread(target=read_first)
    t1.start()
    t2=Thread(target=read_last)
    t2.start()     

线程queue

进程queue和线程不是一个

from multiprocessing import	Queue

线程queue
from queue import Queue,LifoQueue,PriorityQueue

# 线程间通信，因为共享变量会出现数据不安全问题，用线程queue通信，不需要加锁，内部自带
# queue是线程安全的


'''
三种线程Queue
    -Queue:队列，先进先出
    -PriorityQueue：优先级队列，谁小谁先出
    -LifoQueue：栈，后进先出
'''
# 如何使用
# q=Queue(5)
# q.put("lqz")
# q.put("egon")
# q.put("铁蛋")
# q.put("钢弹")
# q.put("金蛋")
#
#
# # q.put("银蛋")
# # q.put_nowait("银蛋")
# # 取值
# print(q.get())
# print(q.get())
# print(q.get())
# print(q.get())
# print(q.get())
# # 卡住
# # print(q.get())
# # q.get_nowait()
# # 是否满，是否空
# print(q.full())
# print(q.empty())

# LifoQueue

# q=LifoQueue(5)
# q.put("lqz")
# q.put("egon")
# q.put("铁蛋")
# q.put("钢弹")
# q.put("金蛋")
# #
# # q.put("ddd蛋")
# print(q.get())


#PriorityQueue:数字越小，级别越高

# q=PriorityQueue(3)
# q.put((-10,'金蛋'))
# q.put((100,'银蛋'))
# q.put((101,'铁蛋'))
# # q.put((1010,'铁dd蛋'))  # 不能再放了
#
# print(q.get())
# print(q.get())
# print(q.get())

线程池

为什么会出现池:
    不管是开进程还是开线程,不能无限制开,通过池,假设池子里就有10个,不管再怎么开,永远就是这10个
 如何使用

from concurrent.futures import ThreadPoolExecutor
from threading import Thread
import time,random

pool = ThreadPoolExecutor(5) # 数字是池的大小

def task(name):
	print('%s任务开始'%name)
	time.sleep(random.randint(1,4))
	print('任务结束')
	return '%s 返回了'%name


def call_back(f):
	print(f.result())


if __name__ == '__main__':
    for i in range(10): # 起了10个线程
	    #向线程池中提交了一个任务,等待任务执行完成,自动回到call_back函数执行
	pool.submit(task,'屌丝男%s号'%i).add_done_callback(call_back)

进程池

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Process
import time, random

pool = ProcessPoolExecutor(5)  # 数字是池的大小


def task(name):
	print('%s任务开始' % name)
	time.sleep(random.randint(1, 4))
	print('任务结束')
	return '%s 返回了' % name


def call_back(f):
	print(f.result())


if __name__ == '__main__':
	for i in range(10):  # 起了10个线程
		# 向进程池中提交了一个任务,等待任务执行完成,自动回到call_back函数执行
		pool.submit(task, '屌丝男%s号' % i).add_done_callback(call_back)

multiprocess.Pool模块

1 numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值
2 initializer：是每个工作进程启动时要执行的可调用对象，默认为None
3 initargs：是要传给initializer的参数组

主要方法
1 p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
2 '''需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()'''
3 
4 p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。
5 '''此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。'''
6    
7 p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
8 
9 P.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用

其他方法
1 方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。实例具有以下方法
2 obj.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发。
3 obj.ready():如果调用完成，返回True
4 obj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常
5 obj.wait([timeout]):等待结果变为可用。
6 obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数