在__init__上啓動新進程（用於TCP偵聽器 - 服務器）

Question

我正在嘗試爲類Server的每個新實例運行新進程。每個Server實例都應偵聽特定的端口。我有這個（簡體）到目前爲止的代碼：source

class Server(object): 

    def handle(connection, address): 

     print("OK...connected...") 
     try: 
      while True: 
       data = connection.recv(1024) 
       if data == "": 
        break 
       connection.sendall(data) 
     except Exception as e: 
      print(e) 
     finally: 
      connection.close() 

    def __init__(self, port, ip): 

     self.port = port 
     self.ip = ip 
     self.socket = socket(AF_INET, SOCK_STREAM) 
     self.socket.bind((self.ip, self.port)) 
     self.socket.listen(1) 

     while True: 
      print("Listening...") 
      conn, address = self.socket.accept() 
      process = multiprocessing.Process(target=Pmu.handle, args=(conn, address)) 
      process.daemon = True 
      process.start() 

s1 = Server(9001,"127.0.0.1") 
s2 = Server(9002,"127.0.0.1") 

但是當我運行此腳本僅第一服務器S1正在運行並等待連接。如何讓兩臺服務器同時進行監聽？

Answer 1

您當前的服務器實際上是一個SocketServer.ForkingTCPServer，它在其__init__中進入一個緊密循環，共享者接受新連接併爲每個傳入連接創建一個新的子進程。

問題是，__init__永遠不會返回，所以只有一個服務器被實例化，一個套接字被綁定，並且只有一個端口會接受新的請求。

解決此類問題的常見方法是將接受循環移入工作線程。此代碼會是這個樣子：

import multiprocessing 
import threading 
import socket 

class Server(object): 

    def handle(self, connection, address): 
     print("OK...connected...") 
     try: 
      while True: 
       data = connection.recv(1024) 
       if data == "": 
        break 
       connection.sendall(data) 
     except Exception as e: 
      print(e) 
     finally: 
      connection.close() 
      print("Connection closed") 

    def accept_forever(self): 
     while True: 
      # Accept a connection on the bound socket and fork a child process 
      # to handle it. 
      print("Waiting for connection...") 
      conn, address = self.socket.accept() 
      process = multiprocessing.Process(
       target=self.handle, args=(conn, address)) 
      process.daemon = True 
      process.start() 

      # Close the connection fd in the parent, since the child process 
      # has its own reference. 
      conn.close() 

    def __init__(self, port, ip): 
     self.port = port 
     self.ip = ip 
     self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 
     self.socket.bind((self.ip, self.port)) 
     self.socket.listen(1) 

     # Spin up an acceptor thread 
     self.worker = threading.Thread(target=self.accept_forever) 
     self.worker.daemon = True 
     self.worker.start() 

    def join(self): 
     # threading.Thread.join() is not interruptible, so tight loop 
     # in a sleep-based join 
     while self.worker.is_alive(): 
      self.worker.join(0.5) 

# Create two servers that run in the background 
s1 = Server(9001,"127.0.0.1") 
s2 = Server(9002,"127.0.0.1") 

# Wait for servers to shutdown 
s1.join() 
s2.join() 

注意我在這裏悄悄另一個變化：

# Wait for servers to shutdown 
s1.join() 
s2.join() 

使用保存的參考Server的接受工人，我們稱之爲.join()主線程在服務器運行時強制阻止事物。如果沒有這個，由於工作人員的.daemon屬性被設置，您的主程序將立即退出。

另外值得一提的是，這種做法將有一些怪癖：

1. 由於處理函數在單獨的進程正在運行，您將需要共享認真使用隊列，價值，管它們之間的數據結構，和其他multiprocessing構造，如果他們互相依賴。

2. 活動併發連接沒有速率限制;爲每一個請求創建一個新的流程可能會很昂貴，並且可以爲您的服務創建一個簡單的DoSed向量。