當前位置：首頁 > 编程语言 > python >内容正文

python

Python开发【第八篇】：网络编程 Socket

發布時間：2025/7/14 python 37 豆豆

生活随笔收集整理的這篇文章主要介紹了 Python开发【第八篇】：网络编程 Socket 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

Socket

socket通常也稱作"套接字"，用于描述IP地址和端口，是一個通信鏈的句柄，應用程序通常通過"套接字"向網絡發出請求或者應答網絡請求。

socket起源于Unix，而Unix/Linux基本哲學之一就是“一切皆文件”，對于文件用【打開】【讀寫】【關閉】模式來操作。socket就是該模式的一個實現，socket即是一種特殊的文件，一些socket函數就是對其進行的操作（讀/寫IO、打開、關閉）

socket和file的區別：

file模塊是針對某個指定文件進行【打開】【讀寫】【關閉】
socket模塊是針對服務器端和客戶端Socket 進行【打開】【讀寫】【關閉】

#!/usr/bin/env python # -*- coding:utf-8 -*-import socket ip_port = ('127.0.0.1',9999) sk = socket.socket() sk.bind(ip_port) sk.listen(5) while True: print 'server waiting...' conn,addr = sk.accept() client_data = conn.recv(1024) print client_data conn.sendall('不要回答,不要回答,不要回答') conn.close() socket server #!/usr/bin/env python # -*- coding:utf-8 -*- import socket ip_port = ('127.0.0.1',9999) sk = socket.socket() sk.connect(ip_port) sk.sendall('請求占領地球') server_reply = sk.recv(1024) print server_reply sk.close() socket client

WEB服務應用：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

#!/usr/bin/env python #coding:utf-8 import socket def handle_request(client): ????buf = client.recv(1024) ????client.send("HTTP/1.1 200 OK\r\n\r\n") ????client.send("Hello, World") def main(): ????sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) ????sock.bind(('localhost',8080)) ????sock.listen(5) ????while True: ????????connection, address = sock.accept() ????????handle_request(connection) ????????connection.close() if __name__ == '__main__': ??main()

更多功能

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 ，則系統就會根據地址格式和套接類別,自動選擇一個合適的協議

import socket ip_port = ('127.0.0.1',9999) sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0) sk.bind(ip_port) while True: data = sk.recv(1024) print data import socket ip_port = ('127.0.0.1',9999) sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0) while True: inp = raw_input('數據：').strip() if inp == 'exit': break sk.sendto(inp,ip_port) sk.close() UDP Demo

sk.bind(address)

　　s.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()

　　套接字的文件描述符

# 服務端 import socket ip_port = ('127.0.0.1',9999) sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0) sk.bind(ip_port) while True: data,(host,port) = sk.recvfrom(1024) print(data,host,port) sk.sendto(bytes('ok', encoding='utf-8'), (host,port)) #客戶端 import socket ip_port = ('127.0.0.1',9999) sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0) while True: inp = input('數據：').strip() if inp == 'exit': break sk.sendto(bytes(inp, encoding='utf-8'),ip_port) data = sk.recvfrom(1024) print(data) sk.close() UDP

實例：智能機器人

#!/usr/bin/env python # -*- coding:utf-8 -*-import socket ip_port = ('127.0.0.1',8888) sk = socket.socket() sk.bind(ip_port) sk.listen(5) while True: conn,address = sk.accept() conn.sendall('歡迎致電 10086，請輸入1xxx,0轉人工服務.') Flag = True while Flag: data = conn.recv(1024) if data == 'exit': Flag = False elif data == '0': conn.sendall('通過可能會被錄音.balabala一大推') else: conn.sendall('請重新輸入.') conn.close() 服務端 #!/usr/bin/env python # -*- coding:utf-8 -*-import socket ip_port = ('127.0.0.1',8005) sk = socket.socket() sk.connect(ip_port) sk.settimeout(5) while True: data = sk.recv(1024) print 'receive:',data inp = raw_input('please input:') sk.sendall(inp) if inp == 'exit': break sk.close() 客戶端

