文章出處

Python自動化 【第八篇】：Python基礎-Socket編程進階

本節內容：

  1. Socket語法及相關
  2. SocketServer實現多并發

 

1.　　Socket語法及相關

　　sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM,0)

　　參數一：地址簇

　　socket.AF_INET IPv4（默認）
　　socket.AF_INET6 IPv6

　　socket.AF_UNIX 只能夠用于單一的Unix系統進程間通信

　　參數二：類型

　　socket.SOCK_STREAM　　流式socket , for TCP （默認）
　　socket.SOCK_DGRAM　　 數據報式socket , for UDP

　　socket.SOCK_RAW 原始套接字，普通的套接字無法處理ICMP、IGMP等網絡報文，而SOCK_RAW可以；其次，SOCK_RAW也可以處理特殊的IPv4報文；此外，利用原始套接字，可以通過IP_HDRINCL套接字選項由用戶構造IP頭。
　　socket.SOCK_RDM 是一種可靠的UDP形式，即保證交付數據報但不保證順序。SOCK_RAM用來提供對原始協議的低級訪問，在需要執行某些特殊操作時使用，如發送ICMP報文。SOCK_RAM通常僅限于高級用戶或管理員運行的程序使用。
　　socket.SOCK_SEQPACKET 可靠的連續數據包服務

　　參數三：協議

　　0　　（默認）與特定的地址家族相關的協議,如果是 0 ，則系統就會根據地址格式和套接類別,自動選擇一個合適的協議

 

  • sk.bind(address)　　將套接字綁定到地址。address地址的格式取決于地址族。在AF_INET下，以元組（host,port）的形式表示地址。
  • sk.listen(backlog)　　開始監聽傳入連接。backlog指定在拒絕連接之前，可以掛起的最大連接數量。backlog等于5，表示內核已經接到了連接請求，但服務器還沒有調用accept進行處理的連接個數最大為5。這個值不能無限大，因為要在內核中維護連接隊列。
  • sk.setblocking(bool)　　是否阻塞（默認True），如果設置False，那么accept和recv時一旦無數據，則報錯。
  • sk.accept()　　接受連接并返回（conn,address）,其中conn是新的套接字對象，可以用來接收和發送數據。address是連接客戶端的地址。接收TCP 客戶的連接（阻塞式）等待連接的到來。
  • sk.connect(address)　　連接到address處的套接字。一般，address的格式為元組（hostname,port）,如果連接出錯，返回socket.error錯誤。
  • sk.connect_ex(address)　　同上，只不過會有返回值，連接成功時返回 0 ，連接失敗時候返回編碼，例如：10061
  • sk.close()　　關閉套接字
  • sk.recv(bufsize[,flag])　　接受套接字的數據。數據以字符串形式返回，bufsize指定最多可以接收的數量。flag提供有關消息的其他信息，通常可以忽略。
  • sk.recvfrom(bufsize[.flag])　　與recv()類似，但返回值是（data,address）。其中data是包含接收數據的字符串，address是發送數據的套接字地址。
  • sk.send(string[,flag])　　將string中的數據發送到連接的套接字。返回值是要發送的字節數量，該數量可能小于string的字節大小。即：可能未將指定內容全部發送。
  • sk.sendall(string[,flag])　　將string中的數據發送到連接的套接字，但在返回之前會嘗試發送所有數據。成功返回None，失敗則拋出異常。內部通過遞歸調用send，將所有內容發送出去。
  • sk.sendto(string[,flag],address)　　將數據發送到套接字，address是形式為（ipaddr，port）的元組，指定遠程地址。返回值是發送的字節數。該函數主要用于UDP協議。
  • sk.settimeout(timeout)　　設置套接字操作的超時期，timeout是一個浮點數，單位是秒。值為None表示沒有超時期。一般，超時期應該在剛創建套接字時設置，因為它們可能用于連接的操作（如 client 連接最多等待5s ）
  • sk.getpeername()　　返回連接套接字的遠程地址。返回值通常是元組（ipaddr,port）。
  • sk.getsockname()　　返回套接字自己的地址。通常是一個元組(ipaddr,port)
  • sk.fileno()　　套接字的文件描述符

 

2.　　SocketServer實現多并發

　　SocketServer內部使用 IO多路復用 以及 “多線程” 和 “多進程” ，從而實現并發處理多個客戶端請求的Socket服務端。

+------------+

| BaseServer |

+------------+

      |

      v

+-----------+        +------------------+

| TCPServer |------->| UnixStreamServer |

+-----------+        +------------------+

      |

      v

