devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Python3并发编程-多线程threading_python3 multithreading 并发请求

作者：从前慢现在也慢 | 2024-02-26 02:54:06

赞

踩

python3 multithreading 并发请求

目录

1：线程的创建

1.1:Thread类创建线程

1.2:继承Thread类的子类创建

2：线程的同步

2.1：锁：Lock

2.2：死锁

2.3：递归锁：RLock

2.4：信号量：BoundedSemaphore

2.5：事件:Event

2.6：线程池

2.6.1：submit方法

2.6.2：map方法

同一进程的各个线程间可以共享主线程的地址空间和各种资源。

1：线程的创建

1.1:Thread类创建线程

  1. # -*- coding: utf-8 -*-
  2.  
  3. from threading import Thread
  4.  
  5. import os
  6. import time
  7.  
  8. def func(index,dic):
  9.     print(f'线程{index}；进程id={os.getpid()}')
  10.  
  11.     dic['cnt'] += 1
  12.     count = 0
  13.  
  14.     while True:
  15.         time.sleep(1)
  16.         if count > 2:
  17.             break
  18.         count += 1
  19.         dic['cnt'] += 1
  20.         print(f'线程{index}；cnt=[{dic["cnt"]}]')
  21.  
  22.     print(f'线程{index}；cnt=[{dic["cnt"]}],退出')
  23.  
  24.  
  25. def main():
  26.     dic = {}
  27.     dic['cnt'] = 0
  28.     t_list = []
  29.     for i in range(5):
  30.         t = Thread(target=func, args=(i, dic,))  # 线程中数据是共享的，看dic中数据的变化
  31.         # t.setDaemon(True)  # 默认为False；设置为True时，主线程退出时，子线程也会退出
  32.         t.start()
  33.         # print(t.getName())  # 获取线程名称
  34.         t_list.append(t)
  35.  
  36.     print(f'主线程,进程id={os.getpid()}')
  37.     for t in t_list:
  38.         t.join()  # 等待线程的结束,非必须。
  39.  
  40.     print(f'主线程；cnt=[{dic["cnt"]}],退出')
  41.  
  42. if __name__ == '__main__':
  43.     main()

1.2:继承Thread类的子类创建

  1. # -*- coding: utf-8 -*-
  2.  
  3. from threading import Thread
  4.  
  5. import os
  6. import time
  7.  
  8.  
  9. class MyThread(Thread):
  10.     def __init__(self, index,dic):
  11.         super().__init__()
  12.         self.index = index
  13.         self.dic = dic
  14.  
  15.  
  16.     def run(self): # 调用start()函数之后会自动调用此函数
  17.         print(f'线程{self.index}；进程id={os.getpid()}')
  18.  
  19.         self.dic['cnt'] += 1
  20.         count = 0
  21.  
  22.         while True:
  23.             time.sleep(1)
  24.             if count > 2:
  25.                 break
  26.             count += 1
  27.             self.dic['cnt'] += 1
  28.             print(f'线程{self.index}；cnt=[{self.dic["cnt"]}]')
  29.  
  30.         print(f'线程{self.index}；cnt=[{self.dic["cnt"]}],退出')
  31.  
  32.  
  33. def main():
  34.     dic = {}
  35.     dic['cnt'] = 0
  36.     t_list = []
  37.     for i in range(5):
  38.         t = MyThread(i, dic)
  39.         # t.setDaemon(True)  # 默认为False；设置为True时，主线程退出时，子线程也会退出
  40.         t.start()
  41.         # print(t.getName())  # 获取线程名称
  42.         t_list.append(t)
  43.  
  44.     print(f'主线程,进程id={os.getpid()}')
  45.     for t in t_list:
  46.         t.join()  # 等待线程的结束,非必须。
  47.  
  48.     print(f'主线程；cnt=[{dic["cnt"]}],退出')
  49.  
  50. if __name__ == '__main__':
  51.     main()

2：线程的同步

同一进程下的线程资源是共享的，但是对共享数据的操作是不安全的，因此需要进行同步操作。

2.1：锁：Lock

  1. # -*- coding: utf-8 -*-
  2.  
  3. from threading import Thread
  4. from multiprocessing import Lock
  5. import os
  6. import time
  7.  
  8. def func(index,dic,lock):
  9.  
  10.     print(f'线程{index}；进程id={os.getpid()}')
  11.     while True:
  12.         try:
  13.             lock.acquire()
  14.             cnt = dic['cnt']
  15.             time.sleep(0.0001)   # 加点延时，不然看不到效果
  16.             if cnt > 0:
  17.                 dic['cnt'] -= 1
  18.                 print(f'线程{index};获取到:{cnt}号票;剩下={dic["cnt"]};')
  19.             else:
  20.                 lock.release()
  21.                 break
  22.             lock.release()
  23.         except:
  24.             break
  25.  
  26.     print(f'线程{index}；cnt=[{dic["cnt"]}],退出')
  27.  
  28.  
  29. def main():
  30.     lock = Lock()
  31.     dic = {}
  32.     dic['cnt'] = 20
  33.     t_list = []
  34.     for i in range(5):
  35.         t = Thread(target=func, args=(i, dic,lock,))  # 线程中数据是共享的，看dic中数据的变化
  36.         # t.setDaemon(True)  # 默认为False；设置为True时，主线程退出时，子线程也会退出
  37.         t.start()
  38.         # print(t.getName())  # 获取线程名称
  39.         t_list.append(t)
  40.  
  41.     print(f'主线程,进程id={os.getpid()}')
  42.     for t in t_list:
  43.         t.join()  # 等待线程的结束,非必须。
  44.  
  45.     print(f'主线程；cnt=[{dic["cnt"]}],退出')
  46.  
  47. if __name__ == '__main__':
  48.     main()

