查看: 527|回复: 0

Python_线程

[复制链接]
发表于 2020-3-8 10:29:58 | 显示全部楼层 |阅读模式
线程概念的引入配景

历程

  之前我们已经了解了操纵体系中历程的概念,程序并不能单独运行,只有将程序装载到内存中,体系为它分配资源才气运行,而这种执行的程序就称之为历程。程序和历程的区别就在于:程序是指令的集合,它是历程运行的静态描述文本;历程是程序的一次执行活动,属于动态概念。在多道编程中,我们允许多个程序同时加载到内存中,在操纵体系的调理下,可以实现并发地执行。这是这样的设计,大大提高了CPU的利用率。历程的出现让每个用户感觉到自己独享CPU,因此,历程就是为了在CPU上实现多道编程而提出的。
有了历程为什么要有线程

  历程有很多长处,它提供了多道编程,让我们感觉我们每个人都拥有自己的CPU和其他资源,可以提高盘算机的利用率。很多人就不明白了,既然历程这么良好,为什么还要线程呢?其实,仔细观察就会发现历程照旧有很多缺陷的,主要体现在两点上:

  • 历程只能在一个时间干一件事,如果想同时干两件事或多件事,历程就无能为力了。
  • 历程在执行的过程中如果阻塞,例如等候输入,整个历程就会挂起,纵然历程中有些工作不依赖于输入的数据,也将无法执行。
  如果这两个缺点明白比较困难的话,举个实际的例子也许你就清晰了:如果把我们上课的过程当作一个历程的话,那么我们要做的是耳朵听老师讲课,手上还要记笔记,脑筋还要思考问题,这样才气高效的完成听课的任务。而如果只提供历程这个机制的话,上面这三件事将不能同时执行,同一时间只能做一件事,听的时候就不能记笔记,也不能用脑筋思考,这是其一;如果老师在黑板上写演算过程,我们开始记笔记,而老师突然有一步推不下去了,阻塞住了,他在那里思考着,而我们呢,也不能干其他事,纵然你想趁此时思考一下刚才没听懂的一个问题都不可,这是其二。
  现在你应该明白了历程的缺陷了,而解决的办法很简单,我们完全可以让听、写、思三个独立的过程,并行起来,这样很明显可以提高听课的服从。而实际的操纵体系中,也同样引入了这种类似的机制——线程。
线程的出现

  60年代,在OS中能拥有资源和独立运行的根本单位是历程,然而随着盘算机技术的发展,历程出现了很多毛病,一是由于历程是资源拥有者,创建、撤消与切换存在较大的时空开销,因此需要引入轻型历程;二是由于对称多处理机(SMP)出现,可以满足多个运行单位,而多个历程并行开销过大。  因此在80年代,出现了能独立运行的根本单位——线程(Threads)  注意:历程是资源分配的最小单位,线程是CPU调理的最小单位.     每一个历程中至少有一个线程。 历程和线程的关系

  

  线程与历程的区别可以归纳为以下4点:  1)地址空间和其它资源(如打开文件):历程间相互独立,同一历程的各线程间共享。某历程内的线程在其它历程不可见。  2)通信:历程间通信IPC,线程间可以直接读写历程数据段(如全局变量)来进行通信——需要历程同步和互斥手段的辅助,以保证数据的一致性。  3)调理和切换:线程上下文切换比历程上下文切换要快得多。  4)在多线程操纵体系中,历程不是一个可执行的实体。  *通过漫画了解线程进城线程的特点

  在多线程的操纵体系中,通常是在一个历程中包罗多个线程,每个线程都是作为利用CPU的根本单位,是耗费最小开销的实体。线程具有以下属性。  1)轻型实体  线程中的实体根本上不拥有体系资源,只是有一点必不可少的、能保证独立运行的资源。  线程的实体包罗程序、数据和TCB。线程是动态概念,它的动态特性由线程控制块TCB(Thread Control Block)描述。
  1. TCB包罗以下信息:(1)线程状态。(2)当线程不运行时,被生存的现场资源。(3)一组执行堆栈。(4)存放每个线程的局部变量主存区。(5)访问同一个历程中的主存和其它资源。用于指示被执行指令序列的程序计数器、保留局部变量、少数状态参数和返回地址等的一组寄存器和堆栈。
