超详细,Python 多线程总结的太到位了

x33g5p2x  于2021-10-31 转载在 Python  
字(4.8k)|赞(0)|评价(0)|浏览(431)

在实际处理数据时,因系统内存有限,我们不可能一次把所有数据都导出进行操作,所以需要批量导出依次操作。为了加快运行,我们会采用多线程的方法进行数据处理,以下为我总结的多线程批量处理数据的模板:

  1. import threading
  2. # 从数据库提取数据的类
  3. class Scheduler():
  4. def __init__(self):
  5. self._lock = threading.RLock()
  6. self.start = 0
  7. # 每次取10000条数据
  8. self.step = 10000
  9. def getdata(self):
  10. # 上锁,以免多线程同时对数据库进行访问,取出重复数据
  11. self._lock.acquire()
  12. # 进行取数据操作
  13. data = 'select * from table' \
  14. 'where id between self.start and self.start + self.step'
  15. # 取完数据后,指针后移
  16. self.start += self.step
  17. self._lock.release()
  18. return data
  19. # 处理数据的过程写在这里
  20. def processdata():
  21. # 从该实例中提取数据
  22. data = scheduler.getdata()
  23. while data:
  24. # 进行处理数据的具体操作:
  25. # 去重、补缺、运算...只要还有数据,本线程就继续取新数据
  26. # 然后再获取数据,进行循环
  27. data = scheduler.getdata()
  28. # 创建多线程,threads_num为创建的线程数
  29. def threads_scheduler(threads_num):
  30. threads = []
  31. for i in range(threads_num):
  32. # 创建线程
  33. td = threading.Thread(target=processdata, name='th'+str(i+1))
  34. threads.append(td)
  35. for t in threads:
  36. # 启动线程
  37. t.start()
  38. for t in threads:
  39. # 子线程守护
  40. t.join()
  41. print('数据已全部处理成功')
  42. if __name__=='__main__':
  43. # 实例化一个调度器,初始化参数
  44. scheduler = Scheduler()
  45. # 创建线程,开始处理数据
  46. threads_scheduler(4)

主要分为三大部分:

Scheduler类,负责初始化参数,getdata方法负责提取数据
*
processdata方法中写具体处理数据的流程
*
threads_scheduler方法负责创建线程

Python多线程的知识我分为4部分进行讲解,以下带大家来回顾重点:

多线程threading

本章先为大家介绍了线程的相关概念:

主线程:当一个程序启动时,就有一个进程被操作系统(OS)创建,与此同时一个线程也立刻运行,该线程通常叫做程序的主线程(Main Thread)。因为它是程序开始时就执行的,如果你需要再创建线程,那么创建的线程就是这个主线程的子线程。

子线程:使用threading、ThreadPoolExecutor创建的线性均为子线程。

主线程的重要性体现在两方面:1.是产生其他子线程的线程;2.通常它必须最后完成执行,比如执行各种关闭动作。

在飞车程序中,如果没有多线程,我们就不能一边听歌一边玩飞车,听歌与玩游戏不能并行;在使用多线程后,我们就可以在玩游戏的同时听背景音乐。在这个例子中启动飞车程序就是一个进程,玩游戏和听音乐是两个线程。

Python提供了threading模块来实现多线程:threading.Thread可以创建线程;setDaemon(True)为守护主线程,默认为False;join()为守护子线程。

  1. from time import sleep
  2. import threading
  3. def music(music_name):
  4. for i in range(2):
  5. print('正在听{}'.format(music_name))
  6. sleep(1)
  7. print('music over')
  8. def game(game_name):
  9. for i in range(2):
  10. print('正在玩{}'.format(game_name))
  11. sleep(3)
  12. print('game over')
  13. threads = []
  14. t1 = threading.Thread(target=music,args=('稻香',))
  15. threads.append(t1)
  16. t2 = threading.Thread(target=game,args=('飞车',))
  17. threads.append(t2)
  18. if __name__ == '__main__':
  19. for t in threads:
  20. # t.setDaemon(True)
  21. t.start()
  22. for t in threads:
  23. t.join()
  24. print('主线程运行结束')

线程池

因为新建线程系统需要分配资源、终止线程系统需要回收资源,所以如果可以重用线程,则可以减去新建/终止的开销以提升性能。同时,使用线程池的语法比自己新建线程执行线程更加简洁。

Python为我们提供了ThreadPoolExecutor来实现线程池,此线程池默认子线程守护。它的适应场景为突发性大量请求或需要大量线程完成任务,但实际任务处理时间较短。

  1. from time import sleep
  2. # fun为定义的待运行函数
  3. with ThreadPoolExecutor(max_workers=5) as executor:
  4. ans = executor.map(fun, [遍历值])
  5. for res in ans:
  6. print(res)
  7. with ThreadPoolExecutor(max_workers=5) as executor:
  8. list = [遍历值]
  9. ans = [executor.submit(fun, i) for i in list]
  10. for res in as_completed(ans):
  11. print(res.result())

