快捷導(dǎo)航

Python 如何創(chuàng)建一個(gè)線程池

更新時(shí)間：2020年07月28日 14:40:27 作者：David Beazley

這篇文章主要介紹了Python 如何創(chuàng)建一個(gè)線程池，文中講解非常細(xì)致，代碼幫助大家更好的理解和學(xué)習(xí)，感興趣的朋友可以了解下

問(wèn)題

你創(chuàng)建一個(gè)工作者線程池，用來(lái)響應(yīng)客戶端請(qǐng)求或執(zhí)行其他的工作。

解決方案

concurrent.futures 函數(shù)庫(kù)有一個(gè) ThreadPoolExecutor 類可以被用來(lái)完成這個(gè)任務(wù)。下面是一個(gè)簡(jiǎn)單的TCP服務(wù)器，使用了一個(gè)線程池來(lái)響應(yīng)客戶端：

from socket import AF_INET, SOCK_STREAM, socket
from concurrent.futures import ThreadPoolExecutor

def echo_client(sock, client_addr):
  '''
  Handle a client connection
  '''
  print('Got connection from', client_addr)
  while True:
    msg = sock.recv(65536)
    if not msg:
      break
    sock.sendall(msg)
  print('Client closed connection')
  sock.close()

def echo_server(addr):
  pool = ThreadPoolExecutor(128)
  sock = socket(AF_INET, SOCK_STREAM)
  sock.bind(addr)
  sock.listen(5)
  while True:
    client_sock, client_addr = sock.accept()
    pool.submit(echo_client, client_sock, client_addr)

echo_server(('',15000))

如果你想手動(dòng)創(chuàng)建你自己的線程池，通常可以使用一個(gè)Queue來(lái)輕松實(shí)現(xiàn)。下面是一個(gè)稍微不同但是手動(dòng)實(shí)現(xiàn)的例子：

from socket import socket, AF_INET, SOCK_STREAM
from threading import Thread
from queue import Queue

def echo_client(q):
  '''
  Handle a client connection
  '''
  sock, client_addr = q.get()
  print('Got connection from', client_addr)
  while True:
    msg = sock.recv(65536)
    if not msg:
      break
    sock.sendall(msg)
  print('Client closed connection')

  sock.close()

def echo_server(addr, nworkers):
  # Launch the client workers
  q = Queue()
  for n in range(nworkers):
    t = Thread(target=echo_client, args=(q,))
    t.daemon = True
    t.start()

  # Run the server
  sock = socket(AF_INET, SOCK_STREAM)
  sock.bind(addr)
  sock.listen(5)
  while True:
    client_sock, client_addr = sock.accept()
    q.put((client_sock, client_addr))

echo_server(('',15000), 128)

使用 ThreadPoolExecutor 相對(duì)于手動(dòng)實(shí)現(xiàn)的一個(gè)好處在于它使得任務(wù)提交者更方便的從被調(diào)用函數(shù)中獲取返回值。例如，你可能會(huì)像下面這樣寫：

from concurrent.futures import ThreadPoolExecutor
import urllib.request

def fetch_url(url):
  u = urllib.request.urlopen(url)
  data = u.read()
  return data

pool = ThreadPoolExecutor(10)
# Submit work to the pool
a = pool.submit(fetch_url, 'http://www.python.org')
b = pool.submit(fetch_url, 'http://www.pypy.org')

# Get the results back
x = a.result()
y = b.result()

例子中返回的handle對(duì)象會(huì)幫你處理所有的阻塞與協(xié)作，然后從工作線程中返回?cái)?shù)據(jù)給你。特別的，a.result() 操作會(huì)阻塞進(jìn)程直到對(duì)應(yīng)的函數(shù)執(zhí)行完成并返回一個(gè)結(jié)果。

討論

通常來(lái)講，你應(yīng)該避免編寫線程數(shù)量可以無(wú)限制增長(zhǎng)的程序。例如，看看下面這個(gè)服務(wù)器：

from threading import Thread
from socket import socket, AF_INET, SOCK_STREAM

def echo_client(sock, client_addr):
  '''
  Handle a client connection
  '''
  print('Got connection from', client_addr)
  while True:
    msg = sock.recv(65536)
    if not msg:
      break
    sock.sendall(msg)
  print('Client closed connection')
  sock.close()

def echo_server(addr, nworkers):
  # Run the server
  sock = socket(AF_INET, SOCK_STREAM)
  sock.bind(addr)
  sock.listen(5)
  while True:
    client_sock, client_addr = sock.accept()
    t = Thread(target=echo_client, args=(client_sock, client_addr))
    t.daemon = True
    t.start()

echo_server(('',15000))

盡管這個(gè)也可以工作，但是它不能抵御有人試圖通過(guò)創(chuàng)建大量線程讓你服務(wù)器資源枯竭而崩潰的攻擊行為。通過(guò)使用預(yù)先初始化的線程池，你可以設(shè)置同時(shí)運(yùn)行線程的上限數(shù)量。

你可能會(huì)關(guān)心創(chuàng)建大量線程會(huì)有什么后果。現(xiàn)代操作系統(tǒng)可以很輕松的創(chuàng)建幾千個(gè)線程的線程池。甚至，同時(shí)幾千個(gè)線程等待工作并不會(huì)對(duì)其他代碼產(chǎn)生性能影響。當(dāng)然了，如果所有線程同時(shí)被喚醒并立即在CPU上執(zhí)行，那就不同了——特別是有了全局解釋器鎖GIL。通常，你應(yīng)該只在I/O處理相關(guān)代碼中使用線程池。

創(chuàng)建大的線程池的一個(gè)可能需要關(guān)注的問(wèn)題是內(nèi)存的使用。例如，如果你在OS X系統(tǒng)上面創(chuàng)建2000個(gè)線程，系統(tǒng)顯示Python進(jìn)程使用了超過(guò)9GB的虛擬內(nèi)存。不過(guò)，這個(gè)計(jì)算通常是有誤差的。當(dāng)創(chuàng)建一個(gè)線程時(shí)，操作系統(tǒng)會(huì)預(yù)留一個(gè)虛擬內(nèi)存區(qū)域來(lái) 放置線程的執(zhí)行棧（通常是8MB大?。?。但是這個(gè)內(nèi)存只有一小片段被實(shí)際映射到真實(shí)內(nèi)存中。因此，Python進(jìn)程使用到的真實(shí)內(nèi)存其實(shí)很小（比如，對(duì)于2000個(gè)線程來(lái)講，只使用到了70MB的真實(shí)內(nèi)存，而不是9GB）。如果你擔(dān)心虛擬內(nèi)存大小，可以使用 threading.stack_size() 函數(shù)來(lái)降低它。例如：

import threading
threading.stack_size(65536)

如果你加上這條語(yǔ)句并再次運(yùn)行前面的創(chuàng)建2000個(gè)線程試驗(yàn)，你會(huì)發(fā)現(xiàn)Python進(jìn)程只使用到了大概210MB的虛擬內(nèi)存，而真實(shí)內(nèi)存使用量沒有變。注意線程棧大小必須至少為32768字節(jié)，通常是系統(tǒng)內(nèi)存頁(yè)大?。?096、8192等）的整數(shù)倍。

以上就是Python 如何創(chuàng)建一個(gè)線程池的詳細(xì)內(nèi)容，更多關(guān)于Python 創(chuàng)建線程池的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: