python多线程

墨蓝 2022-10-31 13:42 223阅读 0赞

# 多线程(mult-tHreadings) #

## 目录 ##

### 什么是多线程 ###

### 添加线程 add thread ###

### join功能 ###

### Queue功能 ###

### 不一定有效率 GIL ###

### 锁 lock ###

## 什么是多线程 ##

可以分配任务为一个个的小任务让电脑同一时间完成任务。将每一个小任务放在一个线程里运算，所有线程同时计算工作，当所有线程的子任务同时完成后，总任务也被完成，从而大大的降低计算时间，比如我们可以在同一时间内运行多个程序，运行多个函数。多线程是加速程序计算的有效方式，Python的多线程模块threading上手快速简单，从这节开始我们就教大家如何使用它。

## 添加线程 add thread ##

本节我们来学习threading模块的一些基本操作，如获取线程数，添加线程等。

### 首先别忘了导入模块： ###

import threading

### threading模块内常见函数 ###

获取已激活的线程数
    threading.active_count()
    #查看所有线程信息
    threading.enumerate()
    #查看现在正在运行的线程
    threading.current_thread()

#### 实例 ####

def main():
        print(threading.active_count())
        print(threading.enumerate())
        print(threading.current_thread())
        
    if __name__ == '__main__':
        main()

#### 运行结果： ####

> 2
> 
> \[<\_MainThread(MainThread, started 140736011932608)>, <Thread(SockThread, started daemon 123145376751616)>\]
> 
> <\_MainThread(MainThread, started 140736011932608)>

#### 结果分析： ####

> 输出的所有的线程结果是一个<\_MainThread(…)>带多个<Thread(…)>，现在一共两个线程，但是现在只有MainThread即主程序运行所需的线程正在工作其他线程都在空闲。

### 添加线程 ###

threading.Thread()接收参数target代表这个线程要完成的任务，需自行定义。我们可以定义一个thread\_job()函数作为任务，我们需要将将这个功能的名称作为参数传递给target，代表了我们创建了一个线程，这个线程的任务就是去完整成函数名为thread\_job()的函数所描述的功能。

这样我们就定义好了一个线程，但是如果想要线程真正的开始工作我们需要使用thread.start()函数来启动线程，开始完成线程指定的任务。

同时我们可以使用thread\_job()函数中的name参数来设定线程的名称而取代系统自己取定的名称。

同时我们可以使用thread\_job()函数中的args参数来设定线程所要完成功能所需的参数，参数以元组的形式按找调用功能形参表对应顺序排列起来，即args = (arg1,arg2,arg3,……)

def thread_job():
        print('This is a thread of %s' % threading.current_thread())
    
    def main():
        thread = threading.Thread(target=thread_job,)   # 定义线程 
        thread.start()  # 让线程开始工作
        
    if __name__ == '__main__':
        main()

#### 运行结果： ####

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-bZNjdqHt-1614003422526)(1322EA0D9A604314803A0A3822B09EB9)\]

#### 运行分析： ####

每增加一个线程我们就输出一下该线程的名称

## join 功能 ##

### 不加 join() 的结果 ###

我们让T1线程工作的耗时增加，让他循环十次，每一步的时候都暂停0.1秒什么都不干，休息或睡一秒，所以引入函数time.sleep(0.1)

import threading
    import time
    
    def thread_job():
        print("T1 start\n")
        for i in range(10):
            time.sleep(0.1) # 任务间隔0.1s
        print("T1 finish\n")
    
    added_thread = threading.Thread(target=thread_job, name='T1')
    added_thread.start()
    print("all done\n")

#### 预想中输出的结果是否为： ####

> T1 start
> 
> T1 finish
> 
> all done

#### 但实际却是： ####

> T1 start
> 
> all done
> 
> T1 finish

#### 结果分析： ####

运行结果来看，我们运行完语句added\_thread.start()后，直接跳下来执行print(“all done\\n”)，而并未等待thread\_job()函数在线程T1运行结束后，在运行语句print(“all done\\n”),证明多线程是在同时运行的，我们的主函数所在的主线程和我们调用执行thread\_job()函数的T1线程是同时运行的，而不存在先后顺序。而我们有的时候希望所有线程或部分线程运行结束之后再执行我们调用函数所在的线程(这里为主线程)，我们就需要用到join语句

### 加入 join() 的结果 ###