把锁注释掉，看运行结果：

  1. # -*- coding: utf-8 -*-
  2.  
  3. from threading import Thread
  4. from multiprocessing import Lock
  5. import os
  6. import time
  7.  
  8. def func(index,dic,lock):
  9.  
  10.     print(f'线程{index}；进程id={os.getpid()}')
  11.     while True:
  12.         try:
  13.             # lock.acquire()
  14.             cnt = dic['cnt']
  15.             time.sleep(0.0001)   # 加点延时，不然看不到效果
  16.             if cnt > 0:
  17.                 dic['cnt'] -= 1
  18.                 print(f'线程{index};获取到:{cnt}号票;剩下={dic["cnt"]};')
  19.             else:
  20.                 # lock.release()
  21.                 break
  22.             # lock.release()
  23.         except:
  24.             break
  25.  
  26.     print(f'线程{index}；cnt=[{dic["cnt"]}],退出')
  27.  
  28.  
  29. def main():
  30.     lock = Lock()
  31.     dic = {}
  32.     dic['cnt'] = 20
  33.     t_list = []
  34.     for i in range(5):
  35.         t = Thread(target=func, args=(i, dic,lock,))  # 线程中数据是共享的，看dic中数据的变化
  36.         # t.setDaemon(True)  # 默认为False；设置为True时，主线程退出时，子线程也会退出
  37.         t.start()
  38.         # print(t.getName())  # 获取线程名称
  39.         t_list.append(t)
  40.  
  41.     print(f'主线程,进程id={os.getpid()}')
  42.     for t in t_list:
  43.         t.join()  # 等待线程的结束,非必须。
  44.  
  45.     print(f'主线程；cnt=[{dic["cnt"]}],退出')
  46.  
  47. if __name__ == '__main__':
  48.     main()

2.2：死锁

死锁就是有多个锁，多个线程中互相等待对方的锁。

比如：

有两个锁：锁A，锁B；

有两个线程：线程A，线程B

在线程A中已获取锁A，在线程B中已获取锁B，然后在线程A中要获取锁B，而此时锁B已在线程B中被占有，线程A中只能等待；在线程B中，又要获取锁A，而锁A已在线程A中被占有，这时线程B只能等待；就这样形成了，线程A，线程B分别在等待对方释放自己所要的锁。
 

  1. # -*- coding: utf-8 -*-
  2.  
  3. from threading import Thread
  4. from multiprocessing import Lock
  5. import os
  6. import time
  7.  
  8.  
  9. from threading import Thread,Lock
  10. import time
  11.  
  12.  
  13. def funcA(lockA,lockB):
  14.     print(" funcA:需要A锁")
  15.     lockA.acquire()
  16.     time.sleep(0.2)
  17.     print(" funcA:获得了:A锁")
  18.     print(" funcA:需要B锁")
  19.     lockB.acquire()  # B锁 在funcB中被占有，而在funcB中又在等待A锁的释放
  20.     print(" funcA:获得了:B锁")
  21.     lockB.release()
  22.     lockA.release()
  23.  
  24. def funcB(lockA,lockB):
  25.     print(" funcB:需要B锁")
  26.     lockB.acquire()
  27.     time.sleep(0.2)
  28.     print(" funcB:获得了:B锁")
  29.  
  30.     print(" funcB:需要A锁")  # A锁 在funcA
  31.     lockA.acquire()  # A锁 在funcA中被占有，而在funcA中又在等待B锁的释放
  32.     print(" funcB:获得了:A锁")
  33.  
  34.     lockA.release()
  35.     lockB.release()
  36.  
  37. def main():
  38.     lockA = Lock()
  39.     lockB = Lock()
  40.  
  41.     t_a = Thread(target=funcA, args=(lockA, lockB,))
  42.     t_a.start()
  43.  
  44.     t_b = Thread(target=funcB, args=(lockA, lockB,))
  45.     t_b.start()
  46.  
  47.     print("结束主线程")
  48.  
  49. if __name__ == "__main__":
  50.     main()
  51.