+-----------+        +--------------------+

| UDPServer |------->| UnixDatagramServer |

 

　　SocketServer簡化了網絡服務器的編寫。它有4個類：TCPServer，UDPServer，UnixStreamServer，UnixDatagramServer。
　　這4個類是同步進行處理的，另外通過ForkingMixIn和ThreadingMixIn類來支持異步。 

　　創建socketserver的步驟：

  • 自己創建一個請求處理類，并且這個類要繼承BaseRequestHandler,并且重寫父類里的handle方法
  • 必須實例化TCPserver，并且傳遞server ip和 你上面創建的請求處理類給這個TCPserver
  • server.handle_request() #只處理一個請求 （不建議用）;　　server.serve_forever() #處理多個請求 ，永遠執行。
 

　　ThreadingTCPServer

　　ThreadingTCPServer實現的Soket服務器內部會為每個client創建一個 “線程”，該線程用來和客戶端進行交互。

　　1）ThreadingTCPServer基礎

　　使用ThreadingTCPServer:

　　 
import SocketServer



class MyTCPserver(SocketServer.BaseRequestHandler):

    def handle(self):
        pass

if __name__ == '__main__':
    server = SocketServer.ThreadingTCPServer(('127.0.0.1',5566), MyTCPserver)
    server.serve_forever()
Threading TCP server

　　2）ThreadingTCPServer源碼剖析

　　ThreadingTCPServer的類圖關系如下：

　　

　　內部調用流程為：

  • 啟動服務端程序
  • 執行 TCPServer.__init__ 方法，創建服務端Socket對象并綁定 IP 和 端口
  • 執行 BaseServer.__init__ 方法，將自定義的繼承自SocketServer.BaseRequestHandler 的類 MyRequestHandle賦值給self.RequestHandlerClass
  • 執行 BaseServer.server_forever 方法，While 循環一直監聽是否有客戶端請求到達 ...
  • 當客戶端連接到達服務器
  • 執行 ThreadingMixIn.process_request 方法，創建一個 “線程” 用來處理請求
  • 執行 ThreadingMixIn.process_request_thread 方法
  • 執行 BaseServer.finish_request 方法，執行 self.RequestHandlerClass()  即：執行 自定義 MyRequestHandler 的構造方法（自動調用基類BaseRequestHandler的構造方法，在該構造方法中又會調用 MyRequestHandler的handle方法）

 

　　ThreadingTCPServer相關源碼：

