线程（Threads）
进程是资源分配的最小单位,线程是CPU调度的最小单位.
每一个进程中至少有一个线程
Python代码的执行由Python虚拟机(也叫解释器主循环)来控制。Python在设计之初就考虑到要在主循环中，同时只有一个线程在执行。虽然 Python 解释器中可以“运行”多个线程，但在任意时刻只有一个线程在解释器中运行。
Cpython解释器下 GIL 全局解释器锁 保证了同一时刻下只有一个线程可以被CPU操作 threading模块 创建子线程 Thread类 start 开启子线程 join 阻塞等待子线程结束 setDeamon 设置守护线程 会等待主线程结束(包含所有非守护的子线程)之后守护线程才结束 currentthread,enumerate,activecount 查看当前线程,所有的线程对象组成的列表,列表的长度
线程的特点:
1）轻型实体
2）独立调度和分派的基本单位。
3）共享进程资源。
4）可并发执行。(在一个进程中的多个线程之间，可以并发执行，甚至允许在一个进程中所有线程都能并发执行；同样，不同进程中的线程也能并发执行，充分利用和发挥了处理机与外围设备并行工作的能力。)
threading模块
线程的创建Threading.Thread类
from threading import Threadimport timedef sayhi(name): time.sleep(2) print('%s say hello' %name)if __name__ == '__main__': t=Thread(target=sayhi,args=('egon',)) t.start() print('主线程')创建线程的方式1
from threading import Threadimport timeclass Sayhi(Thread): def __init__(self,name): super().__init__() self.name=name def run(self): time.sleep(2) print('%s say hello' % self.name)if __name__ == '__main__': t = Sayhi('egon') t.start() print('主线程')创建线程的方式2 多线程与多进程
from threading import Threadfrom multiprocessing import Processimport osdef work(): print('hello',os.getpid())if __name__ == '__main__': #part1:在主进程下开启多个线程,每个线程都跟主进程的pid一样 t1=Thread(target=work) t2=Thread(target=work) t1.start() t2.start() print('主线程/主进程pid',os.getpid()) #part2:开多个进程,每个进程都有不同的pid p1=Process(target=work) p2=Process(target=work) p1.start() p2.start() print('主线程/主进程pid',os.getpid())pid的比较 from threading import Threadfrom multiprocessing import Processimport osdef work(): global n n=0if __name__ == '__main__': # n=100 # p=Process(target=work) # p.start() # p.join() # print('主',n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100 n=1 t=Thread(target=work) t.start() t.join() print('主',n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据同一进程内的线程共享该进程的数据？内存数据的共享问题 from threading import Threadfrom multiprocessing import Processimport osdef work(): print('hello')if __name__ == '__main__': #在主进程下开启线程 t=Thread(target=work) t.start() print('主线程/主进程') ''' 打印结果: hello 主线程/主进程 ''' #在主进程下开启子进程 t=Process(target=work) t.start() print('主线程/主进程') ''' 打印结果: 主线程/主进程 hello '''开启效率的较量 Thread类的其他方法
Thread实例对象的方法 # isAlive(): 返回线程是否活动的。 # getName(): 返回线程名。 # setName(): 设置线程名。threading模块提供的一些方法： # threading.currentThread(): 返回当前的线程变量。 # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
守护线程
无论是进程还是线程，都遵循：守护xx会等待主xx运行完毕后被销毁。需要强调的是：运行完毕并非终止运行
#1.对主进程来说，运行完毕指的是主进程代码运行完毕#2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕
#1 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束，#2 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。
from threading import Threadimport timedef sayhi(name): time.sleep(2) print('%s say hello' %name)if __name__ == '__main__': t=Thread(target=sayhi,args=('egon',)) t.setDaemon(True) #必须在t.start()之前设置 t.start() print('主线程') print(t.is_alive()) ''' 主线程 True '''守护线程例1 from threading import Threadimport timedef foo(): print(123) time.sleep(1) print("end123")def bar(): print(456) time.sleep(3) print("end456")t1=Thread(target=foo)t2=Thread(target=bar)t1.daemon=Truet1.start()t2.start()print("main-------")守护线程例2 死锁与递归锁
所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁
# 科学家吃面问题import timefrom threading import Thread,Locknoodle_lock = Lock()fork_lock = Lock()# 死锁不是时刻发生的,有偶然的情况整个程序都崩了# 每一个线程之中不止一把锁,并且套着使用# 如果某一件事情需要两个资源同时出现,那么不应该将这两个资源通过两把锁控制# 而应看做一个资源def eat1(name): noodle_lock.acquire() print('%s拿到面条了'%name) fork_lock.acquire() print('%s拿到叉子了'%name) print('%s开始吃面'%name) time.sleep(0.2) fork_lock.release() print('%s放下叉子了' % name) noodle_lock.release() print('%s放下面了' % name)def eat2(name): fork_lock.acquire() print('%s拿到叉子了' % name) noodle_lock.acquire() print('%s拿到面条了' % name) print('%s开始吃面' % name) time.sleep(0.2) noodle_lock.release() print('%s放下面了' % name) fork_lock.release() print('%s放下叉子了' % name)Thread(target=eat1,args=('wei',)).start()Thread(target=eat2,args=('hao',)).start()Thread(target=eat1,args=('太',)).start()Thread(target=eat2,args=('宝',)).start() 出现了死锁现象,最快的解决方法: noodle_lock = fork_lock = RLock() 但是这只作为一种应急,本质上解决就是设置一个锁,示例代码如下:
lock = Lock() def eat1(name): lock.acquire() print('%s拿到面条了'%name) print('%s拿到叉子了'%name) print('%s开始吃面'%name) time.sleep(0.2) lock.release() print('%s放下叉子了' % name) print('%s放下面了' % name) def eat2(name): lock.acquire() print('%s拿到叉子了' % name) print('%s拿到面条了' % name) print('%s开始吃面' % name) time.sleep(0.2) lock.release() print('%s放下面了' % name) print('%s放下叉子了' % name) Thread(target=eat1,args=('alex',)).start() Thread(target=eat2,args=('wusir',)).start() Thread(target=eat1,args=('太白',)).start() Thread(target=eat2,args=('宝元',)).start() 递归锁
RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源.
lock = Lock()def eat1(name): lock.acquire() print('%s拿到面条了'%name) print('%s拿到叉子了'%name) print('%s开始吃面'%name) time.sleep(0.2) lock.release() print('%s放下叉子了' % name) print('%s放下面了' % name)def eat2(name): lock.acquire() print('%s拿到叉子了' % name) print('%s拿到面条了' % name) print('%s开始吃面' % name) time.sleep(0.2) lock.release() print('%s放下面了' % name) print('%s放下叉子了' % name)Thread(target=eat1,args=('alex',)).start()Thread(target=eat2,args=('wusir',)).start()Thread(target=eat1,args=('太白',)).start()Thread(target=eat2,args=('宝元',)).start() 互斥锁 无论在相同的线程还是不同的线程,都只能连续acquire一次 要想再acquire,必须先release 递归锁 在同一个线程中,可以无限次的acquire 但是要想在其他线程中也acquire, 必须现在自己的线程中添加和acquire次数相同的release
dis 模块
import disn = 1def func(): n = 100 n -= 1dis.dis(func)# 会出现线程不安全的两个条件# 1.是全局变量# 2.出现 += -=这样的操作# 列表 字典# 方法 l.append l.pop l.insert dic.update 都是线程安全的# l[0] += 1# d[k] += 1
信号量
import timefrom threading import Semaphore,Threaddef func(name,sem): sem.acquire() print(name,'start') time.sleep(1) print(name,'stop') sem.release()sem = Semaphore(5)for i in range(20): Thread(target=func,args=(i,sem)).start()信号量和池进程池 有1000个任务 一个进程池中有5个进程 所有的1000个任务会多次利用这五个进程来完成任务信号量 有1000个任务 有1000个进程/线程 所有的1000个任务由于信号量的控制,只能5个5个的执行
事件
线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行
from threading import Event# 事件# wait() 阻塞 到事件内部标识为True就停止阻塞# 控制标识 # set # clear # is_set# 连接数据库import timeimport randomfrom threading import Thread,Eventdef connect_sql(e): count = 0 while count < 3: e.wait(0.5) if e.is_set(): print('连接数据库成功') break else: print('数据库未连接成功') count += 1def test(e): time.sleep(random.randint(0,3)) e.set()e = Event()Thread(target=test,args=(e,)).start()Thread(target=connect_sql,args=(e,)).start() 条件 condition
使得线程等待，只有满足某条件时，才释放n个线程
Python提供的Condition对象提供了对复杂线程同步问题的支持。Condition被称为条件变量，除了提供与Lock类似的acquire和release方法外，还提供了wait和notify方法。线程首先acquire一个条件变量，然后判断一些条件。如果条件不满足则wait；如果条件满足，进行一些处理改变条件后，通过notify方法通知其他线程，其他处于wait状态的线程接到通知后会重新判断条件。不断的重复这一过程，从而解决复杂的同步问题。
# wait 阻塞# notify(n) 给信号import threadingdef run(n): con.acquire() con.wait() print("run the thread: %s" % n) con.release()if __name__ == '__main__': con = threading.Condition() for i in range(10): t = threading.Thread(target=run, args=(i,)) t.start() while True: inp = input('>>>') if inp == 'q': break con.acquire() con.notify(int(inp)) con.release() print('****') 定时器
from threading import Timerdef func(): print('执行我啦')t = Timer(3,func)# 现在这个时间点我不想让它执行,而是预估一下大概多久之后它执行比较合适t.start()print('主线程的逻辑') 队列
import queue 线程队列 线程之间数据安全q = queue.Queue(1)普通队列q.put(1)print(q.get())try: q.put_nowait(2)except queue.Full: print('您丢失了一个数据2')print(q.get_nowait()) # 如果有数据我就取,如果没数据不阻塞而是报错非阻塞的情况下q.put(10)print(q.get(timeout=2))算法里 栈lfq = queue.LifoQueue() # 栈lfq.put(1)lfq.put(2)lfq.put(3)print(lfq.get())print(lfq.get())print(lfq.get())优先级队列,是根据第一个值的大小来排定优先级的ascii码越小,优先级越高q = queue.PriorityQueue()q.put((2,'a'))q.put((1,'c'))q.put((1,'b'))print(q.get())线程+队列 实现生产者消费者模型