线程任务还未完成便输出all done。如果要遵循顺序，可以在启动线程后对它调用join：

added_thread.start()
    added_thread.join()
    print("all done\n")

这样的意思就是当只有当我们调用的线程上的任务完成之后才会接着执行调用函数所在的线程。

#### 运行结果： ####

> T1 start
> 
> T1 finish
> 
> all done

### join控制多线程执行顺序 ###

#### 使用join对控制多个线程的执行顺序非常关键，join函数类似是将我们想要join的线程进行插队，暂时不执行我们正在执行的线程而让join的线程插队进来先执行，当插队完成任务之后再执行我们当前正常的线程。举个例子，假设我们现在再加一个线程T2，T2的任务量较小，会比T1更快完成： ####

def T1_job():
        print("T1 start\n")
        for i in range(10):
            time.sleep(0.1)
        print("T1 finish\n")
    
    def T2_job():
        print("T2 start\n")
        print("T2 finish\n")
    
    thread_1 = threading.Thread(target=T1_job, name='T1')
    thread_2 = threading.Thread(target=T2_job, name='T2')
    thread_1.start() # 开启T1
    thread_2.start() # 开启T2
    print("all done\n")

#### 输出的一种结果是： ####

T1 start

T2 start

all done

T2 finish

T1 finish

> 现在T1和T2都没有join，所以在开启两个线程以后，主线程接着运行主函数中的语句输出all done，又由于T2线程中执行的函数任务较少，所需时间更多，所以会比T1线程更早完成。

#### 输出的另一种结果是： ####

T1 start

T2 start

T2 finish

all done

T1 finish

> 注意这里说一种是因为all done的出现完全取决于两个线程的执行速度， 完全有可能T2 finish出现在all done之后。这种杂乱的执行方式是我们不能忍受的，因此要使用join加以控制。

#### 我们试试在T1启动后，T2启动前加上thread\_1.join(): ####

thread_1.start()
    thread_1.join() # notice the difference!
    thread_2.start()
    print("all done\n")

输出结果：

T1 start

T2 start

T2 finish

T1 finish

all done

可以看到，我们的主程序会在T1线程被调用结束后输出语句all done，而T2的线程运行时间比T1短所一会先输出T2 finish，接着输出T1 finish，在T1线程结束之后执行主程序当中的all done。

#### 如果我们在T2启动后放上thread\_1.join()会怎么样呢？ ####

thread\_1.start()  
thread\_2.start()  
thread\_1.join() \# notice the difference!  
print(“all done\\n”)  
输出结果：

T1 start

T2 start

T2 finish

all done

T1 finish

T2在T1之后启动，并且因为T2任务量小会在T1之前完成；而T2也因为加了join，all done在它完成后才显示。而此时T1线程还未结束，所以最后输出T1 finish

#### 所线程同时join ####

为了规避不必要的麻烦，推荐如下这种1221的V型排布：

thread\_1.start() \# start T1  
thread\_2.start() \# start T2  
thread\_2.join() \# join for T2  
thread\_1.join() \# join for T1  
print(“all done\\n”)

输出结果：

T1 start

T2 start

T2 finish

T1 finish

all done

## queue 功能 ##

因为使用多线程调用功能是没有办法得到返回值的，也就是无法获得带用函数的return值，所以我们需要把运算得到的结果放在一个长的队列当中。将数据列表中的数据传入，使用多个个线程处理，将结果保存在Queue中，线程执行完后，从Queue中获取存储的结果，从而替代return功能。