　　BaseServer

　　 
class BaseServer:

    """Base class for server classes.

    Methods for the caller:

    - __init__(server_address, RequestHandlerClass)
    - serve_forever(poll_interval=0.5)
    - shutdown()
    - handle_request()  # if you do not use serve_forever()
    - fileno() -> int   # for select()

    Methods that may be overridden:

    - server_bind()
    - server_activate()
    - get_request() -> request, client_address
    - handle_timeout()
    - verify_request(request, client_address)
    - server_close()
    - process_request(request, client_address)
    - shutdown_request(request)
    - close_request(request)
    - handle_error()

    Methods for derived classes:

    - finish_request(request, client_address)

    Class variables that may be overridden by derived classes or
    instances:

    - timeout
    - address_family
    - socket_type
    - allow_reuse_address

    Instance variables:

    - RequestHandlerClass
    - socket

    """

    timeout = None

    def __init__(self, server_address, RequestHandlerClass):
        """Constructor.  May be extended, do not override."""
        self.server_address = server_address
        self.RequestHandlerClass = RequestHandlerClass
        self.__is_shut_down = threading.Event()
        self.__shutdown_request = False

    def server_activate(self):
        """Called by constructor to activate the server.

        May be overridden.

        """
        pass

    def serve_forever(self, poll_interval=0.5):
        """Handle one request at a time until shutdown.

        Polls for shutdown every poll_interval seconds. Ignores
        self.timeout. If you need to do periodic tasks, do them in
        another thread.
        """
        self.__is_shut_down.clear()
        try:
            while not self.__shutdown_request:
                # XXX: Consider using another file descriptor or
                # connecting to the socket to wake this up instead of
                # polling. Polling reduces our responsiveness to a
                # shutdown request and wastes cpu at all other times.
                r, w, e = _eintr_retry(select.select, [self], [], [],
                                       poll_interval)
                if self in r:
                    self._handle_request_noblock()
        finally:
            self.__shutdown_request = False
            self.__is_shut_down.set()

    def shutdown(self):
        """Stops the serve_forever loop.

        Blocks until the loop has finished. This must be called while
        serve_forever() is running in another thread, or it will
        deadlock.
        """
        self.__shutdown_request = True
        self.__is_shut_down.wait()

    # The distinction between handling, getting, processing and
    # finishing a request is fairly arbitrary.  Remember:
    #
    # - handle_request() is the top-level call.  It calls
    #   select, get_request(), verify_request() and process_request()
    # - get_request() is different for stream or datagram sockets
    # - process_request() is the place that may fork a new process
    #   or create a new thread to finish the request
    # - finish_request() instantiates the request handler class;
    #   this constructor will handle the request all by itself

    def handle_request(self):
        """Handle one request, possibly blocking.

        Respects self.timeout.
        """
        # Support people who used socket.settimeout() to escape
        # handle_request before self.timeout was available.
        timeout = self.socket.gettimeout()
        if timeout is None:
            timeout = self.timeout
        elif self.timeout is not None:
            timeout = min(timeout, self.timeout)
        fd_sets = _eintr_retry(select.select, [self], [], [], timeout)
        if not fd_sets[0]:
            self.handle_timeout()
            return
        self._handle_request_noblock()

    def _handle_request_noblock(self):
        """Handle one request, without blocking.

        I assume that select.select has returned that the socket is
        readable before this function was called, so there should be
        no risk of blocking in get_request().
        """
        try:
            request, client_address = self.get_request()
        except socket.error:
            return
        if self.verify_request(request, client_address):
            try:
                self.process_request(request, client_address)
            except:
                self.handle_error(request, client_address)
                self.shutdown_request(request)

    def handle_timeout(self):
        """Called if no new request arrives within self.timeout.

        Overridden by ForkingMixIn.
        """
        pass

    def verify_request(self, request, client_address):
        """Verify the request.  May be overridden.

        Return True if we should proceed with this request.

        """
        return True

    def process_request(self, request, client_address):
        """Call finish_request.

        Overridden by ForkingMixIn and ThreadingMixIn.

        """
        self.finish_request(request, client_address)
        self.shutdown_request(request)

    def server_close(self):
        """Called to clean-up the server.

        May be overridden.

        """
        pass

    def finish_request(self, request, client_address):
        """Finish one request by instantiating RequestHandlerClass."""
        self.RequestHandlerClass(request, client_address, self)

    def shutdown_request(self, request):
        """Called to shutdown and close an individual request."""
        self.close_request(request)

    def close_request(self, request):
        """Called to clean up an individual request."""
        pass

    def handle_error(self, request, client_address):
        """Handle an error gracefully.  May be overridden.

        The default is to print a traceback and continue.

        """
        print
        '-' * 40
        print
        'Exception happened during processing of request from',
        print
        client_address
        import traceback
        traceback.print_exc()  # XXX But this goes to stderr!
        print
        '-' * 40
base server 

　　TCPServer

　　 
class TCPServer(BaseServer):

    """Base class for various socket-based server classes.

    Defaults to synchronous IP stream (i.e., TCP).

    Methods for the caller:

    - __init__(server_address, RequestHandlerClass, bind_and_activate=True)
    - serve_forever(poll_interval=0.5)
    - shutdown()
    - handle_request()  # if you don't use serve_forever()
    - fileno() -> int   # for select()

    Methods that may be overridden:

    - server_bind()
    - server_activate()
    - get_request() -> request, client_address
    - handle_timeout()
    - verify_request(request, client_address)
    - process_request(request, client_address)
    - shutdown_request(request)
    - close_request(request)
    - handle_error()

    Methods for derived classes:

    - finish_request(request, client_address)

    Class variables that may be overridden by derived classes or
    instances:

    - timeout
    - address_family
    - socket_type
    - request_queue_size (only for stream sockets)
    - allow_reuse_address

    Instance variables:

    - server_address
    - RequestHandlerClass
    - socket

    """

    address_family = socket.AF_INET

    socket_type = socket.SOCK_STREAM

    request_queue_size = 5

    allow_reuse_address = False

    def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True):
        """Constructor.  May be extended, do not override."""
        BaseServer.__init__(self, server_address, RequestHandlerClass)
        self.socket = socket.socket(self.address_family,
                                    self.socket_type)
        if bind_and_activate:
            try:
                self.server_bind()
                self.server_activate()
            except:
                self.server_close()
                raise

    def server_bind(self):
        """Called by constructor to bind the socket.

        May be overridden.

        """
        if self.allow_reuse_address:
            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.bind(self.server_address)
        self.server_address = self.socket.getsockname()

    def server_activate(self):
        """Called by constructor to activate the server.

        May be overridden.

        """
        self.socket.listen(self.request_queue_size)

    def server_close(self):
        """Called to clean-up the server.

        May be overridden.

        """
        self.socket.close()

    def fileno(self):
        """Return socket file number.

        Interface required by select().

        """
        return self.socket.fileno()

    def get_request(self):
        """Get the request and client address from the socket.

        May be overridden.

        """
        return self.socket.accept()

    def shutdown_request(self, request):
        """Called to shutdown and close an individual request."""
        try:
            # explicitly shutdown.  socket.close() merely releases
            # the socket and waits for GC to perform the actual close.
            request.shutdown(socket.SHUT_WR)
        except socket.error:
            pass  # some platforms may raise ENOTCONN here
        self.close_request(request)

    def close_request(self, request):
        """Called to clean up an individual request."""
        request.close()