复制代码
TCB包罗以下信息  2)独立调理和分派的根本单位。  在多线程OS中,线程是能独立运行的根本单位,因而也是独立调理和分派的根本单位。由于线程很“轻”,故线程的切换非常敏捷且开销小(在同一历程中的)。  3)共享历程资源。  线程在同一历程中的各个线程,都可以共享该历程所拥有的资源,这首先表现在:所有线程都具有雷同的历程id,这意味着,线程可以访问该历程的每一个内存资源;此外,还可以访问历程所拥有的已打开文件、定时器、信号量机构等。由于同一个历程内的线程共享内存和文件,以是线程之间互相通信不必调用内核。  4)可并发执行。  在一个历程中的多个线程之间,可以并发执行,甚至允许在一个历程中所有线程都能并发执行;同样,不同历程中的线程也能并发执行,充分利用和发挥了处理机与外围设备并行工作的能力。  使用线程的实际场景



  开启一个字处理软件历程,该历程肯定需要办不止一件事情,比如监听键盘输入,处理文字,定时自动将文字生存到硬盘,这三个任务操纵的都是同一块数据,因而不能用多历程。只能在一个历程里并发地开启三个线程,如果是单线程,那就只能是,键盘输入时,不能处理文字和自动生存,自动生存时又不能输入和处理文字。
内存中的线程



  多个线程共享同一个历程的地址空间中的资源,是对一台盘算机上多个历程的模仿,有时也称线程为轻量级的历程。
  而对一台盘算机上多个历程,则共享物理内存、磁盘打印机等其他物理资源。多线程的运行也多历程的运行类似,是cpu在多个线程之间的快速切换。
  不同的历程之间是充满敌意的,彼此是抢占、竞争cpu的关系,如果迅雷会和QQ抢资源。而同一个历程是由一个程序员的程序创建,以是同一历程内的线程是合作关系,一个线程可以访问另外一个线程的内存地址,大家都是共享的,一个线程干死了另外一个线程的内存,那纯属程序员脑筋有问题。
  类似于历程,每个线程也有自己的堆栈,不同于历程,线程库无法利用时钟中断强制线程让出CPU,可以调用thread_yield运行线程自动放弃cpu,让另外一个线程运行。
  线程通常是有益的,但是带来了不小程序设计难度,线程的问题是:
  1. 父历程有多个线程,那么开启的子线程是否需要同样多的线程
  2. 在同一个历程中,如果一个线程关闭了文件,而另外一个线程正准备往该文件内写内容呢?
  因此,在多线程的代码中,需要更多的心思来设计程序的逻辑、保护程序的数据。
用户级线程和内核级线程(了解)

  线程的实现可以分为两类:用户级线程(User-Level Thread)和内核线线程(Kernel-Level Thread),后者又称为内核支持的线程或轻量级历程。在多线程操纵体系中,各个体系的实现方式并不雷同,在有的体系中实现了用户级线程,有的体系中实现了内核级线程。
用户级线程

  内核的切换由用户态程序自己控制内核切换,不需要内核干涉,少了收支内核态的斲丧,但不能很好的利用多核Cpu。
  

  在用户空间模仿操纵体系对历程的调理,来调用一个历程中的线程,每个历程中都会有一个运行时体系,用来调理线程。此时当该历程获取cpu时,历程内再调理出一个线程去执行,同一时刻只有一个线程执行。
内核级线程

   内核级线程:切换由内核控制,当线程进行切换的时候,由用户态转化为内核态。切换完毕要从内核态返回用户态;可以很好的利用smp,即利用多核cpu。Windows线程就是这样的。
  

用户级与内核级线程的对比

  1. 1 内核支持线程是OS内核可感知的,而用户级线程是OS内核不可感知的。2 用户级线程的创建、撤消和调理不需要OS内核的支持,是在语言(如Java)这一级处理的;而内核支持线程的创建、撤消和调理都需OS内核提供支持,而且与历程的创建、撤消和调理大体是雷同的。3 用户级线程执行体系调用指令时将导致其所属历程被中断,而内核支持线程执行体系调用指令时,只导致该线程被中断。4 在只有用户级线程的体系内,CPU调理照旧以历程为单位,处于运行状态的历程中的多个线程,由用户程序控制线程的轮换运行;在有内核支持线程的体系内,CPU调理则以线程为单位,由OS的线程调理程序负责线程的调理。5 用户级线程的程序实体是运行在用户态下的程序,而内核支持线程的程序实体则是可以运行在任何状态下的程序。
