线程基础
初识线程
- 在操作系统中, 每一个进程都有一个地址空间,而且默认就有一个控制线程,cpu真正的执行单位是线程.(就像是在工厂中,每一个车间都有房子,而且每个车间默认就有一条流水线)
- 操作系统 ===> 工厂
- 进程 ===> 车间
- 线程 ===> 流水线
- cpu ===> 电源
- 线程: cpu最小的执行单位
- 进程: 资源集合/资源单位
- 线程运行: 运行代码
- 进程运行: 各种资源 + 线程
当你右键运行:
- 首先会申请内存空间,先把解释器丢进去并且把代码也给丢进去(进程的是事情),运行代码(线程)
进程与线程的区别
- 过程描述的区别
- 线程 ==> 单指代码的执行过程
- 进程 ==> 资源的申请与销毁的过程
- 进程内存空间彼此隔离而同一个进程下的线程共享资源
- 进程和线程的创建速度
- 进程需要申请资源开辟空间 (慢)
- 线程就是告诉操作系统一个执行方案 (快)
线程开启的两种方式
# 方案一
from threading import Thread
import time
def task():
print('线程 start')
time.sleep(3)
print('线程 end')
if __name__ == '__main__': #线程可以不用这行
t = Thread(target=task)
t.start() #告诉操作系统开一个线程
print('主')
# 补充: 进程会等待所有线程结束才会结束
# 方案二
from threading import Thread
import time
class MyT(Thread):
def run(self):
print('子线程 start')
time.sleep(3)
print('子进程 end')
子进程vs子进程创建速度对比
from thteading import Thread
from multiprocessing import Process
import time
def task(name):
print(f'{name} is running')
time.sleep(2)
print(f'{name} is end')
if __name__ == '__main__':
t = Thread(target=task, args=('子线程',))
p = Process(target=task, args=('子进程',))
t.start# 这个子线程较快
p.start
print('')
子线程共享资源
from threading import Thead
import time, os
x = 100
def task():
global x
x = 50
print(os.getpid()) #5204(pid)
if __name__ == '__main__':
t = Thread(target=task)
t.start()
time.sleep(2)
print(x) #50
print(os.getpid()) #5204
线程join的方法
from threading import Thread
import time
def task():
print('子线程 start')
time.sleep(2)
print('子线程 end')
t = Thread(target=task)
t.start
t.join() #等待子线程运行结束
print('主线程')
#案例1
from threading import Thread
import time
def task(name , n):
print(f'{name} start')
time.sleep(n)
print(f'{name} end')
t1 = Thread(target=task,args=('线程1', 1))
t2 = Thread(target=task,args=('线程2', 2))
t3 = Thread(target=task,args=('线程3', 3))
start = time.time()
t1.start
t2.start
t3.start
t1.join() #等待子线程运行结束
t2.join()
t3.join()
end = time.time()
print(end-start) #阻塞时间为3点多秒
print('主线程')
#案例二(了解部分)
from multiprocessing import Process
from threading import Thread
import time
def task():
print('进程 开启') #3
time.sleep(10)
print('进程 结束') #5
def task2():
print('子线程 开启') #1
time.sleep(2)
print('子线程 结束') #4
if __name__ == '__main__':
p = Process(target=task)
t = Thread(target=task2)
t.start() #开线程
p.start() #开进程
print('子进程join开始') #2
p.join() #主进程的主线程等待子进程运行结束
print('主') #6
#结果
子线程 开启
子进程join开始
进程 开启
子线程 结束
进程 结束
主
守护线程
# 守护线程 守护的是进程的运行周期
from threading import Thread,enumerate,currentThread
import time
def task():
print('守护线程开始')
print(currentThread())
time.sleep(20)
# print('守护线程结束')
def task2():
print('子线程 start')
time.sleep(5)
print(enumerate())
print('子线程 end')
if __name__ == '__main__':
t1 = Thread(target=task)
t2 = Thread(target=task2)
t1.daemon = True
t2.start()
t1.start()
print('主')
Thread类的其他用法
-
Thread实例化对象的方法:
-
isAlive(): 返回线程是否活动的 -
getName(): 返回线程名字 -
setName(): 设置
threading模块提供的一些方法:
-
threading.currentThead(): 返回当前的线程变量 -
threading.enumerate(): 返回一个包含正在运行的线程list, 正在运行指线程启动后,结束前,不包含启动前和终止后线程 -
threading.activeCount(): 返回正在运行的线程数量,与len(threading.enumerate())有相同的结果
-
线程锁
线程锁出现的主要原因就是为了数据安全
form threading import Thread, Lock
x = 0
mutex = Lock
def task():
global x
multex.acquire()
for i in range(10000):
x = x+1
#假如t1的x刚那到0 保存了状态 被切换了
#t2的x刚拿到0 进行了+1
#t1 这时候又回来了, x = 0 +1
#这就暴露了一个安全问题,按照逻辑t1的x应该是加了两次1, 而最总的结果,是只加了一次.如果使用线程锁的话就不会出现这个问题
multex.release()
if __name__ == '__main__':
t1 = Threading(target=task)
t2 = Threading(target=task)
t3 = Threading(target=task)
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()
print(x)
死锁问题
from threading import Threading,Lock
import time
mutex1 = Lock()
mutex2 = Lock()
class MyThreada(Thread):
def run(self):
self.task1()
self.task2()
def task1(self):
mutex1.acquire()
print(f'{self.name} 拿到了锁1')
mutex2.acquire()
print(f'{self.name} 拿到了锁2')
mutex2.release()
print(f'{self.name} 释放了锁2')
mutex1.release()
print(f'{self.name} 释放了锁1')
def task2(self):
mutex2.acquire()
print(f'{self.name} 拿到了锁2')
time.sleep(1)
mutex1.acquire()
print(f'{self.name} 拿到了锁1')
mutex1.release()
print(f'{self.name} 释放了锁1')
mutex2.release()
print(f'{self.name} 释放了锁2')
for i in range(3):
t = MyThreada()
t.start()
#死锁问题剖析
#两个线程
#线程1拿到了(锁头2)想要往下执行需要(锁头1)
#线程2拿到了(锁头1)想要往下执行需要(锁头2)
#他们互相拿到了彼此想要往下执行的必须条件, 互相都不放手里的锁头.
递归锁
用于解决死锁问题
- 递归锁:在同一个线程内可以多次被acquire(拿到多次)
- 内部相当于维护了一个计数器也就是说同一个线程 acquire了几次就要release几次
from thteading import Thread ,RLock
import time
mutex1 = RLock()
mutex2 = RLock()
class MyThreada(Thread):
def run(self):
self.task1()
self.task2()
def task1(self):
mutex1.acquire()
print(f'{self.name} 抢到了锁1')
mutex2.acquire()
print(f'{self.name} 抢到了锁2')
mutex2.release()
print(f'{self.name} 释放了锁2')
mutex1.release()
print(f'{self.name} 释放了锁1')
def task2(self):
mutex2.acquire()
print(f'{self.name} 抢到了锁2')
time.sleep(1)
mutex1.acquire()
print(f'{self.name} 抢到了锁1')
mutex1.release()
print(f'{self.name} 释放了锁1')
mutex2.release()
print(f'{self.name} 释放了锁2')
for i in range(3):
t = MyThreada()
t.start()
#完美解决了死锁问题
信号量
from threading import Thread, currentThread,Semaphore
import time
def task():
sm.acquire()
print(f'{currentThread().name} 在执行')
time.sleep(3)
sm.release()
sm = Semaphore(5) #设定一次只能执行5个
for i in range(15):
t = Thread(target=task)
t.start()
GIL全局解释器锁
- 在Cpython解释器中有有一把GIL锁,GIL本质是一把互斥锁.
- 导致了同一个进程下,同一个时间只能运行一个线程, 而无法利用多核优势.
- 同一个进程下多个线程只能是实现并发而不能实现并行.
- 因为cpython自带垃圾回收机制,影响了线程安全.所以要有GIL锁
- 任务分析:
- 加入我们有四个任务需要处理,处理方式肯定要是并发效果, 解决方案可以是:
- 方案一: 开启四个进程
- 方案二: 一个进程下开启四个线程
from threading import Thread
from multiprocessing import Process
import time
# 计算密集型
def work1():
res=0
for i in range(100000000): #1+8个0
res*=i
if __name__ == '__main__':
t_list = []
start = time.time()
for i in range(4):
t = Thread(target=work1)
# t = Process(target=work1)
t_list.append(t)
t.start()
for t in t_list:
t.join()
end = time.time()
print('多线程',end-start) # 多线程 15.413789510726929
# print('多进程',end-start) # 多进程 4.711405515670776
# io密集型
def work1():
x = 1+1
time.sleep(5)
if __name__ == '__main__':
t_list = []
start = time.time()
for i in range(4):
t = Thread(target=work1)
# t = Process(target=work1)
t_list.append(t)
t.start()
for t in t_list:
t.join()
end = time.time()
print('多线程',end-start) # 多线程 5.002625942230225
# print('多进程',end-start) # 多进程 5.660863399505615
线程之queue模块
直接上代码解释
#案例一之join
import queue
q = queue.Queue()
q.put('123')
q.put('456')
print(q.get())
print(q.get())#put几次就get几次
#print(q.get())#因为已经get完了,这里会停顿.
q.task_done()
q.task_done()
q.join() #单使用join也是会停顿在这里,取了两次task_done通知两次.
#案例二之LifoQueue
q = queue.LifoQueue() #会出现堆栈的效果, 就先进先出的效果.
q.put('内衣')
q.put('内裤')
q.put('拖鞋')
print(q.get()) #这里会先取出来'拖鞋'
print(q.get()) #会取出'内裤'
print(q.get()) #会取出'内衣'
#案例三之PriorityQueue
q = queue.PriorityQueue() #可以根据优先级来取数据
q.put((3, '奥特曼')) #这里面要放置元祖类型,通常第一个值是int类型
q.put((2, '布娃娃'))
q.put((1, '娃哈哈'))
print(q.get()) #会优先取出'娃哈哈',根据从小到大的顺序
print(q.get()) #取出'布娃娃'
print(q.get()) #取出'奥特曼'
线程定时器
from threading import Thread, Timer
import time
def task():
print('线程执行了')
time.sleep(2)
print('线程结束了')
t = Timer(n, task) #表示过了n秒后开启一个线程
t.start()
print('下面的代码') #上面的n秒后并不会影响到我.
进程池和线程池
- 池的功能: 限制进程数或者线程数.
- 一般用在当并发任务数量远远大于计算机所能承受的最大范围,既无法一次性开启过多的任务数量,那就应该考虑去限制进程数和线程数,从而保证服务区不宕机.
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from Threading import currentThread
from multiprocessing import current_process
import time
def task(i):
print(f'{currentThead().name} 在指向人物{i}')
time.sleep(1)
return i+2
if __name__ == '__main__':
pool = ThreadPoolExecutor(4) #规定池子里只有4个线程在工作
fu_list = []
for i in range(20):
pool.submit(task,i) #task任务要做20次,4个线程负责这个事情
future = pool.submit(task, i)
print(future.result())# 如果没有结果会一直等待拿到结果,导致所有的任务在串行
fu_list.append(future)#future为返回值结果
pool.shutdown() #会关闭池的入口,等待所有的任务执行完,结束阻塞
for fu in fu_list:
print(fu.result()) #等待所有的任务执行完,才会拿到最后的结果
同步和异步
可以理解为提交任务的两种方式
- 同步: 提交了一个任务,必须要等任务执行完了(拿到返回值),才能执行下一行代码)
- 异步: 提交了一个任务,不需要等待执行完了,就可以直接执行下一行代码.
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import currentThread
from multiprocessing import current_process
import time
def task(i):
print(f'{currentThread().name} 在执行任务 {i}')
# print(f'进程 {current_process().name} 在执行任务 {i}')
time.sleep(1)
return i+2
def parse(future):
print(future.result())#处理拿到的结果
if __name__ == '__main__':
pool = ThreadPoolExecutor(4) #池子里有4个线程
pool = ProcessPoolExecutor(4) #池子里有4个进程
fu_list = []
for i in range(20):
# pool.submit(task,i) # task任务要做20次,4个线程负责做这个事
future = pool.submit(task,i) # task任务要做20次,4个进程负责做这个事.
future.add_done_callback(parse)
#表示为当前任务绑定了一个函数,在当前任务执行结束的时候会触发这个函数
#会把future的对象作为参数传给函数
#这个就称之为回调函数,处理完了回来就调用这个函数.
#这个不会像上面的一个等一下处理完在拿,而是只要你处理完出来一个那我就会拿.
协程
-
python的线程用的是操作系统原生线程
-
协程就是在单线程下实现并发
- 并发: 切换状态+保存状态
- 多线程: 操作系统帮你实现的,如果遇到io切换,执行时间过下长也会切换,实现一个雨露均沾的效果.
-
什么样的协程是由意义的?
-
遇到io切换的时候才有意义
-
具体是: 协程的概念本质上是程序员抽象出来的, 操作系统根本不知道协程存在,也就是说来了一个线程,我自己遇到了io,我自己线程内部直接切到了自己的别的任务上取了,操作系统是根本发现不了的,这也就实现了单线程下效率最高.
优点
自己控制切换要比操作系统切换快的多.降低了单个线程的io时间.
缺点
对比多线程
自己要检测所有的io,但凡有一个阻塞那么整体都会跟着阻塞.
对比多进程
无法利用多核优势.
-
# 对比通过yeild切换运行的时间反而比串行更消耗时间，这样实现的携程是没有意义的。
# import time
#
# def func1():
# for i in range(100000000):
# i+1
# def func2():
# for i in range(100000000):
# i+1
#
# start = time.time()
# func1()
# func2()
# stop = time.time()
# print(stop - start) # 8.630893230438232
#案例二通过导入一个模块
from gevent import mokey;monkey.patch_all()
import gevent
import time
def eat():
print('eat 1')
time.sleep(3)
print('eat 2')
def play():
print('play 1')
time.sleep(3)
print('play 2')
start = time.time()
g1 = gevent.spawn(eat)
g2 = gevent.spawn(play)
g1.join()
g2.join()
end = time.time()
print(end-start) #3点多秒
<完>