Python多线程概念理解

搞任何编程语言肯定都离不开多线程，Python肯定也不例外，虽然Python的多线程并不完美，但还是经常会用到滴，尤其是网络爬虫这些东西，效率很很重要，所以我就屁颠屁颠的来写这篇文章啦！

多线程是为了同步完成多项任务，通过提高资源使用效率来提高系统的效率，线程是在同一时间需要完成多项任务的时候实现的。
多线程就像火车的每一节车厢，而进程则是火车，车厢离开火车是无法跑动的，火车可以有多节车厢，多线程的出现就是为了提高效率，同时使用它也会带来了一些问题。

threading模块

threading模块是python中专门提供用来做多线程编程的模块，threading模块中最常用的类是Thread

import threading import time def coding(): for x in range(3): print('%s正在写代码' % x) time.sleep(1) def drawing(): for x in range(3): print('%s正在画图' % x) time.sleep(1) def single_thread(): # 单线程 coding() drawing() def multi_thread(): print('多线程开始执行') t1 = threading.Thread(target=coding) t2 = threading.Thread(target=drawing) t1.start() t2.start() t1.join() t2.join() print('多线程执行完毕') if __name__ == '__main__': multi_thread() 

查看线程数

使用threading.enumerate()函数便可以看到当前线程的数量

查看当前线程的名字

使用threading.current_thread()可以看到当前线程的信息

继承自threading.Thread类

为了让线程代码更好的封装，可以使用threading模块下的Thread类，继承自这个类，然后实现run方法，线程就会自动运行run方法中的代码

import threading import time class CodingThread(threading.Thread): # 继承Thread类并将代码直接写在run函数里 name = '铁牛' def run(self): for x in range(3): # print('%s正在写代码' % threading.current_thread()) print(f'{threading.current_thread()}  {CodingThread.name} 正在写代码') time.sleep(1) class DrawingThread(threading.Thread): name = '二蛋' def run(self): for x in range(3): # print('%s正在画图' % threading.current_thread()) print(f'{threading.current_thread()}  {DrawingThread.name} 正在画画') time.sleep(1) def multi_thread(): t1 = CodingThread() t2 = DrawingThread() t1.start() t2.start() if __name__ == '__main__': multi_thread() 

多线程共享全局变量的问题

多线程都是在同一个进程中运行的，因此在进程中的全局变量所有线程都是可共享的，这就造成了一个问题，因为线程执行的顺序是无序的，所以有可能会造成数据错误

import threading

tickets = 0 def get_ticket(): global tickets # 函数内部调用并修改外面的全局变量需要用global定义 for x in range(1000000): tickets += 1 print('tickets:%d' % tickets) def main(): for x in range(2): # 遍历两条线程 t = threading.Thread(target=get_ticket) t.start() if __name__ == '__main__': main() # 结果正常来讲应该是6，但是因为多线程运行的不确定性，因此最后的结果可能是随机的 

锁机制

为了解决共享全局变量的问题，threading提供了一个Lock类，这个类可以在某个线程访问某个变量的时候加锁，其他线程此时就不能进来，直到当前线程处理完后，把锁释放了，其他线程才能进来处理

import threading

VALUE = 0 gLock = threading.Lock() def add_value(): global VALUE # 函数内部调用并修改外面的全局变量需要用global定义 gLock.acquire() for x in range(1000000): VALUE += 1 # 锁一般加在多条线程修改全局变量的的地方 gLock.release() print('value：%d' % VALUE) # 不加锁最后打印的结果是错的 def main(): for x in range(2): t = threading.Thread(target=add_value) t.start() if __name__ == '__main__': main() 

生产者和消费者模式

生产者和消费者模式是多线程开发中经常见到的一种模式，生产者的线程专门用来生产一些数据，然后存放到一个中间的变量中，消费者再从这个中间的变量中取出数据进行消费，但是因为要使用中间变量，中间变量经常是一些全局变量，因此需要使用锁来保证数据完整性