IO多路復用

I/O多路復用指：通過一種機制，可以監視多個描述符，一旦某個描述符就緒（一般是讀就緒或者寫就緒），能夠通知程序進行相應的讀寫操作。

Linux

Linux中的?select，poll，epoll 都是IO多路復用的機制。

+ View Code?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

select select最早于1983年出現在4.2BSD中，它通過一個select()系統調用來監視多個文件描述符的數組，當select()返回后，該數組中就緒的文件描述符便會被內核修改標志位，使得進程可以獲得這些文件描述符從而進行后續的讀寫操作。 select目前幾乎在所有的平臺上支持，其良好跨平臺支持也是它的一個優點，事實上從現在看來，這也是它所剩不多的優點之一。 select的一個缺點在于單個進程能夠監視的文件描述符的數量存在最大限制，在Linux上一般為1024，不過可以通過修改宏定義甚至重新編譯內核的方式提升這一限制。另外，select()所維護的存儲大量文件描述符的數據結構，隨著文件描述符數量的增大，其復制的開銷也線性增長。同時，由于網絡響應時間的延遲使得大量TCP連接處于非活躍狀態，但調用select()會對所有socket進行一次線性掃描，所以這也浪費了一定的開銷。 poll poll在1986年誕生于System V Release 3，它和select在本質上沒有多大差別，但是poll沒有最大文件描述符數量的限制。 poll和select同樣存在一個缺點就是，包含大量文件描述符的數組被整體復制于用戶態和內核的地址空間之間，而不論這些文件描述符是否就緒，它的開銷隨著文件描述符數量的增加而線性增大。另外，select()和poll()將就緒的文件描述符告訴進程后，如果進程沒有對其進行IO操作，那么下次調用select()和poll()的時候將再次報告這些文件描述符，所以它們一般不會丟失就緒的消息，這種方式稱為水平觸發（Level Triggered）。 epoll 直到Linux2.6才出現了由內核直接支持的實現方法，那就是epoll，它幾乎具備了之前所說的一切優點，被公認為Linux2.6下性能最好的多路I/O就緒通知方法。 epoll可以同時支持水平觸發和邊緣觸發（Edge Triggered，只告訴進程哪些文件描述符剛剛變為就緒狀態，它只說一遍，如果我們沒有采取行動，那么它將不會再次告知，這種方式稱為邊緣觸發），理論上邊緣觸發的性能要更高一些，但是代碼實現相當復雜。 epoll同樣只告知那些就緒的文件描述符，而且當我們調用epoll_wait()獲得就緒文件描述符時，返回的不是實際的描述符，而是一個代表就緒描述符數量的值，你只需要去epoll指定的一個數組中依次取得相應數量的文件描述符即可，這里也使用了內存映射（mmap）技術，這樣便徹底省掉了這些文件描述符在系統調用時復制的開銷。另一個本質的改進在于epoll采用基于事件的就緒通知方式。在select/poll中，進程只有在調用一定的方法后，內核才對所有監視的文件描述符進行掃描，而epoll事先通過epoll_ctl()來注冊一個文件描述符，一旦基于某個文件描述符就緒時，內核會采用類似callback的回調機制，迅速激活這個文件描述符，當進程調用epoll_wait()時便得到通知。

Python

Python中有一個select模塊，其中提供了：select、poll、epoll三個方法，分別調用系統的?select，poll，epoll 從而實現IO多路復用。

1 2 3 4 5 6

Windows Python： ????提供： select Mac Python： ????提供： select Linux Python： ????提供： select、poll、epoll

注意：網絡操作、文件操作、終端操作等均屬于IO操作，對于windows只支持Socket操作，其他系統支持其他IO操作，但是無法檢測普通文件操作自動上次讀取是否已經變化。

對于select方法：

1 2 3 4 5 6 7 8 9 10 11