import threading
    import time
    
    from queue import Queue
    
    #函数的参数是一个列表l和一个队列q，函数的功能是，对列表的每个元素进行平方计算，将结果保存在队列中
    def job(l,q):
        for i in range (len(l)):
            l[i] = l[i]**2
        q.put(l)   #多线程调用的函数不能用return返回值
    
    #在多线程函数中定义一个Queue，用来保存返回值，代替return，定义一个多线程列表，初始化一个多维数据列表
    def multithreading():
        q =Queue() #  #q中存放返回值，代替return的返回值
        threads = []
        data = [[1,2,3],[3,4,5],[4,4,4],[5,5,5]]
    #在多线程函数中定义四个线程，启动线程，将每个线程添加到多线程的列表中
        for i in range(4):#定义四个线程
            t = threading.Thread(target=job,args=(data[i],q))
            #Thread首字母要大写，被调用的job函数没有括号，只是一个索引，参数在后面
            t.start()#开始线程
            threads.append(t)#把每个线程append到线程列表中
    
    #该函数可以保证在所有线程结束之后，再从结果queue中依次提取结果数据存储到results列表当中，如果不这样操作可能会导致，线程中调用的函数还没执行完毕并将结果存在queue当中，你就开始取值会得到空结果
    #并且该join不能合并放在第一个循环之中，这样分开循环可以使线程并发的同步执行，如果你每开始一个线程就插入到当前线程，则循环会暂时停止等到你的线程结束之后，在开始定义下一个线程并开始，相当于四个子线程依次执行
        for thread in threads:
            #分别join四个线程插入到主线程
            thread.join()
    
        #定义一个空的列表results，将四个线运行后保存在队列中的结果返回给空列表results
        results = []
        for _ in range(4):
            results.append(q.get())#q.get()按顺序从q中拿出一个值
        print(results)
    
    if __name__=='__main__':
        multithreading()

以上我们即完成把四批数据输入到不同线程中计算平方，并且将每一批数据的结果放在queue当中存储，最后提取出来。

## GIL 不一定有效率 ##

python 的多线程 threading 有时候并不是特别理想. 最主要的原因是就是, Python 的设计上, 有一个必要的环节, 就是 Global Interpreter Lock (GIL：全局排他所锁). 这个东西让 Python 还是一次性只能处理一个东西.所以虽然理论上讲我们做一个东西需要10s，开了五个线程则就需要2s，但是其实不然，其实python的多线程本质上也是同一时间以内只有一个线程在工作，只不过是多个线程时，线程之间交替运行，导致一种人们感觉他们在同时运行的假象。

我从这里摘抄了一段对于 GIL 的解释.

尽管Python完全支持多线程编程， 但是解释器的C语言实现部分在完全并行执行时并不是线程安全的。 实际上，解释器被一个全局解释器锁保护着，它确保任何时候都只有一个Python线程执行。 GIL最大的问题就是Python的多线程程序并不能利用多核CPU的优势 （比如一个使用了多个线程的计算密集型程序只会在一个单CPU上面运行）。

在讨论普通的GIL之前，有一点要强调的是GIL只会影响到那些严重依赖CPU的程序（比如计算型的）。 如果你的程序大部分只会涉及到I/O，比如网络交互，那么使用多线程就很合适， 因为它们大部分时间都在等待。实际上，你完全可以放心的创建几千个Python线程， 现代操作系统运行这么多线程没有任何压力，没啥可担心的。

测试 GIL  
我们创建一个 job, 分别用 threading 和 一般的方式执行这段程序. 并且创建一个 list 来存放我们要处理的数据. 在 Normal 的时候, 我们这个 list 扩展4倍, 在 threading 的时候, 我们建立4个线程, 并对运行时间进行对比.

import threading
    from queue import Queue
    import copy
    import time
    
    def job(l, q):
        res = sum(l)
        q.put(res)
    
    def multithreading(l):
        q = Queue()
        threads = []
        for i in range(4):
            t = threading.Thread(target=job, args=(copy.copy(l), q), name='T%i' % i)
            t.start()
            threads.append(t)
        [t.join() for t in threads]
        total = 0
        for _ in range(4):
            total += q.get()
        print(total)
    
    def normal(l):
        total = sum(l)
        print(total)
    
    if __name__ == '__main__':
        l = list(range(1000000))
        s_t = time.time()
        normal(l*4)
        print('normal: ',time.time()-s_t)
        s_t = time.time()
        multithreading(l)
        print('multithreading: ', time.time()-s_t)

运行结果：  
1999998000000  
normal: 0.10034608840942383  
1999998000000  
multithreading: 0.08421492576599121

如果你成功运行整套程序, 你大概会有这样的输出. 我们的运算结果没错, 所以程序 threading 和 Normal 运行了一样多次的运算. 但是我们发现 threading 却没有快多少, 按理来说, 我们预期会要快3-4倍, 因为有建立4个线程, 但是并没有. 这就是其中的 GIL 在作怪.

