Python开发【第八篇】：网络编程 Socket
生活随笔
收集整理的這篇文章主要介紹了
Python开发【第八篇】：网络编程 Socket
小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.
Socket
socket通常也稱作"套接字"，用于描述IP地址和端口，是一個通信鏈的句柄，應(yīng)用程序通常通過"套接字"向網(wǎng)絡(luò)發(fā)出請求或者應(yīng)答網(wǎng)絡(luò)請求。
socket起源于Unix，而Unix/Linux基本哲學(xué)之一就是“一切皆文件”，對于文件用【打開】【讀寫】【關(guān)閉】模式來操作。socket就是該模式的一個實現(xiàn)，socket即是一種特殊的文件，一些socket函數(shù)就是對其進行的操作（讀/寫IO、打開、關(guān)閉）
socket和file的區(qū)別：
- file模塊是針對某個指定文件進行【打開】【讀寫】【關(guān)閉】
- socket模塊是針對 服務(wù)器端 和 客戶端Socket 進行【打開】【讀寫】【關(guān)閉】
?
#!/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('請求占領(lǐng)地球') server_reply = sk.recv(1024) print server_reply sk.close() socket clientWEB服務(wù)應(yīng)用：
?| 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)
參數(shù)一：地址簇
socket.AF_INET IPv4（默認(rèn)） socket.AF_INET6 IPv6
socket.AF_UNIX 只能夠用于單一的Unix系統(tǒng)進程間通信
參數(shù)二：類型
socket.SOCK_STREAM 流式socket , for TCP （默認(rèn)） socket.SOCK_DGRAM 數(shù)據(jù)報式socket , for UDP
socket.SOCK_RAW 原始套接字，普通的套接字無法處理ICMP、IGMP等網(wǎng)絡(luò)報文，而SOCK_RAW可以；其次，SOCK_RAW也可以處理特殊的IPv4報文；此外，利用原始套接字，可以通過IP_HDRINCL套接字選項由用戶構(gòu)造IP頭。 socket.SOCK_RDM 是一種可靠的UDP形式，即保證交付數(shù)據(jù)報但不保證順序。SOCK_RAM用來提供對原始協(xié)議的低級訪問，在需要執(zhí)行某些特殊操作時使用，如發(fā)送ICMP報文。SOCK_RAM通常僅限于高級用戶或管理員運行的程序使用。 socket.SOCK_SEQPACKET 可靠的連續(xù)數(shù)據(jù)包服務(wù)
參數(shù)三：協(xié)議
0 （默認(rèn)）與特定的地址家族相關(guān)的協(xié)議,如果是 0 ，則系統(tǒng)就會根據(jù)地址格式和套接類別,自動選擇一個合適的協(xié)議
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('數(shù)據(jù)：').strip() if inp == 'exit': break sk.sendto(inp,ip_port) sk.close() UDP Demosk.bind(address)
s.bind(address) 將套接字綁定到地址。address地址的格式取決于地址族。在AF_INET下，以元組（host,port）的形式表示地址。
sk.listen(backlog)
開始監(jiān)聽傳入連接。backlog指定在拒絕連接之前，可以掛起的最大連接數(shù)量。
? ? ? backlog等于5，表示內(nèi)核已經(jīng)接到了連接請求，但服務(wù)器還沒有調(diào)用accept進行處理的連接個數(shù)最大為5 ? ? ? 這個值不能無限大，因為要在內(nèi)核中維護連接隊列
sk.setblocking(bool)
是否阻塞（默認(rèn)True），如果設(shè)置False，那么accept和recv時一旦無數(shù)據(jù)，則報錯。
sk.accept()
接受連接并返回（conn,address）,其中conn是新的套接字對象，可以用來接收和發(fā)送數(shù)據(jù)。address是連接客戶端的地址。
接收TCP 客戶的連接（阻塞式）等待連接的到來
sk.connect(address)
連接到address處的套接字。一般，address的格式為元組（hostname,port）,如果連接出錯，返回socket.error錯誤。
sk.connect_ex(address)
同上，只不過會有返回值，連接成功時返回 0 ，連接失敗時候返回編碼，例如：10061
sk.close()
關(guān)閉套接字
sk.recv(bufsize[,flag])
接受套接字的數(shù)據(jù)。數(shù)據(jù)以字符串形式返回，bufsize指定最多可以接收的數(shù)量。flag提供有關(guān)消息的其他信息，通?？梢院雎浴?/p>
sk.recvfrom(bufsize[.flag])
與recv()類似，但返回值是（data,address）。其中data是包含接收數(shù)據(jù)的字符串，address是發(fā)送數(shù)據(jù)的套接字地址。
sk.send(string[,flag])
將string中的數(shù)據(jù)發(fā)送到連接的套接字。返回值是要發(fā)送的字節(jié)數(shù)量，該數(shù)量可能小于string的字節(jié)大小。即：可能未將指定內(nèi)容全部發(fā)送。
sk.sendall(string[,flag])
將string中的數(shù)據(jù)發(fā)送到連接的套接字，但在返回之前會嘗試發(fā)送所有數(shù)據(jù)。成功返回None，失敗則拋出異常。
? ? ? 內(nèi)部通過遞歸調(diào)用send，將所有內(nèi)容發(fā)送出去。
sk.sendto(string[,flag],address)
將數(shù)據(jù)發(fā)送到套接字，address是形式為（ipaddr，port）的元組，指定遠(yuǎn)程地址。返回值是發(fā)送的字節(jié)數(shù)。該函數(shù)主要用于UDP協(xié)議。
sk.settimeout(timeout)
設(shè)置套接字操作的超時期，timeout是一個浮點數(shù)，單位是秒。值為None表示沒有超時期。一般，超時期應(yīng)該在剛創(chuàng)建套接字時設(shè)置，因為它們可能用于連接的操作（如 client 連接最多等待5s ）
sk.getpeername()
返回連接套接字的遠(yuǎn)程地址。返回值通常是元組（ipaddr,port）。
sk.getsockname()
返回套接字自己的地址。通常是一個元組(ipaddr,port)
sk.fileno()
套接字的文件描述符
# 服務(wù)端 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('數(shù)據(jù)：').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轉(zhuǎn)人工服務(wù).') Flag = True while Flag: data = conn.recv(1024) if data == 'exit': Flag = False elif data == '0': conn.sendall('通過可能會被錄音.balabala一大推') else: conn.sendall('請重新輸入.') conn.close() 服務(wù)端 #!/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多路復(fù)用
I/O多路復(fù)用指：通過一種機制，可以監(jiān)視多個描述符，一旦某個描述符就緒（一般是讀就緒或者寫就緒），能夠通知程序進行相應(yīng)的讀寫操作。
Linux
Linux中的?select，poll，epoll 都是IO多路復(fù)用的機制。
+ View Code?| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | select select最早于1983年出現(xiàn)在4.2BSD中，它通過一個select()系統(tǒng)調(diào)用來監(jiān)視多個文件描述符的數(shù)組，當(dāng)select()返回后，該數(shù)組中就緒的文件描述符便會被內(nèi)核修改標(biāo)志位，使得進程可以獲得這些文件描述符從而進行后續(xù)的讀寫操作。 select目前幾乎在所有的平臺上支持，其良好跨平臺支持也是它的一個優(yōu)點，事實上從現(xiàn)在看來，這也是它所剩不多的優(yōu)點之一。 select的一個缺點在于單個進程能夠監(jiān)視的文件描述符的數(shù)量存在最大限制，在Linux上一般為1024，不過可以通過修改宏定義甚至重新編譯內(nèi)核的方式提升這一限制。 另外，select()所維護的存儲大量文件描述符的數(shù)據(jù)結(jié)構(gòu)，隨著文件描述符數(shù)量的增大，其復(fù)制的開銷也線性增長。同時，由于網(wǎng)絡(luò)響應(yīng)時間的延遲使得大量TCP連接處于非活躍狀態(tài)，但調(diào)用select()會對所有socket進行一次線性掃描，所以這也浪費了一定的開銷。 poll poll在1986年誕生于System V Release 3，它和select在本質(zhì)上沒有多大差別，但是poll沒有最大文件描述符數(shù)量的限制。 poll和select同樣存在一個缺點就是，包含大量文件描述符的數(shù)組被整體復(fù)制于用戶態(tài)和內(nèi)核的地址空間之間，而不論這些文件描述符是否就緒，它的開銷隨著文件描述符數(shù)量的增加而線性增大。 另外，select()和poll()將就緒的文件描述符告訴進程后，如果進程沒有對其進行IO操作，那么下次調(diào)用select()和poll()的時候?qū)⒃俅螆蟾孢@些文件描述符，所以它們一般不會丟失就緒的消息，這種方式稱為水平觸發(fā)（Level Triggered）。 epoll 直到Linux2.6才出現(xiàn)了由內(nèi)核直接支持的實現(xiàn)方法，那就是epoll，它幾乎具備了之前所說的一切優(yōu)點，被公認(rèn)為Linux2.6下性能最好的多路I/O就緒通知方法。 epoll可以同時支持水平觸發(fā)和邊緣觸發(fā)（Edge Triggered，只告訴進程哪些文件描述符剛剛變?yōu)榫途w狀態(tài)，它只說一遍，如果我們沒有采取行動，那么它將不會再次告知，這種方式稱為邊緣觸發(fā)），理論上邊緣觸發(fā)的性能要更高一些，但是代碼實現(xiàn)相當(dāng)復(fù)雜。 epoll同樣只告知那些就緒的文件描述符，而且當(dāng)我們調(diào)用epoll_wait()獲得就緒文件描述符時，返回的不是實際的描述符，而是一個代表就緒描述符數(shù)量的值，你只需要去epoll指定的一個數(shù)組中依次取得相應(yīng)數(shù)量的文件描述符即可，這里也使用了內(nèi)存映射（mmap）技術(shù)，這樣便徹底省掉了這些文件描述符在系統(tǒng)調(diào)用時復(fù)制的開銷。 另一個本質(zhì)的改進在于epoll采用基于事件的就緒通知方式。在select/poll中，進程只有在調(diào)用一定的方法后，內(nèi)核才對所有監(jiān)視的文件描述符進行掃描，而epoll事先通過epoll_ctl()來注冊一個文件描述符，一旦基于某個文件描述符就緒時，內(nèi)核會采用類似callback的回調(diào)機制，迅速激活這個文件描述符，當(dāng)進程調(diào)用epoll_wait()時便得到通知。 |
Python
Python中有一個select模塊，其中提供了：select、poll、epoll三個方法，分別調(diào)用系統(tǒng)的?select，poll，epoll 從而實現(xiàn)IO多路復(fù)用。
?| 1 2 3 4 5 6 | Windows Python： ????提供： select Mac Python： ????提供： select Linux Python： ????提供： select、poll、epoll |
注意：網(wǎng)絡(luò)操作、文件操作、終端操作等均屬于IO操作，對于windows只支持Socket操作，其他系統(tǒng)支持其他IO操作，但是無法檢測 普通文件操作 自動上次讀取是否已經(jīng)變化。
對于select方法：
?| 1 2 3 4 5 6 7 8 9 10 11 | 句柄列表11, 句柄列表22, 句柄列表33 = select.select(句柄序列1, 句柄序列2, 句柄序列3, 超時時間) 參數(shù)： 可接受四個參數(shù)（前三個必須） 返回值：三個列表 select方法用來監(jiān)視文件句柄，如果句柄發(fā)生變化，則獲取該句柄。 1、當(dāng) 參數(shù)1 序列中的句柄發(fā)生可讀時（accetp和read），則獲取發(fā)生變化的句柄并添加到 返回值1 序列中 2、當(dāng) 參數(shù)2 序列中含有句柄時，則將該序列中所有的句柄添加到 返回值2 序列中 3、當(dāng) 參數(shù)3 序列中的句柄發(fā)生錯誤時，則將該發(fā)生錯誤的句柄添加到 返回值3 序列中 4、當(dāng) 超時時間 未設(shè)置，則select會一直阻塞，直到監(jiān)聽的句柄發(fā)生變化 ???當(dāng) 超時時間 ＝ 1時，那么如果監(jiān)聽的句柄均無任何變化，則select會阻塞 1 秒，之后返回三個空列表，如果監(jiān)聽的句柄有變化，則直接執(zhí)行。 |
此處的Socket服務(wù)端相比與原生的Socket，他支持當(dāng)某一個請求不再發(fā)送數(shù)據(jù)時，服務(wù)器端不會等待而是可以去處理其他請求的數(shù)據(jù)。但是，如果每個請求的耗時比較長時，select版本的服務(wù)器端也無法完成同時操作。
#!/usr/bin/env python #coding:utf8''' 服務(wù)器的實現(xiàn) 采用select的方式 ''' import select import socket import sys import Queue #創(chuàng)建套接字并設(shè)置該套接字為非阻塞模式 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變成服務(wù)模式 #backlog等于5，表示內(nèi)核已經(jīng)接到了連接請求，但服務(wù)器還沒有調(diào)用accept進行處理的連接個數(shù)最大為5 #這個值不能無限大，因為要在內(nèi)核中維護連接隊列 server.listen(5) #初始化讀取數(shù)據(jù)的監(jiān)聽列表,最開始時希望從server這個套接字上讀取數(shù)據(jù) inputs = [server] #初始化寫入數(shù)據(jù)的監(jiān)聽列表，最開始并沒有客戶端連接進來，所以列表為空 outputs = [] #要發(fā)往客戶端的數(shù)據(jù) message_queues = {} while inputs: print >>sys.stderr,'waiting for the next event' #調(diào)用select監(jiān)聽所有監(jiān)聽列表中的套接字，并將準(zhǔn)備好的套接字加入到對應(yīng)的列表中 readable,writable,exceptional = select.select(inputs,outputs,inputs)#列表中的socket 套接字 如果是文件呢？ #監(jiān)控文件句柄有某一處發(fā)生了變化 可寫 可讀 異常屬于Linux中的網(wǎng)絡(luò)編程 #屬于同步I/O操作，屬于I/O復(fù)用模型的一種 #rlist--等待到準(zhǔn)備好讀 #wlist--等待到準(zhǔn)備好寫 #xlist--等待到一種異常 #處理可讀取的套接字 ''' 如果server這個套接字可讀，則說明有新鏈接到來 此時在server套接字上調(diào)用accept,生成一個與客戶端通訊的套接字 并將與客戶端通訊的套接字加入inputs列表，下一次可以通過select檢查連接是否可讀 然后在發(fā)往客戶端的緩沖中加入一項，鍵名為:與客戶端通訊的套接字，鍵值為空隊列 select系統(tǒng)調(diào)用是用來讓我們的程序監(jiān)視多個文件句柄(file descrīptor)的狀態(tài)變化的。程序會停在select這里等待， 直到被監(jiān)視的文件句柄有某一個或多個發(fā)生了狀態(tài)改變 ''' ''' 若可讀的套接字不是server套接字,有兩種情況:一種是有數(shù)據(jù)到來，另一種是鏈接斷開 如果有數(shù)據(jù)到來,先接收數(shù)據(jù),然后將收到的數(shù)據(jù)填入往客戶端的緩存區(qū)中的對應(yīng)位置，最后 將于客戶端通訊的套接字加入到寫數(shù)據(jù)的監(jiān)聽列表: 如果套接字可讀.但沒有接收到數(shù)據(jù)，則說明客戶端已經(jīng)斷開。這時需要關(guān)閉與客戶端連接的套接字 進行資源清理 ''' for s in readable: if s is server: connection,client_address = s.accept() print >>sys.stderr,'connection from',client_address connection.setblocking(0)#設(shè)置非阻塞 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] #處理可寫的套接字 ''' 在發(fā)送緩沖區(qū)中取出響應(yīng)的數(shù)據(jù)，發(fā)往客戶端。 如果沒有數(shù)據(jù)需要寫，則將套接字從發(fā)送隊列中移除，select中不再監(jiān)視 ''' 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實現(xiàn)socket服務(wù)端SocketServer模塊
SocketServer內(nèi)部使用 IO多路復(fù)用 以及 “多線程” 和 “多進程” ，從而實現(xiàn)并發(fā)處理多個客戶端請求的Socket服務(wù)端。即：每個客戶端請求連接到服務(wù)器時，Socket服務(wù)端都會在服務(wù)器是創(chuàng)建一個“線程”或者“進程” 專門負(fù)責(zé)處理當(dāng)前客戶端的所有請求。
ThreadingTCPServer
ThreadingTCPServer實現(xiàn)的Soket服務(wù)器內(nèi)部會為每個client創(chuàng)建一個 “線程”，該線程用來和客戶端進行交互。
1、ThreadingTCPServer基礎(chǔ)
使用ThreadingTCPServer:
- 創(chuàng)建一個繼承自 SocketServer.BaseRequestHandler 的類
- 類中必須定義一個名稱為 handle 的方法
- 啟動ThreadingTCPServer
2、ThreadingTCPServer源碼剖析
ThreadingTCPServer的類圖關(guān)系如下：
?
內(nèi)部調(diào)用流程為：
- 啟動服務(wù)端程序
- 執(zhí)行 TCPServer.__init__ 方法，創(chuàng)建服務(wù)端Socket對象并綁定 IP 和 端口
- 執(zhí)行 BaseServer.__init__ 方法，將自定義的繼承自SocketServer.BaseRequestHandler 的類 MyRequestHandle賦值給 self.RequestHandlerClass
- 執(zhí)行 BaseServer.server_forever 方法，While 循環(huán)一直監(jiān)聽是否有客戶端請求到達(dá) ...
- 當(dāng)客戶端連接到達(dá)服務(wù)器
- 執(zhí)行 ThreadingMixIn.process_request 方法，創(chuàng)建一個 “線程” 用來處理請求
- 執(zhí)行?ThreadingMixIn.process_request_thread 方法
- 執(zhí)行 BaseServer.finish_request 方法，執(zhí)行?self.RequestHandlerClass() ?即：執(zhí)行 自定義 MyRequestHandler 的構(gòu)造方法（自動調(diào)用基類BaseRequestHandler的構(gòu)造方法，在該構(gòu)造方法中又會調(diào)用 MyRequestHandler的handle方法）
ThreadingTCPServer相關(guān)源碼：
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 ThreadingTCPServerRequestHandler相關(guān)源碼
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轉(zhuǎn)人工服務(wù).') 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() 服務(wù)端 #!/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轉(zhuǎn)人工服務(wù).') 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 兩個東西，其實本質(zhì)上就是在服務(wù)器端為每一個客戶端創(chuàng)建一個線程，當(dāng)前線程用來處理對應(yīng)客戶端的請求，所以，可以支持同時n個客戶端鏈接（長連接）。
ForkingTCPServer
ForkingTCPServer和ThreadingTCPServer的使用和執(zhí)行流程基本一致，只不過在內(nèi)部分別為請求者建立 “線程” ?和 “進程”。
基本使用：
#!/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轉(zhuǎn)人工服務(wù).') 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() 服務(wù)端 #!/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?兩個東西，其實本質(zhì)上就是在服務(wù)器端為每一個客戶端創(chuàng)建一個進程，當(dāng)前新創(chuàng)建的進程用來處理對應(yīng)客戶端的請求，所以，可以支持同時n個客戶端鏈接（長連接）。
源碼剖析參考 ThreadingTCPServer
Twisted
Twisted是一個事件驅(qū)動的網(wǎng)絡(luò)框架，其中包含了諸多功能，例如：網(wǎng)絡(luò)協(xié)議、線程、數(shù)據(jù)庫管理、網(wǎng)絡(luò)操作、電子郵件等。
事件驅(qū)動
簡而言之，事件驅(qū)動分為二個部分：第一，注冊事件；第二，觸發(fā)事件。
自定義事件驅(qū)動框架，命名為：“弒君者”：
#!/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): """ 用戶必須繼承該類，從而規(guī)范所有類的方法（類似于接口的功能） """ def execute(self): raise Exception('you must overwrite execute') 最牛逼的事件驅(qū)動框架程序員使用“弒君者框架”：
#!/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如上述代碼，事件驅(qū)動只不過是框架規(guī)定了執(zhí)行順序，程序員在使用框架時，可以向原執(zhí)行順序中注冊“事件”，從而在框架執(zhí)行時可以出發(fā)已注冊的“事件”。
基于事件驅(qū)動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() |
程序執(zhí)行流程：
- 運行服務(wù)端程序
- 創(chuàng)建Protocol的派生類Echo
- 創(chuàng)建ServerFactory對象，并將Echo類封裝到其protocol字段中
- 執(zhí)行reactor的 listenTCP 方法，內(nèi)部使用?tcp.Port 創(chuàng)建socket server對象，并將該對象添加到了 reactor的set類型的字段 _read 中
- 執(zhí)行reactor的 run 方法，內(nèi)部執(zhí)行 while 循環(huán)，并通過 select 來監(jiān)視?_read 中文件描述符是否有變化，循環(huán)中...
- 客戶端請求到達(dá)
- 執(zhí)行reactor的?_doReadOrWrite 方法，其內(nèi)部通過反射調(diào)用 tcp.Port 類的?doRead 方法，內(nèi)部 accept 客戶端連接并創(chuàng)建Server對象實例（用于封裝客戶端socket信息）和?創(chuàng)建 Echo 對象實例（用于處理請求）?，然后調(diào)用 Echo 對象實例的?makeConnection 方法，創(chuàng)建連接。
- 執(zhí)行 tcp.Server 類的?doRead 方法，讀取數(shù)據(jù)，
- 執(zhí)行 tcp.Server 類的?_dataReceived 方法，如果讀取數(shù)據(jù)內(nèi)容為空（關(guān)閉鏈接），否則，出發(fā) Echo 的?dataReceived 方法
- 執(zhí)行?Echo 的?dataReceived 方法
從源碼可以看出，上述實例本質(zhì)上使用了事件驅(qū)動的方法 和 IO多路復(fù)用的機制來進行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/
轉(zhuǎn)載于:https://www.cnblogs.com/cp-miao/p/5658507.html
總結(jié)
以上是生活随笔為你收集整理的Python开发【第八篇】：网络编程 Socket的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Windows Server 2008
- 下一篇： Runtime 总结