句柄列表11, 句柄列表22, 句柄列表33 = select.select(句柄序列1, 句柄序列2, 句柄序列3, 超時時間) 參數：可接受四個參數（前三個必須）返回值：三個列表 select方法用來監視文件句柄，如果句柄發生變化，則獲取該句柄。 1、當參數1 序列中的句柄發生可讀時（accetp和read），則獲取發生變化的句柄并添加到返回值1 序列中 2、當參數2 序列中含有句柄時，則將該序列中所有的句柄添加到返回值2 序列中 3、當參數3 序列中的句柄發生錯誤時，則將該發生錯誤的句柄添加到返回值3 序列中 4、當超時時間未設置，則select會一直阻塞，直到監聽的句柄發生變化 ???當超時時間＝ 1時，那么如果監聽的句柄均無任何變化，則select會阻塞 1 秒，之后返回三個空列表，如果監聽的句柄有變化，則直接執行。

#!/usr/bin/env python # -*- coding:utf-8 -*-import select import threading import sys while True: readable, writeable, error = select.select([sys.stdin,],[],[],1) if sys.stdin in readable: print 'select get stdin',sys.stdin.readline() 利用select監聽終端操作實例 #!/usr/bin/env python # -*- coding:utf-8 -*-import socket import select sk1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sk1.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sk1.bind(('127.0.0.1',8002)) sk1.listen(5) sk1.setblocking(0) inputs = [sk1,] while True: readable_list, writeable_list, error_list = select.select(inputs, [], inputs, 1) for r in readable_list: # 當客戶端第一次連接服務端時 if sk1 == r: print 'accept' request, address = r.accept() request.setblocking(0) inputs.append(request) # 當客戶端連接上服務端之后，再次發送數據時 else: received = r.recv(1024) # 當正常接收客戶端發送的數據時 if received: print 'received data:', received # 當客戶端關閉程序時 else: inputs.remove(r) sk1.close() 利用select實現偽同時處理多個Socket客戶端請求：服務端 #!/usr/bin/env python # -*- coding:utf-8 -*- import socket ip_port = ('127.0.0.1',8002) sk = socket.socket() sk.connect(ip_port) while True: inp = raw_input('please input:') sk.sendall(inp) sk.close() 利用select實現偽同時處理多個Socket客戶端請求：客戶端

此處的Socket服務端相比與原生的Socket，他支持當某一個請求不再發送數據時，服務器端不會等待而是可以去處理其他請求的數據。但是，如果每個請求的耗時比較長時，select版本的服務器端也無法完成同時操作。