其中max_workers为线程池中的线程个数,常用的遍历方法有map和submit+as_completed。根据业务场景的不同,若我们需要输出结果按遍历顺序返回,我们就用map方法,若想谁先完成就返回谁,我们就用submit+as_complete方法。

线程互斥

我们把一个时间段内只允许一个线程使用的资源称为临界资源,对临界资源的访问,必须互斥的进行。互斥,也称间接制约关系。线程互斥指当一个线程访问某临界资源时,另一个想要访问该临界资源的线程必须等待。当前访问临界资源的线程访问结束,释放该资源之后,另一个线程才能去访问临界资源。锁的功能就是实现线程互斥。

我把线程互斥比作厕所包间上大号的过程,因为包间里只有一个坑,所以只允许一个人进行大号。当第一个人要上厕所时,会将门上上锁,这时如果第二个人也想大号,那就必须等第一个人上完,将锁解开后才能进行,在这期间第二个人就只能在门外等着。这个过程与代码中使用锁的原理如出一辙,这里的坑就是临界资源。

Python 的 threading 模块引入了锁。threading 模块提供了 Lock 类,它有如下方法加锁和释放锁:

acquire():对 Lock加锁,其中timeout参数指定加锁多少秒
*
release():释放锁

  1. class Account:
  2. def __init__(self, card_id, balance):
  3. # 封装账户ID、账户余额的两个变量
  4. self.card_id= card_id
  5. self.balance = balance
  6. def withdraw(account, money):
  7. # 进行加锁
  8. lock.acquire()
  9. # 账户余额大于取钱数目
  10. if account.balance >= money:
  11. # 吐出钞票
  12. print(threading.current_thread().name + "取钱成功!吐出钞票:" + str(money),end=' ')
  13. # 修改余额
  14. account.balance -= money
  15. print("\t余额为: " + str(account.balance))
  16. else:
  17. print(threading.current_thread().name + "取钱失败!余额不足")
  18. # 进行解锁
  19. lock.release()
  20. # 创建一个账户,银行卡id为8888,存款1000元
  21. acct = Account("8888" , 1000)
  22. # 模拟两个对同一个账户取钱
  23. # 在主线程中创建一把锁
  24. lock = threading.Lock()
  25. threading.Thread(name='窗口A', target=withdraw , args=(acct , 800)).start()
  26. threading.Thread(name='窗口B', target=withdraw , args=(acct , 800)).start()

lock与Rlock的区别

区别一:Lock被称为原始锁,一个线程只能请求一次;RLock被称为重入锁,可以被一个线程请求多次,即锁中可以嵌套锁。

  1. import threading
  2. def main():
  3. lock.acquire()
  4. print('第一道锁')
  5. lock.acquire()
  6. print('第二道锁')
  7. lock.release()
  8. lock.release()
  9. if __name__ == '__main__':
  10. lock = threading.Lock()
  11. main()

我们会发现这个程序只会打印“第一道锁”,而且程序既没有终止,也没有继续运行。这是因为Lock锁在同一线程内第一次加锁之后还没有释放时,就进行了第二次acquire请求,导致无法执行release,所以锁永远无法释放,这就是死锁。如果我们使用RLock就能正常运行,不会发生死锁的状态。

区别二:当Lock处于锁定状态时,不属于特定线程,可在另一个线程中进行解锁释放;而RLock只有当前线程才能释放本线程上的锁,不可由其他线程进行释放,所以在使用RLock时,acquire与release必须成对出现,即解铃还须系铃人。

  1. import threading
  2. def main():
  3. lock.release()
  4. print("在子线程解锁后打印")
  5. if __name__ == '__main__':
  6. lock = threading.Lock()
  7. lock.acquire()
  8. t = threading.Thread(target=main)
  9. t.start()

在主线程中定义Lock锁,然后上锁,再创建一个子线程t运行main函数释放锁,结果正常输出,说明主线程上的锁,可由子线程解锁。

如果把上面的锁改为RLock则报错。在实际中设计程序时,我们会将每个功能分别封装成一个函数,每个函数中都可能会有临界区域,所以就需要用到RLock。

  1. import threading
  2. import time
  3. def fun_1():
  4. print('开始')
  5. time.sleep(1)
  6. lock.acquire()
  7. print("第一道锁")
  8. fun_2()
  9. lock.release()
  10. def fun_2():
  11. lock.acquire()
  12. print("第二道锁")
  13. lock.release()
  14. if __name__ == '__main__':
  15. lock = threading.RLock()
  16. t1 = threading.Thread(target=fun_1)
  17. t2 = threading.Thread(target=fun_1)
  18. t1.start()
  19. t2.start()

一句话总结就是Lock不能套娃,RLock可以套娃;Lock可以由其他线程中的锁进行操作,RLock只能由本线程进行操作。

以上就是多线程所有内容,喜欢的小伙伴点赞支持,收藏。

相关文章