复制代码
用户级线程和内核级线程的区别
  1. 长处:当有多个处理机时,一个历程的多个线程可以同时执行。缺点:由内核进行调理。
复制代码
内核线程的优缺点
  1. 长处:线程的调理不需要内核直接参与,控制简单。可以在不支持线程的操纵体系中实现。创建和销毁线程、线程切换代价等线程管理的代价比内核线程少得多。允许每个历程定制自己的调理算法,线程管理比较机动。线程能够利用的表空间和堆栈空间比内核级线程多。同一历程中只能同时有一个线程在运行,如果有一个线程使用了体系调用而阻塞,那么整个历程都会被挂起。另外,页面失效也会产生同样的问题。缺点:资源调理按照历程进行,多个处理机下,同一个历程中的线程只能在同一个处理机下分时复用
复制代码
用户级线程的优缺点混淆实现

  用户级与内核级的多路复用,内核同一调理内核线程,每个内核线程对应n个用户线程
  

linux操纵体系的 NPTL    

  1. 汗青在内核2.6以前的调理实体都是历程,内核并没有真正支持线程。它是能过一个体系调用clone()来实现的,这个调用创建了一份调用历程的拷贝,跟fork()不同的是,这份历程拷贝完全共享了调用历程的地址空间。LinuxThread就是通过这个体系调用来提供线程在内核级的支持的(许多以前的线程实现都完全是在用户态,内核根本不知道线程的存在)。非常不幸的是,这种方法有相称多的地方没有遵循POSIX标准,特殊是在信号处理,调理,历程间通信原语等方面。很显然,为了改进LinuxThread必须得到内核的支持,而且需要重写线程库。为了实现这个需求,开始有两个相互竞争的项目:IBM启动的NGTP(Next Generation POSIX Threads)项目,以及Redhat公司的NPTL。在2003年的年中,IBM放弃了NGTP,也就是大约那时,Redhat发布了最初的NPTL。NPTL最开始在redhat linux 9里发布,现在从RHEL3起内核2.6起都支持NPTL,而且完全成了GNU C库的一部分。 设计NPTL使用了跟LinuxThread雷同的办法,在内核里面线程仍然被当作是一个历程,而且仍然使用了clone()体系调用(在NPTL库里调用)。但是,NPTL需要内核级的特殊支持来实现,比如需要挂起然后再唤醒线程的线程同步原语futex.NPTL也是一个1*1的线程库,就是说,当你使用pthread_create()调用创建一个线程后,在内核里就相应创建了一个调理实体,在linux里就是一个新历程,这个方法最大大概的简化了线程的实现。除NPTL的1*1模型外还有一个m*n模型,通常这种模型的用户线程数会比内核的调理实体多。在这种实现里,线程库自己必须去处理大概存在的调理,这样在线程库内部的上下文切换通常都会相称的快,因为它克制了体系调用转到内核态。然而这种模型增加了线程实现的复杂性,并大概出现诸如优先级反转的问题,此外,用户态的调理怎样跟内核态的调理进行调和也是很难让人满足。
复制代码
介绍线程和python

理论知识

全局表明器锁GIL

  Python代码的执行由Python假造机(也叫表明器主循环)来控制。Python在设计之初就思量到要在主循环中,同时只有一个线程在执行。虽然 Python 表明器中可以“运行”多个线程,但在恣意时刻只有一个线程在表明器中运行。
  对Python假造机的访问由全局表明器锁(GIL)来控制,正是这个锁能保证同一时刻只有一个线程在运行。
  在多线程情况中,Python 假造机按以下方式执行:
  a、设置 GIL;
  b、切换到一个线程去运行;
  c、运行指定命量的字节码指令或者线程自动让出控制(可以调用 time.sleep(0));
  d、把线程设置为睡眠状态;
  e、解锁 GIL;
  d、再次重复以上所有步骤。
  在调用外部代码(如 C/C++扩展函数)的时候,GIL将会被锁定,直到这个函数结束为止(由于在这期间没有Python的字节码被运行,以是不会做线程切换)编写扩展的程序员可以自动解锁GIL。