Lock版本

使用threading.Lock锁实现生产者与消费者模式

import threading import random import time

gMoney = 1000 # 总金额初始值为1000 gLock = threading.Lock() # 锁 gTotalTimes = 20 # 总生产次数 gTimes = 0 # 起始生产次数(计次)，记录生产者生产的次数，达到总生产次数就不再生产 class Producer(threading.Thread): # 生产者类(继承Thread)，赚钱 def run(self): global gMoney global gTimes while True: money = random.randint(100, 1000) # 随机生产100-1000元钱 gLock.acquire() # 创建锁 if gTimes >= gTotalTimes: # 如果生产次数已经达到总生产次数 gLock.release() # 就先释放掉锁 break # 再跳出循环；在锁里面跳出必须要先释放掉锁 要不然会卡死在这里 gMoney += money # 没有达到总生产次数就把生产的钱加入到总金额 # print('%s当前存入%s元钱，剩余%s元钱' % (threading.current_thread(), money, gMoney)) print('生产者[%s] 赚了%d元钱，总余额%d元钱' % (threading.current_thread().name, money, gMoney)) gTimes += 1 # 生产次数计次 gLock.release() # 释放锁 time.sleep(0.5) # 延迟一段时间再继续生产 class Consumer(threading.Thread): # 消费者类(继承Thread)，花钱 def run(self): global gMoney while True: # 消费者一般是个死循环，不停的进行消费 money = random.randint(100, 1000) # 随机花掉100-1000元钱 gLock.acquire() if gMoney > money: # 如果总金额>=要花掉的金额 gMoney -= money # 就花掉 # print('%s当前取出%s元钱，剩余%s元钱' % (threading.current_thread(), money, gMoney)) print('消费者[%s] 消费了%d元钱，总余额%d元钱' % (threading.current_thread().name, money, gMoney)) time.sleep(0.5) else: # 如果总金额<要花掉的金额 # 如果钱不够了，有可能是已经超过了次数，这时候就判断一下 if gTimes >= gTotalTimes: # 判断生产者还在不在生产 gLock.release() # 如果生产者已经不生产了就别傻乎乎的等了！先释放锁 break # 再跳出循环；在锁里面跳出必须要先释放掉锁 要不然会卡死在这里 # print("%s当前想取%s元钱，剩余%s元钱，不足！" % (threading.current_thread(),money,gMoney)) print('消费者[%s] 准备消费%d元钱，总余额%d元钱，余额不足！' % (threading.current_thread().name, money, gMoney)) # 如果生产者还在生产，只是余额不足，就打印提示余额不足 gLock.release() time.sleep(0.5) def main(): for x in range(6): # 创建六个消费者 Consumer(name='消费者线程%d' % x).start() for x in range(5): # 创建五个生产者 Producer(name='生产者线程%d' % x).start() if __name__ == '__main__': main() 

重写Thread类，创建生产者类和消费者类，并用Lock实现生产者和消费者

加一些循环、判断、延迟、变量记录、上锁、释放锁来控制达到生产者和消费者进行生成产和消费

Condition版本

Lock版本的生产者与消费者模式可以正常的运行，但是存在一个不足，在消费者中，总是通过while True死循环并且上锁的方式去判断钱够不够，上锁是一个很耗费CPU资源的行为，因此这种方式不是最好的

还有一种更好的方式便是使用threading.Condition来实现，可以在没有数据的时候让线程处于阻塞等待状态(使用wait方法)，一旦有合适的数据了，还可以使用notify相关的函数来通知其他处于等待状态的线程，这样就可以不用做一些无用的上锁和解锁操作，可以提高程序的性能

threading.Condition类似threading.Lock，可以在修改全局数据的时候进行上锁，在修改完毕后进行解锁

常用的函数