#!/usr/bin/env python #coding:utf8''' 服務器的實現采用select的方式 ''' import select import socket import sys import Queue #創建套接字并設置該套接字為非阻塞模式 server = socket.socket(socket.AF_INET,socket.SOCK_STREAM) server.setblocking(0) #綁定套接字 server_address = ('localhost',10000) print >>sys.stderr,'starting up on %s port %s'% server_address server.bind(server_address) #將該socket變成服務模式 #backlog等于5，表示內核已經接到了連接請求，但服務器還沒有調用accept進行處理的連接個數最大為5 #這個值不能無限大，因為要在內核中維護連接隊列 server.listen(5) #初始化讀取數據的監聽列表,最開始時希望從server這個套接字上讀取數據 inputs = [server] #初始化寫入數據的監聽列表，最開始并沒有客戶端連接進來，所以列表為空 outputs = [] #要發往客戶端的數據 message_queues = {} while inputs: print >>sys.stderr,'waiting for the next event' #調用select監聽所有監聽列表中的套接字，并將準備好的套接字加入到對應的列表中 readable,writable,exceptional = select.select(inputs,outputs,inputs)#列表中的socket 套接字如果是文件呢？ #監控文件句柄有某一處發生了變化可寫可讀異常屬于Linux中的網絡編程 #屬于同步I/O操作，屬于I/O復用模型的一種 #rlist--等待到準備好讀 #wlist--等待到準備好寫 #xlist--等待到一種異常 #處理可讀取的套接字 ''' 如果server這個套接字可讀，則說明有新鏈接到來此時在server套接字上調用accept,生成一個與客戶端通訊的套接字并將與客戶端通訊的套接字加入inputs列表，下一次可以通過select檢查連接是否可讀然后在發往客戶端的緩沖中加入一項，鍵名為:與客戶端通訊的套接字，鍵值為空隊列 select系統調用是用來讓我們的程序監視多個文件句柄(file descrīptor)的狀態變化的。程序會停在select這里等待，直到被監視的文件句柄有某一個或多個發生了狀態改變 ''' ''' 若可讀的套接字不是server套接字,有兩種情況:一種是有數據到來，另一種是鏈接斷開如果有數據到來,先接收數據,然后將收到的數據填入往客戶端的緩存區中的對應位置，最后將于客戶端通訊的套接字加入到寫數據的監聽列表: 如果套接字可讀.但沒有接收到數據，則說明客戶端已經斷開。這時需要關閉與客戶端連接的套接字進行資源清理 ''' for s in readable: if s is server: connection,client_address = s.accept() print >>sys.stderr,'connection from',client_address connection.setblocking(0)#設置非阻塞 inputs.append(connection) message_queues[connection] = Queue.Queue() else: data = s.recv(1024) if data: print >>sys.stderr,'received "%s" from %s'% \ (data,s.getpeername()) message_queues[s].put(data) if s not in outputs: outputs.append(s) else: print >>sys.stderr,'closing',client_address if s in outputs: outputs.remove(s) inputs.remove(s) s.close() del message_queues[s] #處理可寫的套接字 ''' 在發送緩沖區中取出響應的數據，發往客戶端。如果沒有數據需要寫，則將套接字從發送隊列中移除，select中不再監視 ''' for s in writable: try: next_msg = message_queues[s].get_nowait() except Queue.Empty: print >>sys.stderr,' ',s,getpeername(),'queue empty' outputs.remove(s) else: print >>sys.stderr,'sending "%s" to %s'% \ (next_msg,s.getpeername()) s.send(next_msg) #處理異常情況 for s in exceptional: for s in exceptional: print >>sys.stderr,'exception condition on',s.getpeername() inputs.remove(s) if s in outputs: outputs.remove(s) s.close() del message_queues[s] 基于select實現socket服務端

SocketServer模塊

SocketServer內部使用 IO多路復用以及 “多線程” 和 “多進程” ，從而實現并發處理多個客戶端請求的Socket服務端。即：每個客戶端請求連接到服務器時，Socket服務端都會在服務器是創建一個“線程”或者“進程” 專門負責處理當前客戶端的所有請求。

ThreadingTCPServer

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

1、ThreadingTCPServer基礎

使用ThreadingTCPServer:

創建一個繼承自 SocketServer.BaseRequestHandler 的類
類中必須定義一個名稱為 handle 的方法
啟動ThreadingTCPServer

#!/usr/bin/env python # -*- coding:utf-8 -*- import SocketServer class MyServer(SocketServer.BaseRequestHandler): def handle(self): # print self.request,self.client_address,self.server conn = self.request conn.sendall('歡迎致電 10086，請輸入1xxx,0轉人工服務.') Flag = True while Flag: data = conn.recv(1024) if data == 'exit': Flag = False elif data == '0': conn.sendall('通過可能會被錄音.balabala一大推') else: conn.sendall('請重新輸入.') if __name__ == '__main__': server = SocketServer.ThreadingTCPServer(('127.0.0.1',8009),MyServer) server.serve_forever() SocketServer實現服務器 #!/usr/bin/env python # -*- coding:utf-8 -*-import socket ip_port = ('127.0.0.1',8009) sk = socket.socket() sk.connect(ip_port) sk.settimeout(5) while True: data = sk.recv(1024) print 'receive:',data inp = raw_input('please input:') sk.sendall(inp) if inp == 'exit': break sk.close() 客戶端

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相關源碼：

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 orinstances:- timeout- address_family- socket_type- allow_reuse_addressInstance 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 BaseServer 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 orinstances:- timeout- address_family- socket_type- request_queue_size (only for stream sockets)- allow_reuse_addressInstance 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() TCPServer 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() ThreadingMixIn class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass ThreadingTCPServer