python线程模块的选择

  Python提供了几个用于多线程编程的模块,包罗thread、threading和Queue等。thread和threading模块允许程序员创建和管理线程。thread模块提供了根本的线程和锁的支持,threading提供了更高级别、功能更强的线程管理的功能。Queue模块允许用户创建一个可以用于多个线程之间共享数据的队列数据布局。
  克制使用thread模块,因为更高级别的threading模块更为先辈,对线程的支持更为完善,而且使用thread模块里的属性有大概会与threading出现冲突;其次低级别的thread模块的同步原语很少(实际上只有一个),而threading模块则有很多;再者,thread模块中当主线程结束时,所有的线程都会被强制结束掉,没有告诫也不会有正常的清除工作,至少threading模块能确保重要的子线程退出后历程才退出。
  thread模块不支持保卫线程,当主线程退出时,所有的子线程岂论它们是否还在工作,都会被强行退出。而threading模块支持保卫线程,保卫线程一般是一个等候客户请求的服务器,如果没有客户提出请求它就在那等着,如果设定一个线程为保卫线程,就表现这个线程是不重要的,在历程退出的时候,不用等候这个线程退出。
threading模块

multiprocess模块的完全模仿了threading模块的接口,二者在使用层面,有很大的相似性,因而不再详细介绍(官方链接
线程的创建Threading.Thread类

线程的创建

  1. 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
  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多线程与多历程

  1. 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的比较
  1. 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    '''
复制代码
开启服从的比力
  1. 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,因为同一历程内的线程之间共享历程内的数据同一历程内的线程共享该历程的数据?
复制代码
内存数据的共享问题训练 :多线程实现socket

  1. #_*_coding:utf-8_*_#!/usr/bin/env pythonimport multiprocessingimport threadingimport sockets=socket.socket(socket.AF_INET,socket.SOCK_STREAM)s.bind(('127.0.0.1',8080))s.listen(5)def action(conn):    while True:        data=conn.recv(1024)        print(data)        conn.send(data.upper())if __name__ == '__main__':    while True:        conn,addr=s.accept()        p=threading.Thread(target=action,args=(conn,))        p.start()
复制代码
server
  1. #_*_coding:utf-8_*_#!/usr/bin/env pythonimport sockets=socket.socket(socket.AF_INET,socket.SOCK_STREAM)s.connect(('127.0.0.1',8080))while True:    msg=input('>>: ').strip()    if not msg:continue    s.send(msg.encode('utf-8'))    data=s.recv(1024)    print(data)
复制代码
clientThread类的其他方法

  1. Thread实例对象的方法  # isAlive(): 返回线程是否活动的。  # getName(): 返回线程名。  # setName(): 设置线程名。threading模块提供的一些方法:  # threading.currentThread(): 返回当前的线程变量。  # threading.enumerate(): 返回一个包罗正在运行的线程的list。正在运行指线程启动后、结束前,不包罗启动前和终止后的线程。  # threading.activeCount(): 返回正在运行的线程数量,与len(threading.enumerate())有雷同的结果。
复制代码
  1. from threading import Threadimport threadingfrom multiprocessing import Processimport osdef work():    import time    time.sleep(3)    print(threading.current_thread().getName())if __name__ == '__main__':    #在主历程下开启线程    t=Thread(target=work)    t.start()    print(threading.current_thread().getName())    print(threading.current_thread()) #主线程    print(threading.enumerate()) #连同主线程在内有两个运行的线程    print(threading.active_count())    print('主线程/主历程')    '''    打印结果:    MainThread        [, ]    主线程/主历程    Thread-1    '''
复制代码
代码示例
  1. 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()    t.join()    print('主线程')    print(t.is_alive())    '''    egon say hello    主线程    false    '''
复制代码
join方法保卫线程

无论是历程照旧线程,都遵循:保卫xx会等候主xx运行完毕后被销毁。需要强调的是:运行完毕并非终止运行
  1. #1.对主历程来说,运行完毕指的是主历程代码运行完毕#2.对主线程来说,运行完毕指的是主线程所在的历程内所有非保卫线程统统运行完毕,主线程才算运行完毕
复制代码
  1. #1 主历程在其代码结束后就已经算运行完毕了(保卫历程在此时就被接纳),然后主历程会一直等非保卫的子历程都运行完毕后接纳子历程的资源(否则会产生僵尸历程),才会结束,#2 主线程在其他非保卫线程运行完毕后才算运行完毕(保卫线程在此时就被接纳)。因为主线程的结束意味着历程的结束,历程团体的资源都将被接纳,而历程必须保证非保卫线程都运行完毕后才气结束。
复制代码
详细表明
  1. 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
  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

锁与GIL


同步锁

  1. from threading import Threadimport os,timedef work():    global n    temp=n    time.sleep(0.1)    n=temp-1if __name__ == '__main__':    n=100    l=[]    for i in range(100):        p=Thread(target=work)        l.append(p)        p.start()    for p in l:        p.join()    print(n) #结果大概为99
复制代码
多个线程抢占资源的情况
  1. import threadingR=threading.Lock()R.acquire()'''对公共数据的操纵'''R.release()
复制代码
  1. from threading import Thread,Lockimport os,timedef work():    global n    lock.acquire()    temp=n    time.sleep(0.1)    n=temp-1    lock.release()if __name__ == '__main__':    lock=Lock()    n=100    l=[]    for i in range(100):        p=Thread(target=work)        l.append(p)        p.start()    for p in l:        p.join()    print(n) #结果肯定为0,由原来的并发执行变成串行,牺牲了执行服从保证了数据安全
复制代码
同步锁的引用
  1. #不加锁:并发执行,速率快,数据不安全from threading import current_thread,Thread,Lockimport os,timedef task():    global n    print('%s is running' %current_thread().getName())    temp=n    time.sleep(0.5)    n=temp-1if __name__ == '__main__':    n=100    lock=Lock()    threads=[]    start_time=time.time()    for i in range(100):        t=Thread(target=task)        threads.append(t)        t.start()    for t in threads:        t.join()    stop_time=time.time()    print('主:%s n:%s' %(stop_time-start_time,n))'''Thread-1 is runningThread-2 is running......Thread-100 is running主:0.5216062068939209 n:99'''#不加锁:未加锁部分并发执行,加锁部分串行执行,速率慢,数据安全from threading import current_thread,Thread,Lockimport os,timedef task():    #未加锁的代码并发运行    time.sleep(3)    print('%s start to run' %current_thread().getName())    global n    #加锁的代码串行运行    lock.acquire()    temp=n    time.sleep(0.5)    n=temp-1    lock.release()if __name__ == '__main__':    n=100    lock=Lock()    threads=[]    start_time=time.time()    for i in range(100):        t=Thread(target=task)        threads.append(t)        t.start()    for t in threads:        t.join()    stop_time=time.time()    print('主:%s n:%s' %(stop_time-start_time,n))'''Thread-1 is runningThread-2 is running......Thread-100 is running主:53.294203758239746 n:0'''#有的同学大概有疑问:既然加锁会让运行变成串行,那么我在start之后立纵然用join,就不用加锁了啊,也是串行的结果啊#没错:在start之后立即使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的#单从保证数据安全方面,二者都可以实现,但很明显是加锁的服从更高.from threading import current_thread,Thread,Lockimport os,timedef task():    time.sleep(3)    print('%s start to run' %current_thread().getName())    global n    temp=n    time.sleep(0.5)    n=temp-1if __name__ == '__main__':    n=100    lock=Lock()    start_time=time.time()    for i in range(100):        t=Thread(target=task)        t.start()        t.join()    stop_time=time.time()    print('主:%s n:%s' %(stop_time-start_time,n))'''Thread-1 start to runThread-2 start to run......Thread-100 start to run主:350.6937336921692 n:0 #耗时是多么的可怕''')
复制代码
互斥锁与join的区别死锁与递归锁

历程也有死锁与递归锁,在历程那里忘记说了,放到这里统统说了额
所谓死锁: 是指两个或两个以上的历程或线程在执行过程中,因争夺资源而造成的一种互相称候的现象,若无外力作用,它们都将无法推进下去。此时称体系处于死锁状态或体系产生了死锁,这些永久在互相称候的历程称为死锁历程,如下就是死锁
  1. from threading import Lock as Lockimport timemutexA=Lock()mutexA.acquire()mutexA.acquire()print(123)mutexA.release()mutexA.release()
复制代码
死锁解决方法,递归锁,在Python中为了支持在同一线程中多次请求同一资源,python提供了可重入锁RLock。
这个RLock内部维护着一个Lock和一个counter变量,counter记录了acquire的次数,从而使得资源可以被多次require。直到一个线程所有的acquire都被release,其他的线程才气获得资源。上面的例子如果使用RLock代替Lock,则不会发生死锁:
  1. from threading import RLock as Lockimport timemutexA=Lock()mutexA.acquire()mutexA.acquire()print(123)mutexA.release()mutexA.release()
复制代码
递归锁RLock典型问题:科学家吃面
  1. 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)    fork_lock.release()    noodle_lock.release()def eat2(name):    fork_lock.acquire()    print('%s 抢到了叉子' % name)    time.sleep(1)    noodle_lock.acquire()    print('%s 抢到了面条' % name)    print('%s 吃面' % name)    noodle_lock.release()    fork_lock.release()for name in ['哪吒','egon','yuan']:    t1 = Thread(target=eat1,args=(name,))    t2 = Thread(target=eat2,args=(name,))    t1.start()    t2.start()