### 总结 ###

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-N5oG8ptZ-1614003422528)(65B4B47F30904CCEADAA825E62CC278F)\]  
比如我们有三个线程，当第一个线程进行完成了任务一的一部分运算的时候，GIL会释放线程1，即线程1失去GIL，并且进入新一轮数据的输入，此时线程2获得GIL，获得运算和禁止其他线程运算的权利，当他运算一部分任务2之后也将会失去GIL，进行下一轮数据的补充与本轮运算结果的输出，相似的线程3获得GIL，三个线程循环交替执行知道所有线程的任务完成。由此可见多线程完成任务所需时间不会有大量改变，减少的少量时间来自于数据输入与输出的时候同时开启工作而减少的时间。只有只有针对特定的功能多线才可能节约大量的时间，比如在聊天是发送与接受数据的多线程会大量减少时间于两者同用一个线程相比，但是如果你是有一大堆的数据要处理，你把它们分份之后放在多个线程中执行希望大量节省时间，那就是有点痴心妄想了。真正的可以实现完全独立并行运算，我们需要采用多核处理，因为每一个核内有独立的逻辑空间，可以独立的计算处理数据，才能真正的并行，进而提高计算效率。

## 线程锁 Lock ##

当我们想用第一个线程调用功能对数据处理完得到一个初步的结果，再将这个结果作为输入交给第二个线程进行处理的时候，我们可能需要用到线程锁(Lock)，先将第一个线程锁住，等待它处理完，我再开始第二个线程接着处理。我们一般对共享内存的数据进行加工处理的时候才会用到线程锁。

### 不使用 Lock 的情况 ###

完整代码：
    导入线程标准模块
    import threading
    
    函数一：全局变量A的值每次加1，循环10次，并打印
    
    def job1():
        global A
        for i in range(10):
            A+=1
            print('job1',A)
    函数二：全局变量A的值每次加10，循环10次，并打印
    
    def job2():
        global A
        for i in range(10):
            A+=10
            print('job2',A)
    
    主函数：定义全局变量A=0,定义两个线程，分别执行函数一和函数二
    
    if __name__== '__main__':
        A=0
        t1=threading.Thread(target=job1)
        t2=threading.Thread(target=job2)
        t1.start()
        t2.start()

运行结果：

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fQgaCKJl-1614003422530)(822D08D4D5514A0592B615FA91482D5B)\]  
结果十分混乱就是两个线程同时执行，A随时改变，A改变成什么样我们就在什么样A上运算并输出。我们希望将所有的job1都执行完事之后即每次加1并输出，然后再执行job2，我们需要使用线程锁。

### 使用 Lock 的情况 ###

lock在不同线程使用同一共享内存时，能够确保线程之间互不影响，使用lock的方法是，我们首先需要定义一个线程锁，我们可以定义成global的形式然后再每个函数内部声明全局变量，或者直接在函数内部定义一个线程锁的局部变量，然后在我们希望锁定的代码块，也就是想让当前线程不切换，锁定执行直到运行结束的代码块之前，即 在每个线程执行运算修改共享内存之前，执行lock.acquire()将共享内存上锁， 确保当前线程执行时，内存不会被其他线程访问，执行运算完毕后，使用lock.release()将锁打开， 保证其他的线程可以使用该共享内存。

完整的代码：

import threading
    函数一和函数二加锁
    
    def job1():
        global A,lock
        lock.acquire()
        for i in range(10):
            A+=1
            print('job1',A)
        lock.release()
    
    def job2():
        global A,lock
        lock.acquire()
        for i in range(10):
            A+=10
            print('job2',A)
        lock.release()
    主函数中定义一个Lock
    
    if __name__== '__main__':
        lock=threading.Lock()
        A=0
        t1=threading.Thread(target=job1)
        t2=threading.Thread(target=job2)
        t1.start()
        t2.start()
        t1.join()
        t2.join()

运行结果：

job1 1  
job1 2  
job1 3  
job1 4  
job1 5  
job1 6  
job1 7  
job1 8  
job1 9  
job1 10  
job2 20  
job2 30  
job2 40  
job2 50  
job2 60  
job2 70  
job2 80  
job2 90  
job2 100  
job2 110  
从打印结果来看，使用lock后，第一个线程中的循环结束后才执行第二个线程中的循环。使用lock和不使用lock，最后打印输出的结果是不同的。

> join相当于将某个线程插入到某个线程之前，必须先执行插入的，lock是上锁，锁内临界区的资源只有自己当前线程可以占用，而其他线程无法占用，只有当release之后资源就可以共享给别人去加载了。