2.3：递归锁：RLock

  1. # -*- coding: utf-8 -*-
  2.  
  3. from threading import Thread
  4. from multiprocessing import Lock
  5. import os
  6. import time
  7.  
  8.  
  9. from threading import Thread,RLock  # 递归锁
  10. import time
  11.  
  12.  
  13. def funcA(lockA,lockB):
  14.     print(" funcA:需要A锁")
  15.     lockA.acquire()
  16.     time.sleep(0.2)
  17.     print(" funcA:获得了:A锁")
  18.     print(" funcA:需要B锁")
  19.     lockB.acquire()
  20.     print(" funcA:获得了:B锁")
  21.     lockB.release()
  22.     lockA.release()
  23.  
  24. def funcB(lockA,lockB):
  25.     print(" funcB:需要B锁")
  26.     lockB.acquire()
  27.     time.sleep(0.2)
  28.     print(" funcB:获得了:B锁")
  29.  
  30.     print(" funcB:需要A锁")
  31.     lockA.acquire()
  32.     print(" funcB:获得了:A锁")
  33.  
  34.     lockA.release()
  35.     lockB.release()
  36.  
  37. def main():
  38.  
  39.     lockA = lockB = RLock()
  40.  
  41.     t_a = Thread(target=funcA, args=(lockA, lockB,))
  42.     t_a.start()
  43.  
  44.     t_b = Thread(target=funcB, args=(lockA, lockB,))
  45.     t_b.start()
  46.  
  47.     print("结束主线程")
  48.  
  49. if __name__ == "__main__":
  50.     main()
  51.

2.4：信号量：BoundedSemaphore

互斥锁同时只允许一个线程更改数据，而信号量是同时允许多少个线程同时运行。

  1. # -*- coding: utf-8 -*-
  2.  
  3. from threading import Thread,BoundedSemaphore,active_count
  4. import time
  5.  
  6. def func(index, semaphore):
  7.     semaphore.acquire()   #加锁
  8.     print(f"线程:{index} 运行中...,当前活动线程数:{active_count()}")
  9.     time.sleep(5) # 为了看到效果，休眠久点
  10.     semaphore.release()     #释放
  11.  
  12.  
  13. def main():
  14.     semaphore = BoundedSemaphore(3)  # 最多允许3个线程同时运行
  15.     for i in range(20):
  16.         t = Thread(target=func, args=(i, semaphore))
  17.         t.start()
  18.  
  19.     print(f"主线程:当前活动线程数:{active_count()}")
  20.  
  21. if __name__ == '__main__':
  22.     main()

2.5：事件:Event

  1. # -*- coding: utf-8 -*-
  2. import time
  3. from threading import Thread,Event
  4.  
  5. def light(e):
  6.     while 1:
  7.         print('现在是红灯：')
  8.         time.sleep(5)
  9.         e.set()   # 设置event的状态值为True ；
  10.         print('现在是绿灯：')
  11.         time.sleep(3)
  12.         e.clear()  # 恢复event的状态值为False。
  13.  
  14. def car(index,e):
  15.     if e.is_set():  # 返回event的状态值；
  16.         # 状态为True
  17.         print(f'    现在是绿灯，{index}:过马路中！')
  18.     else:
  19.         print(f'    现在是红灯{index}:等待中！')
  20.         e.wait()  #  如果 event.isSet()==False将阻塞
  21.         print(f'    红灯变绿灯，{index}:可以走了!')
  22.  
  23. def main():
  24.     e = Event()
  25.     lgh = Thread(target=light, args=(e,))
  26.     lgh.start()
  27.     cnt = 0
  28.  
  29.     while 1:
  30.         time.sleep(1)  # 每隔1秒来一辆car
  31.         t1 = Thread(target=car, args=(cnt, e,))
  32.         t1.start()
  33.         cnt += 1
  34.  
  35. if __name__ == '__main__':
  36.     main()

2.6：线程池

2.6.1：submit方法

  1. # -*- coding: utf-8 -*-
  2.  
  3. from  concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
  4. import time
  5. from threading import current_thread
  6.  
  7.  
  8. def func(index):
  9.     # print(index,current_thread().ident)
  10.     time.sleep(0.1)
  11.     return [index,index**2]
  12.  
  13.  
  14. if __name__ == "__main__":
  15.     t_p = ThreadPoolExecutor(max_workers=6)
  16.     t_ret_list = []
  17.     for i  in range(20):
  18.         t = t_p.submit(func, i)
  19.         t_ret_list.append(t)
  20.  
  21.     for ret in t_ret_list:
  22.         print(ret.result())
  23.  
  24.

2.6.2：map方法

  1. # -*- coding: utf-8 -*-
  2.  
  3. from  concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
  4. import time
  5. from threading import current_thread
  6.  
  7. def func(index):
  8.     # print(index,current_thread().ident)
  9.     time.sleep(0.1)
  10.     return [index,index**2]
  11.  
  12. if __name__ == "__main__":
  13.     t_p = ThreadPoolExecutor(max_workers=6)
  14.     map_ret = t_p.map(func,range(20))
  15.     print(map_ret)
  16.     for ret in map_ret:
  17.         print(ret)

 

本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号