复制代码
死锁问题
  1. import timefrom threading import Thread,RLockfork_lock = noodle_lock = RLock()def eat1(name):    noodle_lock.acquire()    print('%s 抢到了面条'%name)    fork_lock.acquire()    print('%s 抢到了叉子'%name)    print('%s 吃面'%name)    fork_lock.release()    noodle_lock.release()def eat2(name):    fork_lock.acquire()    print('%s 抢到了叉子' % name)    time.sleep(1)    noodle_lock.acquire()    print('%s 抢到了面条' % name)    print('%s 吃面' % name)    noodle_lock.release()    fork_lock.release()for name in ['哪吒','egon','yuan']:    t1 = Thread(target=eat1,args=(name,))    t2 = Thread(target=eat2,args=(name,))    t1.start()    t2.start()
复制代码
递归锁解决死锁问题线程队列

queue队列 :使用import queue,用法与历程Queue一样
queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.
class queue.Queue(maxsize=0) #先辈先出
  1. import queueq=queue.Queue()q.put('first')q.put('second')q.put('third')print(q.get())print(q.get())print(q.get())'''结果(先辈先出):firstsecondthird'''
复制代码
先辈先出class queue.LifoQueue(maxsize=0) #last in fisrt out
  1. import queueq=queue.LifoQueue()q.put('first')q.put('second')q.put('third')print(q.get())print(q.get())print(q.get())'''结果(后进先出):thirdsecondfirst'''
