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

多线程 Threading Multiprocessing(Python）

作者：羊村懒王 | 2024-02-27 08:43:36

赞

踩

python threading multiprocessing多线程 return

多线程是加速程序计算的有效方式，Python的多线程模块threading上手快速简单，学习莫烦多线程教程动手操作了一遍，这里记录一下。

1 Threading

1.1 添加线程
  1. import threading
  2. #获取已激活的线程数
  3. print(threading.active_count())  #1
  4.  
  5. #查看所有线程信息
  6. print(threading.enumerate())  #[<_MainThread(MainThread, started 18496)>]
  7.  
  8. #查看现在正在运行的线程
  9. print(threading.current_thread()) #<_MainThread(MainThread, started 18496)>
  10.  
  11.  
  12. import threading
  13.  
  14. def thread_job():
  15.     print('This is a thread of %s' % threading.current_thread())
  16.  
  17. def runMain():
  18.     thread = threading.Thread(target=thread_job,) #定义线程
  19.     thread.start()  #线程开始工作
  20.  
  21. if __name__ == '__main__':
  22.     runMain()
  23.  
  24. #输出
  25. This is a thread of <Thread(Thread-1, started 12324)>
1.2 join功能

不加join功能，线程任务还未完成便输出all done。

  1. import threading
  2. import time
  3.  
  4. def thread_job():
  5.     print('T1 start.\n')
  6.     for i in range(10):
  7.         time.sleep(0.1)  #任务间隔0.1秒
  8.     print('T1 finish.\n')
  9.  
  10. add_thread = threading.Thread(target=thread_job, name='T1')
  11. add_thread.start()
  12.  
  13. print('all done.\n')
  14. #输出
  15. T1 start.
  16. all done.
  17. T1 finish.

若要遵循顺序，在启动线程后调用join ， 使用join控制多个线程的执行顺序，效果如下。

  1. import threading
  2. import time
  3.  
  4. def thread_job():
  5.     print('T1 start.\n')
  6.     for i in range(10):
  7.         time.sleep(0.1)  #任务间隔0.1秒
  8.     print('T1 finish.\n')
  9.  
  10. add_thread = threading.Thread(target=thread_job, name='T1')
  11. add_thread.start()
  12. add_thread.join()
  13.  
  14. print('all done.\n')
  15. #输出
  16. T1 start.
  17. T1 finish.
  18. all done.
1.3 存储进程结果Queue

将数据列表中的数据传入，使用四个线程处理，将结果保存在Queue中，线程执行完后，从Queue中获取存储的结果

  1. #导入线程 队列的标准模块
  2. import threading
  3. import time
  4. from queue import Queue

定义一个被多线程调用的函数：函数的参数时一个列表l和一个队列q，函数的功能是对列表的每个元素进行平方计算,将结果保存在队列中

  1. def job(l,q):
  2.     for i in range(len(l)):
  3.         l[i] = l[i]**2
  4.     q.put(l)

定义一个多线程函数：在多线程函数中定义一个Queue ,用来保存返回值 ,代替return ,定义一个多线程列表 ,初始化一个多维数据列表

  1. def mulithreading():
  2.     q = Queue() #q中存放返回值 代替return的返回值
  3.     threads = []
  4.     data = [[1,2,3],[3,4,5],[4,4,4],[5,5,5]]

在多线程函数中定义四个线程,启动线程,将每个线程添加到多线程的列表中

  1. for i in range(4): #定义四个线程
  2.         t = threading.Thread(target=job,args=(data[i],q))
  3.         t.start()
  4.         threads.append(t) #把每个线程append到线程列表中

分别join四个线程到主线程

  1. for thread in threads:
  2.         thread.join()

定义一个空列表results 将四个线程运行后保存在队列中的结果返回给results

  1. results = []
  2.     for _ in range(4):
  3.         results.append(q.get()) #q.get()按顺序从q中拿出一个值
  4.     print(results)

完整代码：

  1. #导入线程 队列的标准模块
  2. import threading
  3. import time
  4. from queue import Queue
  5.  
  6. #定义一个被多线程调用的函数
  7. def job(l,q):
  8.     for i in range(len(l)):
  9.         l[i] = l[i]**2
  10.     q.put(l)
  11. #定义一个多线程函数
  12. def mulithreading():
  13.     q = Queue() #q中存放返回值 代替return的返回值
  14.     threads = []
  15.     data = [[1,2,3],[3,4,5],[4,4,4],[5,5,5]]
  16.     for i in range(4): #定义四个线程
  17.         t = threading.Thread(target=job,args=(data[i],q))
  18.         t.start()
  19.         threads.append(t) #把每个线程append到线程列表中
  20.     #分别join四个线程到主线程
  21.     for thread in threads:
  22.         thread.join()
  23.     #定义一个空列表results 将四个线程运行后保存在队列中的结果返回给results
  24.     results = []
  25.     for _ in range(4):
  26.         results.append(q.get()) #q.get()按顺序从q中拿出一个值
  27.     print(results)
  28. if __name__ == '__main__':
  29.     mulithreading()
  30.  
  31. #输出
  32. [[1, 4, 9], [9, 16, 25], [16, 16, 16], [25, 25, 25]]
1.4 GIL 不一定有效率

python 的多线程 threading 有时候并不是特别理想. 最主要的原因是就是, Python 的设计上, 有一个必要的环节, 就是 Global Interpreter Lock (GIL). 这个东西让 Python 还是一次性只能处理一个东西.

  1. import threading
  2. from queue import Queue
  3. import copy
  4. import time
  5.  
  6. def job(l, q):
  7.     res = sum(l)
  8.     q.put(res)
  9.  
  10. def multithreading(l):
  11.     q = Queue()
  12.     threads = []
  13.     for i in range(4):
  14.         t = threading.Thread(target=job, args=(copy.copy(l), q), name='T%i' % i)
  15.         t.start()
  16.         threads.append(t)
  17.     [t.join() for t in threads]
  18.     total = 0
  19.     for _ in range(4):
  20.         total += q.get()
  21.     print(total)
  22.  
  23. def normal(l):
  24.     total = sum(l)
  25.     print(total)
  26.  
  27. if __name__ == '__main__':
  28.     l = list(range(10000000))
  29.     s_t = time.time()
  30.     normal(l*4)
  31.     print('normal: ',time.time()-s_t)
  32.     s_t = time.time()
  33.     multithreading(l)
  34.     print('multithreading: ', time.time()-s_t)
  35. #输出
  36. 199999980000000
  37. normal:  1.7343778610229492
  38. 199999980000000
  39. multithreading:  2.218825340270996

程序 threading 和 Normal 运行了一样多次的运算. 但是我们发现 threading 却没有快多少, 按理来说, 我们预期会要快3-4倍, 因为有建立4个线程, 但是并没有. 这就是其中的 GIL 在作怪.

1.5 线程锁

不使用锁

  1. import threading
  2.  
  3. def job1():  #全局变量A的值每次加1，循环10次
  4.     global A
  5.     for i in range(10):
  6.         A += 1
  7.         print('job1',A)
  8.  
  9. def job2(): #全局变量A的值每次加10，循环10次
  10.     global A
  11.     for i in range(10):
  12.         A += 10
  13.         print('job2',A)
  14.  
  15. if __name__ == '__main__':
  16.     A = 0
  17.     t1 = threading.Thread(target=job1)
  18.     t2 = threading.Thread(target=job2)
  19.     t1.start()
  20.     t2.start()
  21.     t1.join()
  22.     t2.join()
  23.  
  24. #输出 打印结果非常混乱
  25. job1 1
  26. job1 2
  27. job1 3
  28. job1 4
  29. job2 14
  30. job1 15
  31. job2 25
  32. job1 26
  33. job2 36
  34. job1 37
  35. job2 47
  36. job1 48
  37. job2 58
  38. job1 59
  39. job2 69
  40. job1 70
  41. job2 80
  42. job2 90
  43. job2 100
  44. job2 110

使用锁
lock在不同线程使用同一共享内存时，能够确保线程之间互不影响，使用lock的方法是， 在每个线程执行运算修改共享内存之前，执行lock.acquire()将共享内存上锁， 确保当前线程执行时，内存不会被其他线程访问，执行运算完毕后，使用lock.release()将锁打开， 保证其他的线程可以使用该共享内存。

  1. import threading
  2.  
  3. def job1():
  4.     global A;lock = threading.Lock()
  5.     lock.acquire()
  6.     for i in range(10):
  7.         A += 1
  8.         print('job1',A)
  9.     lock.release()
  10.  
  11. def job2():
  12.     global A;lock = threading.Lock()
  13.     lock.acquire()
  14.     for i in range(10):
  15.         A += 10
  16.         print('job2',A)
  17.     lock.release()
  18.  
  19. if __name__ == '__main__':
  20.     A = 0
  21.     t1 = threading.Thread(target=job1)
  22.     t2 = threading.Thread(target=job2)
  23.     t1.start()
  24.     t2.start()
  25.     t1.join()
  26.     t2.join()
  27.  
  28. #输出  使用lock后 执行完一个线程后再执行另一个线程。使用lock和不使用lock，最后打印输出的结果是不同的。
  29. job1 1
  30. job1 2
  31. job1 3
  32. job1 4
  33. job1 5
  34. job1 6
  35. job1 7
  36. job1 8
  37. job1 9
  38. job1 10
  39. job2 20
  40. job2 30
  41. job2 40
  42. job2 50
  43. job2 60
  44. job2 70
  45. job2 80
  46. job2 90
  47. job2 100
  48. job2 110

2 Multiprocessing

多进程 Multiprocessing 和多线程 threading 类似, 都是在 python 中用来并行运算的。不过既然有了 threading, 为什么 Python 还要出一个 multiprocessing 呢? 因为要用来弥补 threading 的一些劣势, 比如在 threading 教程中提到的GIL， python 把 multiprocessing 和 threading 的使用方法做的几乎差不多，使用多线程发挥电脑多核系统的威力。

2.1添加Process
  1. #导入线程进程标准模块
  2. import multiprocessing as mp
  3. import threading as td
  4.  
  5. #定义一个被线程和进程调用的函数
  6. def job(a,d):
  7.     print('AA')
  8.  
  9. #创建线程和进程
  10. t1=td.Thread(target=job,args=(1,2))
  11. p1=mp.Process(target=job,args=(1,2))
  12.  
  13. #启动线程和进程
  14. t1.start()
  15. p1.start()
  16.  
  17. #连接线程和进程
  18. t1.join()
  19. p1.join()
  20.  
  21. #可以看出线程和进程的使用方式相似

完整代码

  1. #导入线程进程标准模块
  2. import multiprocessing as mp
  3. import threading as td
  4.  
  5. #定义一个被进程调用的函数
  6. def job(a,d):
  7.     print('AA')
  8.  
  9. if __name__ == '__main__':
  10.     p1 = mp.Process(target=job, args=(1, 2)) #创建进程
  11.     p1.start() #启动进程
  12.     p1.join()  #连接进程
  13. #输出
  14. AA
2.2 存储进程输出 Queue

Queue的功能是将每个核或线程的运算结果放在队里中， 等到每个线程或核运行完毕后再从队列中取出结果， 继续加载运算。因为多线程调用的函数不能有返回值, 所以使用Queue存储多个线程运算的结果。
定义一个被多线程调用的函数，q 就像一个队列，用来保存每次函数运行的结果

  1. #定义一个多线程调用函数
  2. def job(q): #注:该函数没有返回值
  3.     res = 0
  4.     for i in range(1000):
  5.         res += i+i**2+i**3
  6.     q.put(res) #queue

定义两个线程函数，用来处理同一个任务, args 的参数只要一个值的时候，参数后面需要加一个逗号，表示args是可迭代的，后面可能还有别的参数，不加逗号会出错

  1. p1 = mp.Process(target=job,args=(q,))
  2. p2 = mp.Process(target=job,args=(q,))

完整代码实现

  1. import multiprocessing as mp
  2.  
  3. #定义一个多线程调用函数
  4. def job(q): #注:该函数没有返回值
  5.     res = 0
  6.     for i in range(1000):
  7.         res += i+i**2+i**3
  8.     q.put(res) #queue
  9. if __name__ == '__main__':
  10.     q=mp.Queue()  #定义一个多线程队列 存储结果
  11.     p1 = mp.Process(target=job, args=(q,))
  12.     p2 = mp.Process(target=job, args=(q,))
  13.     p1.start() #启动线程  分两批处理
  14.     p2.start()
  15.     p1.join() #连接线程
  16.     p2.join()
  17.     res1=q.get() #分两批输出 将结果分别保存
  18.     res2=q.get()
  19.     print(res1+res2)
  20. #输出
  21. 499667166000
2.3效率对比

对比下多进程，多线程和什么都不做时的消耗时间，看看哪种方式更有效率。

  1. import multiprocessing as mp
  2. def job(q):
  3.     res=0
  4.     for i in range(1000000):
  5.         res += i + i**2 + i**3
  6.     q.put(res)
  7.  
  8. #由于多进程是多核运算 多进程代码命名为multicore()
  9. def multicore():
  10.     q = mp.Queue()
  11.     p1 = mp.Process(target=job, args=(q,))
  12.     p2 = mp.Process(target=job, args=(q,))
  13.     p1.start()
  14.     p2.start()
  15.     p1.join()
  16.     p2.join()
  17.     res1 = q.get()
  18.     res2 = q.get()
  19.     print('multicore:',res1 + res2)
  20.  
  21. #创建多线程
  22. import threading as td
  23. def multithread():
  24.     q = mp.Queue() # thread可放入process同样的queue中
  25.     t1 = td.Thread(target=job, args=(q,))
  26.     t2 = td.Thread(target=job, args=(q,))
  27.     t1.start()
  28.     t2.start()
  29.     t1.join()
  30.     t2.join()
  31.     res1 = q.get()
  32.     res2 = q.get()
  33.     print('multithread:', res1 + res2)
  34.  
  35. #创建普通函数
  36. def normal():
  37.     res = 0
  38.     for _ in range(2):
  39.         for i in range(1000000):
  40.             res += i + i**2 + i**3
  41.     print('normal:', res)
  42.  
  43. import time
  44. if __name__ == '__main__':
  45.     st = time.time()
  46.     normal()
  47.     st1 = time.time()
  48.     print('normal time:', st1 - st)
  49.     multithread()
  50.     st2 = time.time()
  51.     print('multithread time:', st2 - st1)
  52.     multicore()
  53.     print('multicore time:', time.time() - st2)
  54.  
  55. #输出
  56. normal: 499999666667166666000000
  57. normal time: 1.6875250339508057
  58. multithread: 499999666667166666000000
  59. multithread time: 3.1562907695770264
  60. multicore: 499999666667166666000000
  61. multicore time: 1.0937612056732178

这次运行时间依然是:多进程 < 普通 < 多线程。 发现多核/多进程最快，说明在同时间运行了多个任务。 而多线程的运行时间居然比什么都不做的程序还要慢一点，说明多线程还是有短板。

2.4 进程池Pool

进程池就是将所要运行的东西，放到池子里，Python会自行解决多进程的问题
2.4.1 进程池Pool()和map()

  1. #定义一个Pool
  2. pool = mp.Pool()

有了池子之后，就可以让池子对应某一个函数，向池子里丢数据，池子就会返回函数返回的值。 Pool和之前的Process的不同点是丢向Pool的函数有返回值，而Process的没有返回值。
接下来用map()获取结果，在map()中需要放入函数和需要迭代运算的值，然后它会自动分配给CPU核，返回结果 res = pool.map(job, range(10))

  1. import multiprocessing as mp
  2.  
  3. def job(x):
  4.     return x*x
  5.  
  6. def multicore():
  7.     pool = mp.Pool()
  8.     res = pool.map(job,range(10))
  9.     print(res)
  10.  
  11. if __name__ == '__main__':
  12.     multicore()
  13.  
  14. #输出
  15. [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

2.4.2 自定义核数量
怎么知道Pool是否真的调用了多个核呢？可以把迭代次数增大些，然后打开CPU负载看下CPU运行情况
打开CPU负载：活动监视器 > CPU > CPU负载(单击一下即可)
Pool默认大小是CPU的核数，我们也可以通过在Pool中传入processes参数即可自定义需要的核数量，

  1. def multicore():
  2.     pool = mp.Pool(processes=3) # 定义CPU核数量为3
  3.     res = pool.map(job, range(10))
  4.     print(res)

2.4.3 apply_async()
Pool除了map()外，还有可以返回结果的方式，就是apply_async()。apply_async()中只能传递一个值，它只会放入一个核进行运算，但是传入值时要注意是可迭代的，所以在传入值后需要加逗号, 同时需要用get()方法获取返回值

  1. def multicore():
  2.     pool = mp.Pool()
  3.     res = pool.map(job, range(10))
  4.     print(res)
  5.     res = pool.apply_async(job, (2,))
  6.     # 用get获得结果
  7.     print(res.get())
  8.  
  9. #运行结果
  10. [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
  11. 4

2.4.4 apply_async()输出多个结果
在apply_async()中多传入几个值
res = pool.apply_async(job, (2,3,4,)) #报错 TypeError: job() takes exactly 1 argument (3 given) 即apply_async()只能输入一组参数。
将apply_async() 放入迭代器中，定义一个新的multi_res
multi_res = [pool.apply_async(job, (i,)) for i in range(10)]
取出值时需要一个一个取出来
print([res.get() for res in multi_res])
合并代码

  1. def multicore():
  2.     pool = mp.Pool()
  3.     res = pool.map(job, range(10))
  4.     print(res)
  5.     res = pool.apply_async(job, (2,))
  6.     # 用get获得结果
  7.     print(res.get())
  8.     # 迭代器，i=0时apply一次，i=1时apply一次等等
  9.     multi_res = [pool.apply_async(job, (i,)) for i in range(10)]
  10.     # 从迭代器中取出
  11.     print([res.get() for res in multi_res])
  12. #运行结果
  13.  
  14. [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]  # map()
  15. 4
  16. [0, 1, 4, 9, 16, 25, 36, 49, 64, 81] # multi_res
  • Pool默认调用是CPU的核数，传入processes参数可自定义CPU核数
  • map() 放入迭代参数，返回多个结果
  • apply_async()只能放入一组参数，并返回一个结果，如果想得到map()的效果需要通过迭代
2.5 共享内存 shared memory

2.5.1 Shared Value
使用Value数据存储在一个共享的内存表中。

  1. import multiprocessing as mp
  2.  
  3. value1 = mp.Value('i', 0)
  4. value2 = mp.Value('d', 3.14)

其中d和i参数用来设置数据类型的，d表示一个双精浮点类型，i表示一个带符号的整型
2.5.2 Shared Array
在Python的mutiprocessing中，有还有一个Array类，可以和共享内存交互，来实现在进程之间共享数据
array = mp.Array('i', [1, 2, 3, 4])
这里的Array和numpy中的不同，它只能是一维的，不能是多维的。同样和Value 一样，需要定义数据形式，否则会报错。

2.6 进程锁

2.6.1 不加锁

  1. import multiprocessing as mp
  2.  
  3. def job(x):
  4.     return x*x
  5.  
  6. def multicore():
  7.     pool = mp.Pool()
  8.     res = pool.map(job, range(10))
  9.     print(res)
  10.     res = pool.apply_async(job, (2,))
  11.     # 用get获得结果
  12.     print(res.get())
  13.     # 迭代器，i=0时apply一次，i=1时apply一次等等
  14.     multi_res = [pool.apply_async(job, (i,)) for i in range(10)]
  15.     # 从迭代器中取出
  16.     print([res.get() for res in multi_res])
  17.  
  18. if __name__ == '__main__':
  19.     multicore()
  20. #输出
  21. 1
  22. 4
  23. 5
  24. 8
  25. 9
  26. 12
  27. 13
  28. 16
  29. 17
  30. 20

上面代码中定义了一个共享变量v，两个进程都可以对它进行操作。 在job()中我们想让v每隔0.1秒输出一次累加num的结果，但是在两个进程p1和p2 中设定了不同的累加值。所以来看下这两个进程是否会出现冲突。
2.6.2 加锁

  1. import multiprocessing as mp
  2. import time
  3.  
  4. def job(v, num, l):
  5.     l.acquire() # 锁住
  6.     for _ in range(5):
  7.         time.sleep(0.1)
  8.         v.value += num # 获取共享内存
  9.         print(v.value)
  10.     l.release() # 释放
  11.  
  12. def multicore():
  13.     l = mp.Lock() # 定义一个进程锁
  14.     v = mp.Value('i', 0) # 定义共享内存
  15.     p1 = mp.Process(target=job, args=(v,1,l)) # 需要将lock传入
  16.     p2 = mp.Process(target=job, args=(v,3,l))
  17.     p1.start()
  18.     p2.start()
  19.     p1.join()
  20.     p2.join()
  21.  
  22. if __name__ == '__main__':
  23.     multicore()
  24. #运行一下，看看是否还会出现抢占资源的情况
  25.  
  26. 1
  27. 2
  28. 3
  29. 4
  30. 5
  31. 8
  32. 11
  33. 14
  34. 17
  35. 20

运行结果显示，进程锁保证了进程p1的完整运行，然后才进行了进程p2的运行

转载于:https://www.cnblogs.com/eugene0/p/11546028.html

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/羊村懒王/article/detail/151503
推荐阅读
相关标签

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

  

闽ICP备14008679号