TCP server

 　　ThreadingMixIn

　　 
class ThreadingMixIn:
    """Mix-in class to handle each request in a new thread."""

    # Decides how threads will act upon termination of the
    # main process
    daemon_threads = False

    def process_request_thread(self, request, client_address):
        """Same as in BaseServer but as a thread.

        In addition, exception handling is done here.

        """
        try:
            self.finish_request(request, client_address)
            self.shutdown_request(request)
        except:
            self.handle_error(request, client_address)
            self.shutdown_request(request)

    def process_request(self, request, client_address):
        """Start a new thread to process the request."""
        t = threading.Thread(target=self.process_request_thread,
                             args=(request, client_address))
        t.daemon = self.daemon_threads
        t.start()
threading mix-in

　　ThreadingTCPServer

　　 
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass
threading TCP server 

　　RequestHandler相關源碼:

　　 
  BaseRequestHandler

class BaseRequestHandler:

    """Base class for request handler classes.

    This class is instantiated for each request to be handled.  The
    constructor sets the instance variables request, client_address
    and server, and then calls the handle() method.  To implement a
    specific service, all you need to do is to derive a class which
    defines a handle() method.

    The handle() method can find the request as self.request, the
    client address as self.client_address, and the server (in case it
    needs access to per-server information) as self.server.  Since a
    separate instance is created for each request, the handle() method
    can define arbitrary other instance variariables.

    """

    def __init__(self, request, client_address, server):
        self.request = request
        self.client_address = client_address
        self.server = server
        self.setup()
        try:
            self.handle()
        finally:
            self.finish()

    def setup(self):
        pass

    def handle(self):
        pass

    def finish(self):
        pass
request handler

 

　　SocketServer的ThreadingTCPServer之所以可以同時處理請求得益于 select 和 Threading 兩個東西，其實本質上就是在服務器端為每一個客戶端創建一個線程，當前線程用來處理對應客戶端的請求，所以，可以支持同時n個客戶端鏈接（長連接）。

 

　　socketserver 實例：

　　server side：

　　 
import socketserver


class MyTCPHandler(socketserver.BaseRequestHandler):
    """
    The request handler class for our server.

    It is instantiated once per connection to the server, and must
    override the handle() method to implement communication to the
    client.
    """

    def handle(self):
        # self.request is the TCP socket connected to the client
        self.data = self.request.recv(1024).strip()
        print("{} wrote:".format(self.client_address[0]))
        print(self.data)
        # just send back the same data, but upper-cased
        self.request.sendall(self.data.upper())


if __name__ == "__main__":
    HOST, PORT = "localhost", 9999

    # Create the server, binding to localhost on port 9999
    server = socketserver.ThreadingTCPServer((HOST, PORT), MyTCPHandler)

    # Activate the server; this will keep running until you
    # interrupt the program with Ctrl-C
    server.serve_forever()
server side

　　client side：

　　 
import socket
import sys

HOST, PORT = "localhost", 9999
data = " ".join(sys.argv[1:])

# Create a socket (SOCK_STREAM means a TCP socket)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

try:
    # Connect to server and send data
    sock.connect((HOST, PORT))
    sock.sendall(bytes(data + "\n", "utf-8"))

    # Receive data from the server and shut down
    received = str(sock.recv(1024), "utf-8")
finally:
    sock.close()

print("Sent:     {}".format(data))
print("Received: {}".format(received))
client side

文章列表


不含病毒。www.avast.com
arrow
arrow
    全站熱搜
    創作者介紹
    創作者 大師兄 的頭像
    大師兄

    IT工程師數位筆記本

    大師兄 發表在 痞客邦 留言(0) 人氣()

    E-mail轉寄