复制代码
后进先出class queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列
  1. import queueq=queue.PriorityQueue()#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以黑白数字之间的比较),数字越小优先级越高q.put((20,'a'))q.put((10,'b'))q.put((30,'c'))print(q.get())print(q.get())print(q.get())'''结果(数字越小优先级越高,优先级高的优先出队):(10, 'b')(20, 'a')(30, 'c')'''
复制代码
优先级队列
  1. Constructor for a priority queue. maxsize is an integer that sets the upperbound limit on the number of items that can be placed in the queue. Insertion will block once this size has been reached, until queue items are consumed. If maxsize is less than or equal to zero, the queue size is infinite.The lowest valued entries are retrieved first (the lowest valued entry is the one returned by sorted(list(entries))[0]). A typical pattern for entries is a tuple in the form: (priority_number, data).exception queue.EmptyException raised when non-blocking get() (or get_nowait()) is called on a Queue object which is empty.exception queue.FullException raised when non-blocking put() (or put_nowait()) is called on a Queue object which is full.Queue.qsize()Queue.empty() #return True if empty  Queue.full() # return True if full Queue.put(item, block=True, timeout=None)Put item into the queue. If optional args block is true and timeout is None (the default), block if necessary until a free slot is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Full exception if no free slot was available within that time. Otherwise (block is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (timeout is ignored in that case).Queue.put_nowait(item)Equivalent to put(item, False).Queue.get(block=True, timeout=None)Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Empty exception if no item was available within that time. Otherwise (block is false), return an item if one is immediately available, else raise the Empty exception (timeout is ignored in that case).Queue.get_nowait()Equivalent to get(False).Two methods are offered to support tracking whether enqueued tasks have been fully processed by daemon consumer threads.Queue.task_done()Indicate that a formerly enqueued task is complete. Used by queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue).Raises a ValueError if called more times than there were items placed in the queue.Queue.join() block直到queue被消费完毕
复制代码
更多方法说明Python标准模块--concurrent.futures

