快捷導(dǎo)航

python實(shí)現(xiàn)通過隊(duì)列完成進(jìn)程間的多任務(wù)功能示例

更新時(shí)間：2019年10月28日 10:14:18 作者：獨(dú)聽鐘聲晚

這篇文章主要介紹了python實(shí)現(xiàn)通過隊(duì)列完成進(jìn)程間的多任務(wù)功能,結(jié)合實(shí)例形式分析了Python隊(duì)列完成進(jìn)程間的多任務(wù)以及進(jìn)程池pool相關(guān)操作技巧,需要的朋友可以參考下

本文實(shí)例講述了python實(shí)現(xiàn)通過隊(duì)列完成進(jìn)程間的多任務(wù)功能。分享給大家供大家參考，具體如下：

1.通過隊(duì)列完成進(jìn)程間的多任務(wù)

import multiprocessing


def download_data(q):
  """下載數(shù)據(jù)"""
  # 模擬從網(wǎng)上下載數(shù)據(jù)
  data = [11, 22, 33, 44]

  # 向隊(duì)列中寫入數(shù)據(jù)
  for temp in data:
    q.put(temp)
  print("----數(shù)據(jù)下載完成并且已存入隊(duì)列----")


def analysis_data(q):
  """數(shù)據(jù)處理"""
  waitting_analysis_data = list()
  # 從隊(duì)列中獲取數(shù)據(jù)
  while True:
    data = q.get()
    waitting_analysis_data.append(data)
    if q.empty():
      break
  print(waitting_analysis_data)


def main():
  # 1.創(chuàng)建一個(gè)隊(duì)列
  q = multiprocessing.Queue()
  q1 = multiprocessing.Process(target=download_data, args=(q,))
  q2 = multiprocessing.Process(target=analysis_data, args=(q,))
  q1.start()
  q2.start()


if __name__ == '__main__':
  main()

在這里插入圖片描述

2.進(jìn)程池pool

在程序?qū)嶋H處理問題過程中，忙時(shí)會(huì)有成千上萬的任務(wù)需要被執(zhí)行，閑時(shí)可能只有零星任務(wù)。那么在成千上萬個(gè)任務(wù)需要被執(zhí)行的時(shí)候，我們就需要去創(chuàng)建成千上萬個(gè)進(jìn)程么？首先，創(chuàng)建進(jìn)程需要消耗時(shí)間，銷毀進(jìn)程也需要消耗時(shí)間。第二即便開啟了成千上萬的進(jìn)程，操作系統(tǒng)也不能讓他們同時(shí)執(zhí)行，這樣反而會(huì)影響程序的效率。因此我們不能無限制的根據(jù)任務(wù)開啟或者結(jié)束進(jìn)程。那么我們要怎么做呢？

在這里，要給大家介紹一個(gè)進(jìn)程池的概念，定義一個(gè)池子，在里面放上固定數(shù)量的進(jìn)程，有需求來了，就拿一個(gè)池中的進(jìn)程來處理任務(wù)，等到處理完畢，進(jìn)程并不關(guān)閉，而是將進(jìn)程再放回進(jìn)程池中繼續(xù)等待任務(wù)。如果有很多任務(wù)需要執(zhí)行，池中的進(jìn)程數(shù)量不夠，任務(wù)就要等待之前的進(jìn)程執(zhí)行任務(wù)完畢歸來，拿到空閑進(jìn)程才能繼續(xù)執(zhí)行。也就是說，池中進(jìn)程的數(shù)量是固定的，那么同一時(shí)間最多有固定數(shù)量的進(jìn)程在運(yùn)行。這樣不會(huì)增加操作系統(tǒng)的調(diào)度難度，還節(jié)省了開閉進(jìn)程的時(shí)間，也一定程度上能夠?qū)崿F(xiàn)并發(fā)效果。

案例：

from multiprocessing import Pool
import os, time, random


def worker(msg):
  t_start = time.time()
  print("進(jìn)程%s開始執(zhí)行，進(jìn)程號(hào)為%d" % (msg, os.getpid()))
  # random.random()隨機(jī)生成0-1之間的浮點(diǎn)數(shù)
  time.sleep(random.random()*2)
  t_stop = time.time()
  print("進(jìn)程",msg,"執(zhí)行完成,耗時(shí)%0.2f" % (t_stop-t_start))


def main():
  # 定義一個(gè)進(jìn)程池，最大進(jìn)程數(shù)為3
  po = Pool(3)
  for i in range(10):
    # Pool().apply_async(要調(diào)用的目標(biāo)，(傳遞給目標(biāo)的參數(shù)元組，))
    # 每次循環(huán)將會(huì)用空閑出來的子進(jìn)程去調(diào)用目標(biāo)
    po.apply_async(worker,(i,))

  print("----start----")
  # 關(guān)閉進(jìn)程池，關(guān)閉后po不再接受新的請求
  po.close()
# 等待po中所有子進(jìn)程執(zhí)行完成，必須放在close語句之后
  po.join()

  print("----end----")


if __name__ == '__main__':
  main()

在這里插入圖片描述