acquire：上锁 类似 threading.Lock的acquire
release：解锁 类似 threading.Lock的release
wait：将当前线程置于阻塞等待状态，并且会释放锁，可以被其他线程使用notify和notify_all函数唤醒，被唤醒后会继续等待上锁，上锁后继续执行下面的代码
notify：通知某个正在等待的线程，默认是第1个等待的线程
notify_all：通知所有正在等待的线程

notify和notify_all不会释放锁，并且需要在release之前调用

import threading import random import time

gMoney = 1000 gCondition = threading.Condition() gTotalTimes = 20 gTimes = 0 class Producer(threading.Thread): def run(self): global gMoney global gTimes while True: money = random.randint(100, 1000) gCondition.acquire() if gTimes >= gTotalTimes: gCondition.release() print('当前生产者总共生产了%s次' % gTimes) break gMoney += money # print('%s当前存入%s元钱，剩余%s元钱' % (threading.current_thread(), money, gMoney)) print('生产者[%s] 赚了%d元钱，总余额%d元钱' % (threading.current_thread().name, money, gMoney)) gTimes += 1 time.sleep(0.5) gCondition.notify_all() # 成功生产后，通知(唤醒)所有等待中的线程 gCondition.release() time.sleep(0.5) class Consumer(threading.Thread): def run(self): global gMoney while True: money = random.randint(100, 1000) gCondition.acquire() # 这里要给个while循环判断，因为等轮到这个线程的时候，条件有可能又不满足了 while gMoney < money: # 用while判断总金额是否<要花掉的金额 if gTimes >= gTotalTimes: # <的话再判断生产者还在不在生产 gCondition.release() # 如果生产者已经不生产了就别傻乎乎的等了！先释放锁 return # 再return # 这里如果用break的话只会跳出这个while循环，外面的会继续执行，所以用return把整个函数返回 # print('%s准备取%s元钱，剩余%s元钱，不足！'%(threading.current_thread(),money,gMoney)) print('消费者[%s] 准备消费%d元钱，总余额%d元钱，余额不足！' % (threading.current_thread().name, money, gMoney)) gCondition.wait() ''''如果只是小于并不是不生产了就用.wait()等待(线程挂起)。然后继续一直这样在这个循环里判断，而不是像是要Lock那样还傻乎乎的一直加锁解锁等待！
            
            这里需要用while这样一直判断下去，如果用if就不行了，用if的话只会判断一次，假如过了一会生产者生产了钱 通知了这边的消费者线程来消费，
            但是这个时候有可能刚好来了另外的线程是没有执行过wait()的(来的早不如来得巧)，那么这条刚来的线程可能会抢先在这条排队等待的线程前面执行，
            等到这条.wait()排队的线程苏醒后再来执行的时候，它会不管三七二十一直接执行下面的gMoney -= money，
            但是这个时候有可能gMoney已经 < money了，因为前面那条刚来的线程已经执行了一次了。
            就好像我排队买票的时候在一边玩手机，这个时候售票的通知有票了，但是刚好了来了插队的抢先了，等他或他们买了之后又没票了，那我肯定不愿意啊，跟售票员说：“你今天死也要卖给我一张票”。''' gMoney -= money # 总金额不<要花掉的金额了就进行消费 # print('%s当前取出%s元钱，剩余%s元钱' % (threading.current_thread(), money, gMoney)) print('消费者[%s] 消费了%d元钱，总余额%d元钱' % (threading.current_thread().name, money, gMoney)) gCondition.release() time.sleep(0.5) def main(): for x in range(6): # 创建六个消费者 Consumer(name='消费者线程%d' % x).start() for x in range(4): # 创建四个生产者 Producer(name='生产者线程%d' % x).start() if __name__ == '__main__': main() 

Queue线程安全队列

在线程中，访问一些全局变量，加锁是一个经常的过程，如果想把一些数据存储到某个队列中，Python内置了一个线程安全的模块叫做queue模块

模块中提供了同步的、线程安全的队列类，包括FIFO（先进先出）队列Queue，LIFO（后入先出）队列LifoQueue，这些队列都实现了锁原语（可以理解为原子操作，即要么不做，要么都一次性做完），能够在多线程中直接使用，可以使用队列来实现线程间的同步

相关的函数

初始化Queue(maxsize)：创建一个先进先出的队列，maxsize英文是最大尺寸的意思
qsize()：返回队列的大小
empty()：判断队列是否为空
full()：判断队列是否满了
get()：从队列中取最后一个数据
put()：将一个数据放到队列中

from queue import Queue import time import threading


q = Queue(4) # 创建一个对象(队列)  指定队列中最多可以放4个元素 q.put(1) # 向队列中添加一个元素 q.put('1') # 向队列中再添加一个元素 print(q.qsize()) # 判断队列有中几个元素 print(q.get()) # 从队列中取最后一个元素；先进先出 print(q.qsize()) # 再判断队列有中几个元素 print(q.empty()) # 判断队列是否为空 print(q.full()) # 判断队列是否满了 for i in range(4): # 通过for循环 q.put(i) # 向队列中添加四个元素 print(q.full()) # 判断队列是否满了 for i in range(4): # 通过for循环 print(q.get()) # 从队列中取出最后四个元素；先进先出 

q.get(block=True)不指定 block默认为真
[从一个队列中获取值的时候，如果队列中没有值就不再取了：一直阻塞在当前位置，直到有值]

q.put(block=True)不指定 block默认为真
[往一个队列中添加值的时候，如果队列满了就不再添加了：一直阻塞在当前位置，直到有空闲]

使用多线程来操作队列

def set_value(q): index = 0 while True: # 通过死循环一直向队列中添加元素 q.put(index) index += 1 time.sleep(3) def get_value(q): while True: # 通过死循环一直从队列中取最后一个元素 print(q.get()) # 每隔三秒才会打印一次，因为上面添加元素的时候是延迟三秒，这个时候才有元素可取 # 因为block=True，所以上面还没添加它就会执行阻塞，就会乖乖得等三秒，等上面添加了元素再获取 def main(): q = Queue(4) # 创建一个对象(队列)  指定队列中最多可以放4个元素 t1 = threading.Thread(target=set_value, args=[q]) # 多线程调用的函数 传参可以用元祖或列表方式传入，一个参数也要用[]或()括起来 t2 = threading.Thread(target=get_value, args=[q]) t1.start() t2.start() if __name__ == '__main__': main() 

单线程下载表情包

import requests from lxml import etree from urllib import request import os import re def parse_page(url): headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.102 Safari/537.36'} response = requests.get(url, headers=headers) text = response.text
    html = etree.HTML(text) imgs = html.xpath('//div[@class="page-content text-center"]//img[@class!="gif"]') for img in imgs: img_url = img.get('data-original') # .get()是获取某个属性下的值 alt = img.get('alt') alt = re.sub(r'[\?？\.，。！\*]', '', alt) # 把图片名称里面的符号替换成空白 suffix = os.path.splitext(img_url)[1] # 使用os.path模块通过分割图片url获取图片的后缀，返回的是一个元组(链接, 后缀)，使用[1]取到后缀 filename = alt + suffix # 图片+后缀构造图片名称 request.urlretrieve(img_url, 'C:\\Users\\Administrator\\Desktop\\images1\\' + filename) print(filename + ' 下载完成！') def main(): for i in range(1, 101): url = 'http://www.doutula.com/photo/list/?page='+str(i) print('当前正在下载第', str(i), '页所有表情包...') parse_page(url) if __name__ == '__main__': main() 

使用生产者与消费者模式多线程下载表情包

import requests from lxml import etree from urllib import request import os import re from queue import Queue import threading import time class Producer(threading.Thread): # 生产者 多线程类，解析每张图片url headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36'} def __init__(self,page_queue,img_queue,*args,**kwargs): # 重写init函数，*args, **kwargs代表任意参数，所以就继承Thread基类所有的参数了 super(Producer, self).__init__(*args,**kwargs) self.page_queue = page_queue # 这两个参数是自己用的，不是继承Thread基类的 self.img_queue = img_queue # 所以在这里定义一下 def run(self): while True: if self.page_queue.empty(): # 如果每一页url的队列是个空的，说明下面的for循环遍历完了，已经没有数据往里面添加了 break # 就跳出死循环，跳出循环后这个run方法就执行完了，这个线程类就会自己结束 url = self.page_queue.get() # 从每一页url的队列获取到每个页面的url self.parse_page(url) # 执行parse_page函数并把获取到每个页面的url传入 def parse_page(self,url): response = requests.get(url,headers=self.headers) text = response.text
        html = etree.HTML(text) imgs = html.xpath('//div[@class="page-content text-center"]//img[@class!="gif"]') for img in imgs: img_url = img.get('data-original') alt = img.get('alt') # .get是获取某个属性下的值，返回字符串str # alt = html.xpath('.//@alt')                   # 用这种方式不行，返回的是列表 # for i in alt: #     alt = i                                   # 遍历也不行，可能是多线程的原因 # print(alt) alt = re.sub(r'[\?？\.，。！\*]', '', alt) # 把图片名称里面的符号替换成空白 suffix = os.path.splitext(img_url)[1] # 使用os.path模块通过分割图片链接获取图片的后缀，返回的是一个元组(链接, 后缀)，使用[1]取到后缀 filename = alt + suffix # 图片+后缀构造图片名称 self.img_queue.put((img_url, filename)) # 把获取到的每张图片url和文件名以元组的方式传入每张图片url的队列 # imgs = html.xpath("//div[@class='page-content text-center']//a//img") # for img in imgs: # if img.get('class') == 'gif': #     continue # img_url = img.xpath(".//@data-original")[0] # suffix = os.path.splitext(img_url)[1] # alt = img.xpath(".//@alt")[0] # alt = re.sub(r'[，。？?,/\\·]','',alt) # img_name = alt + suffix # self.img_queue.put((img_url,img_name)) class Consumer(threading.Thread): # 消费者 多线程类，下载图片 def __init__(self,page_queue,img_queue,*args,**kwargs): super(Consumer, self).__init__(*args,**kwargs) self.page_queue = page_queue
        self.img_queue = img_queue def run(self): while True: if self.img_queue.empty() and self.page_queue.empty(): break # 如果两个队列都是空的，说明已经没有数据了，就跳出循环 img_url, filename = self.img_queue.get() # 因为上面获取到的每张图片url和文件名是以元组的方式传入到队列的的， # 所以这里从队列获取的时候直接通过这种解包方式获取到没张图片url和文件名称，相当于:a,b=(1,2) request.urlretrieve(img_url, 'C:\\Users\\Administrator\\Desktop\\images\\' + filename) time.sleep(1.5) # 多线程太快了所以加些延迟 print(filename + ' 下载完成！') # if self.img_queue.empty(): #     if self.page_queue.empty(): #         return # img = self.img_queue.get(block=True) # url,filename = img # request.urlretrieve(url,'images/'+filename) # print(filename+'  下载完成！') def main(): page_queue = Queue(100) # 每一页url的队列，设置最多可以放100个元素 img_queue = Queue(1000) # 每张图片url的队列，这个尽量设置大一些，要不然队列满了会阻塞，这样耗费在等待队列就划不来了 for x in range(1,101): # 遍历每一页url url = "http://www.doutula.com/photo/list/?page=%d" % x
        page_queue.put(url) # 将for循环遍历出的每一页url添加到每一页url的队列page_queue for x in range(5): # 创建生产者 t = Producer(page_queue,img_queue) # 传入每个页面url的队列和每张图片url的队列 t.start() for x in range(20): # 创建20个消费者，因为每页的图片很多，一页都几十张所以可以设置多些，所以少创建几个生产者 每次同时生产两个url就够消费者消费很久了 t = Consumer(page_queue,img_queue) # 传入每个页面url的队列和每张图片url的队列 t.start() if __name__ == '__main__': main() 

GIL全局解释器锁

Python自带的解释器是CPython，CPython解释器的多线程实际上是一个假的多线程（在多核CPU中，只能利用一核，不能利用多核），同一时刻只有一个线程在执行，为了保证同一时刻只有一个线程在执行，在CPython解释器中有一个东西叫做GIL（Global Intepreter Lock），叫做全局解释器锁，这个解释器锁是有必要的，因为CPython解释器的内存管理不是线程安全的

当然除了CPython解释器，还有其他的解释器，有些解释器是没有GIL锁的

Jython：用Java实现的Python解释器，不存在GIL锁
更多详情请见：https://zh.wikipedia.org/wiki/Jython

IronPython：用.net实现的Python解释器，不存在GIL锁
更多详情请见：https://zh.wikipedia.org/wiki/IronPython

PyPy：用Python实现的Python解释器，存在GIL锁
更多详情请见：https://zh.wikipedia.org/wiki/PyPy

GIL虽然是一个假的多线程，但是在处理一些IO操作（比如文件读写和网络请求）还是可以在很大程度上提高效率的，在IO操作上建议使用多线程提高效率，在一些CPU计算操作上不建议使用多线程，而建议使用多进程

多线程下载百思不得姐段子

import requests from lxml import etree import threading from queue import Queue import csv class BSSpider(threading.Thread): headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36' } def __init__(self,page_queue,joke_queue,*args,**kwargs): super(BSSpider, self).__init__(*args,**kwargs) self.base_domain = 'http://www.budejie.com' self.page_queue = page_queue
        self.joke_queue = joke_queue def run(self): while True: if self.page_queue.empty(): break url = self.page_queue.get() response = requests.get(url, headers=self.headers) text = response.text
            html = etree.HTML(text) descs = html.xpath("//div[@class='j-r-list-c-desc']") for desc in descs: jokes = desc.xpath(".//text()") joke = "\n".join(jokes).strip() link = self.base_domain+desc.xpath(".//a/@href")[0] self.joke_queue.put((joke,link)) print('='*30+"第%s页下载完成！"%url.split('/')[-1]+"="*30) class BSWriter(threading.Thread): headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36' } def __init__(self, joke_queue, writer,gLock, *args, **kwargs): super(BSWriter, self).__init__(*args, **kwargs) self.joke_queue = joke_queue
        self.writer = writer
        self.lock = gLock def run(self): while True: try: joke_info = self.joke_queue.get(timeout=40) joke,link = joke_info
                self.lock.acquire() # 开锁 self.writer.writerow((joke,link)) # 写入csv文件，以元祖方式写入爬取到的结果 self.lock.release() # 解锁 print('保存一条') except: break def main(): page_queue = Queue(10) joke_queue = Queue(500) gLock = threading.Lock() # 定义锁 fp = open('bsbdj.csv', 'a',newline='', encoding='utf-8') writer = csv.writer(fp) writer.writerow(('content', 'link')) # 写入csv文件，以元祖方式写入表头 # 这里先写入表头，然后后面调用线程跑的时候每次写入爬去到的内容 for x in range(1,11): url = 'http://www.budejie.com/text/%d' % x
        page_queue.put(url) for x in range(5): t = BSSpider(page_queue,joke_queue) t.start() for x in range(5): t = BSWriter(joke_queue,writer,gLock) t.start() if __name__ == '__main__': main() 

本文地址：https://blog.csdn.net/weixin_43040873/article/details/109057921