https://docs.python.org/dev/library/concurrent.futures.html
  1. #1 介绍concurrent.futures模块提供了高度封装的异步调用接口ThreadPoolExecutor:线程池,提供异步调用ProcessPoolExecutor: 历程池,提供异步调用Both implement the same interface, which is defined by the abstract Executor class.#2 根本方法#submit(fn, *args, **kwargs)异步提交任务#map(func, *iterables, timeout=None, chunksize=1) 取代for循环submit的操纵#shutdown(wait=True) 相称于历程池的pool.close()+pool.join()操纵wait=True,等候池内所有任务执行完毕接纳完资源后才继承wait=False,立即返回,并不会等候池内的任务执行完毕但不管wait参数为何值,整个程序都会比及所有任务执行完毕submit和map必须在shutdown之前#result(timeout=None)取得结果#add_done_callback(fn)回调函数
  2. # done()
  3. 判定某一个线程是否完成
  4. # cancle()
  5. 取消某个任务
复制代码
  1. #介绍The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls asynchronously. ProcessPoolExecutor uses the multiprocessing module, which allows it to side-step the Global Interpreter Lock but also means that only picklable objects can be executed and returned.class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None)An Executor subclass that executes calls asynchronously using a pool of at most max_workers processes. If max_workers is None or not given, it will default to the number of processors on the machine. If max_workers is lower or equal to 0, then a ValueError will be raised.#用法from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorimport os,time,randomdef task(n):    print('%s is runing' %os.getpid())    time.sleep(random.randint(1,3))    return n**2if __name__ == '__main__':    executor=ProcessPoolExecutor(max_workers=3)    futures=[]    for i in range(11):        future=executor.submit(task,i)        futures.append(future)    executor.shutdown(True)    print('+++>')    for future in futures:        print(future.result())
复制代码
ProcessPoolExecutor
  1. #介绍ThreadPoolExecutor is an Executor subclass that uses a pool of threads to execute calls asynchronously.class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='')An Executor subclass that uses a pool of at most max_workers threads to execute calls asynchronously.Changed in version 3.5: If max_workers is None or not given, it will default to the number of processors on the machine, multiplied by 5, assuming that ThreadPoolExecutor is often used to overlap I/O instead of CPU work and the number of workers should be higher than the number of workers for ProcessPoolExecutor.New in version 3.6: The thread_name_prefix argument was added to allow users to control the threading.Thread names for worker threads created by the pool for easier debugging.#用法与ProcessPoolExecutor雷同
复制代码
ThreadPoolExecutor
  1. from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorimport os,time,randomdef task(n):    print('%s is runing' %os.getpid())    time.sleep(random.randint(1,3))    return n**2if __name__ == '__main__':    executor=ThreadPoolExecutor(max_workers=3)    # for i in range(11):    #     future=executor.submit(task,i)    executor.map(task,range(1,12)) #map取代了for+submit
复制代码
map的用法
  1. from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorfrom multiprocessing import Poolimport requestsimport jsonimport osdef get_page(url):    print(' get %s' %(os.getpid(),url))    respone=requests.get(url)    if respone.status_code == 200:        return {'url':url,'text':respone.text}def parse_page(res):    res=res.result()    print(' parse %s' %(os.getpid(),res['url']))    parse_res='url: size:[%s]\n' %(res['url'],len(res['text']))    with open('db.txt','a') as f:        f.write(parse_res)if __name__ == '__main__':    urls=[        'https://www.baidu.com',        'https://www.python.org',        'https://www.openstack.org',        'https://help.github.com/',        'http://www.sina.com.cn/'    ]    # p=Pool(3)    # for url in urls:    #     p.apply_async(get_page,args=(url,),callback=pasrse_page)    # p.close()    # p.join()    p=ProcessPoolExecutor(3)    for url in urls:        p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj,需要用obj.result()拿到结果
复制代码
回调函数

本帖子中包含更多资源

您需要 登录 才可以下载或查看,没有帐号?用户注册

x

天涯海角也要找到Ni:Python_线程

中发现Ni: Python_线程
中发现Ni: Python_线程
中发现Ni: Python_线程
中发现Ni: Python_线程
中发现Ni: Python_线程
中发现Ni: Python_线程
相关技术服务需求,请联系管理员和客服QQ:2753533861或QQ:619920289
您需要登录后才可以回帖 登录 | 用户注册

本版积分规则

帖子推荐:
客服咨询

QQ:2753533861

服务时间 9:00-22:00

快速回复 返回顶部 返回列表