RequestHandler相關源碼

class BaseRequestHandler:"""Base class for request handler classes.This class is instantiated for each request to be handled. Theconstructor sets the instance variables request, client_addressand server, and then calls the handle() method. To implement aspecific service, all you need to do is to derive a class whichdefines a handle() method.The handle() method can find the request as self.request, theclient address as self.client_address, and the server (in case itneeds access to per-server information) as self.server. Since aseparate instance is created for each request, the handle() methodcan 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 SocketServer.BaseRequestHandler

實例：

#!/usr/bin/env python # -*- coding:utf-8 -*- import SocketServer class MyServer(SocketServer.BaseRequestHandler): def handle(self): # print self.request,self.client_address,self.server conn = self.request conn.sendall('歡迎致電 10086，請輸入1xxx,0轉人工服務.') Flag = True while Flag: data = conn.recv(1024) if data == 'exit': Flag = False elif data == '0': conn.sendall('通過可能會被錄音.balabala一大推') else: conn.sendall('請重新輸入.') if __name__ == '__main__': server = SocketServer.ThreadingTCPServer(('127.0.0.1',8009),MyServer) server.serve_forever() 服務端 #!/usr/bin/env python # -*- coding:utf-8 -*-import socket ip_port = ('127.0.0.1',8009) sk = socket.socket() sk.connect(ip_port) sk.settimeout(5) while True: data = sk.recv(1024) print 'receive:',data inp = raw_input('please input:') sk.sendall(inp) if inp == 'exit': break sk.close() 客戶端

源碼精簡：

import socket import threading import select def process(request, client_address): print request,client_address conn = request conn.sendall('歡迎致電 10086，請輸入1xxx,0轉人工服務.') flag = True while flag: data = conn.recv(1024) if data == 'exit': flag = False elif data == '0': conn.sendall('通過可能會被錄音.balabala一大推') else: conn.sendall('請重新輸入.') sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sk.bind(('127.0.0.1',8002)) sk.listen(5) while True: r, w, e = select.select([sk,],[],[],1) print 'looping' if sk in r: print 'get request' request, client_address = sk.accept() t = threading.Thread(target=process, args=(request, client_address)) t.daemon = False t.start() sk.close()

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

ForkingTCPServer

ForkingTCPServer和ThreadingTCPServer的使用和執行流程基本一致，只不過在內部分別為請求者建立 “線程” ?和 “進程”。

基本使用：

#!/usr/bin/env python # -*- coding:utf-8 -*- import SocketServer class MyServer(SocketServer.BaseRequestHandler): def handle(self): # print self.request,self.client_address,self.server conn = self.request conn.sendall('歡迎致電 10086，請輸入1xxx,0轉人工服務.') Flag = True while Flag: data = conn.recv(1024) if data == 'exit': Flag = False elif data == '0': conn.sendall('通過可能會被錄音.balabala一大推') else: conn.sendall('請重新輸入.') if __name__ == '__main__': server = SocketServer.ForkingTCPServer(('127.0.0.1',8009),MyServer) server.serve_forever() 服務端 #!/usr/bin/env python # -*- coding:utf-8 -*-import socket ip_port = ('127.0.0.1',8009) sk = socket.socket() sk.connect(ip_port) sk.settimeout(5) while True: data = sk.recv(1024) print 'receive:',data inp = raw_input('please input:') sk.sendall(inp) if inp == 'exit': break sk.close() 客戶端

以上ForkingTCPServer只是將?ThreadingTCPServer 實例中的代碼：

1 2 3

server = SocketServer.ThreadingTCPServer(('127.0.0.1',8009),MyRequestHandler) 變更為： server = SocketServer.ForkingTCPServer(('127.0.0.1',8009),MyRequestHandler)

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

源碼剖析參考 ThreadingTCPServer

Twisted

Twisted是一個事件驅動的網絡框架，其中包含了諸多功能，例如：網絡協議、線程、數據庫管理、網絡操作、電子郵件等。

事件驅動

簡而言之，事件驅動分為二個部分：第一，注冊事件；第二，觸發事件。

自定義事件驅動框架，命名為：“弒君者”：

#!/usr/bin/env python # -*- coding:utf-8 -*-# event_drive.py event_list = [] def run(): for event in event_list: obj = event() obj.execute() class BaseHandler(object): """ 用戶必須繼承該類，從而規范所有類的方法（類似于接口的功能） """ def execute(self): raise Exception('you must overwrite execute') 最牛逼的事件驅動框架

程序員使用“弒君者框架”：

#!/usr/bin/env python # -*- coding:utf-8 -*-from source import event_drive class MyHandler(event_drive.BaseHandler): def execute(self): print 'event-drive execute MyHandler' event_drive.event_list.append(MyHandler) event_drive.run() View Code

如上述代碼，事件驅動只不過是框架規定了執行順序，程序員在使用框架時，可以向原執行順序中注冊“事件”，從而在框架執行時可以出發已注冊的“事件”。

基于事件驅動Socket

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

#!/usr/bin/env python # -*- coding:utf-8 -*- from twisted.internet import protocol from twisted.internet import reactor class Echo(protocol.Protocol): ????def dataReceived(self, data): ????????self.transport.write(data) def main(): ????factory = protocol.ServerFactory() ????factory.protocol = Echo ????reactor.listenTCP(8000,factory) ????reactor.run() if __name__ == '__main__': ????main()

程序執行流程：

運行服務端程序
創建Protocol的派生類Echo
創建ServerFactory對象，并將Echo類封裝到其protocol字段中
執行reactor的 listenTCP 方法，內部使用?tcp.Port 創建socket server對象，并將該對象添加到了 reactor的set類型的字段 _read 中
執行reactor的 run 方法，內部執行 while 循環，并通過 select 來監視?_read 中文件描述符是否有變化，循環中...
客戶端請求到達
執行reactor的?_doReadOrWrite 方法，其內部通過反射調用 tcp.Port 類的?doRead 方法，內部 accept 客戶端連接并創建Server對象實例（用于封裝客戶端socket信息）和?創建 Echo 對象實例（用于處理請求）?，然后調用 Echo 對象實例的?makeConnection 方法，創建連接。
執行 tcp.Server 類的?doRead 方法，讀取數據，
執行 tcp.Server 類的?_dataReceived 方法，如果讀取數據內容為空（關閉鏈接），否則，出發 Echo 的?dataReceived 方法
執行?Echo 的?dataReceived 方法

從源碼可以看出，上述實例本質上使用了事件驅動的方法和 IO多路復用的機制來進行Socket的處理。

#!/usr/bin/env python # -*- coding:utf-8 -*-from twisted.internet import reactor, protocol from twisted.web.client import getPage from twisted.internet import reactor import time class Echo(protocol.Protocol): def dataReceived(self, data): deferred1 = getPage('http://cnblogs.com') deferred1.addCallback(self.printContents) deferred2 = getPage('http://baidu.com') deferred2.addCallback(self.printContents) for i in range(2): time.sleep(1) print 'execute ',i def execute(self,data): self.transport.write(data) def printContents(self,content): print len(content),content[0:100],time.time() def main(): factory = protocol.ServerFactory() factory.protocol = Echo reactor.listenTCP(8000,factory) reactor.run() if __name__ == '__main__': main() 異步IO操作

更多請見：

　　https://twistedmatrix.com/trac 　　http://twistedmatrix.com/documents/current/api/

轉載于:https://www.cnblogs.com/cp-miao/p/5658507.html

總結

以上是生活随笔為你收集整理的Python开发【第八篇】：网络编程 Socket的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： Windows Server 2008
下一篇： Runtime 总结