python——socket，IO多路複用（select），socket server實現多併發

時間 2019-11-11

標籤 python socket 多路 select server 實現併發欄目 Python 简体版

原文原文鏈接

1、socketpython

定義：編程

socket本質上就是在2臺網絡互通的電腦之間，架設一個通道，兩臺電腦經過這個通道來實現數據的互相傳遞。咱們知道網絡通訊都是基於 ip+port（套接字）方能定位到目標的具體機器上的具體服務，操做系統有0-65535個端口，每一個端口均可以獨立對外提供服務，若是把一個公司比作一臺電腦，那公司的總機號碼就至關於ip地址，每一個員工的分機號就至關於端口，你想找公司某我的，必須先打電話到總機，而後再轉分機。數組

TCP/IP與socket：

socket是在應用層和傳輸層之間的一個抽象層，它把TCP/IP層複雜的操做抽象爲幾個簡單的接口供應用層調用已實現進程在網絡中通訊。服務器

三次握手

在TCP/IP協議中，TCP協議（傳輸層）經過三次握手創建一個可靠的鏈接網絡

第一次握手：客戶端嘗試鏈接服務器，向服務器發送syn包（同步序列編號Synchronize Sequence Numbers），syn=j，客戶端進入SYN_SEND狀態等待服務器確認數據結構

第二次握手：服務器接收客戶端syn包並確認（ack=j+1），同時向客戶端發送一個SYN包（syn=k），即SYN+ACK包，此時服務器進入SYN_RECV狀態多線程

第三次握手：第三次握手：客戶端收到服務器的SYN+ACK包，向服務器發送確認包ACK(ack=k+1），此包發送完畢，客戶端和服務器進入ESTABLISHED狀態，完成三次握手併發

定睛一看，服務器socket與客戶端socket創建鏈接的部分其實就是大名鼎鼎的三次握手app

創建一個socket必須至少有2端，一個服務端，一個客戶端，服務端被動等待並接收請求，客戶端主動發起請求，鏈接創建以後，雙方能夠互發數據。 socket

服務器端：

 1 #服務器端
 2 import socket
 3 server=socket.socket(地址簇，套接字)    #聲明
 4 
 5 server.bind(("localhost",9999))        #肯定監聽的端口
 6 
 7 server.listen()        #開始監聽
 8 
 9 conn,addr=server.accpet()    #在服務器端生成實例並賦值
10 print("new conn:",addr)
11 
12 while True:        #服務器端循環接收數據
13     
14     data=conn.recv(1024)    #接受數據，最多1K
15     print(data)　　# >>b'abc'
16     
17     conn.send(data.upper)    #將數據改成大寫發回去

客戶端：

 1 #客戶端
 2 import socket
 3 
 4 client=socket.socket()    #實例化
 5 
 6 client.connect("localhost",9999)    #綁定IP和端口
 7 
 8 client.send(b"abc")    #轉爲二進制發送
 9 
10 data=client.recv(1024)
11 
12 print(data)
13 
14 >> b'ABC'

解決粘包問題（數據過大一次傳不完）：

客戶端：

 1 #服務器端
 2 import socket,os
 3 server=socket.socket(地址簇，套接字)    #聲明
 4 
 5 server.bind(("localhost",9999))        #肯定監聽的端口
 6 
 7 server.listen()        #開始監聽
 8 
 9 conn,addr=server.accpet()    #在服務器端生成實例並賦值
10 print("new conn:",addr)
11 
12 while True:        #服務器端循環接收數據
13     
14     data=conn.recv(1024)    #接受數據，最多1K
15     data=data.decode()    #二進制轉成'utf-8'
16     if no data :    
17         print('客戶端已斷開')
18         break
19     print('執行指令：',data)
20     
21     cmd_res=os.popen(data).read()    #執行操做
22     
23     conn.send(len(cmd_res).encode('utf-8'))        #先發送數據的大小
24     
25     time.sleep(0.5)        #先睡0.5s 防止粘包，另外一種方法conn.recv(),客戶端conn.sendall('123')
26     
27     conn.send(cmd_res.encode('utf-8'))    #將數據發回去
28 
29 server.close()    #我好像前面都忘寫關閉了

客戶端：

 1 #客戶端
 2 import socket
 3 
 4 client=socket.socket()    #實例化
 5 
 6 client.connect("localhost",9999)    #綁定IP和端口
 7 
 8 while True:
 9     cmd = input(">>:").strip()
10     
11     client.send(cmd.encode('utf-8'))
12 
13     cmd_res_size=(client.recv(1024)).decode()    #要傳輸的文件的大小
14     received_size=0        #本地已有的文件大小
15     received_data=b''
16     print("命令結果大小",cmd_res_size)
17     while received_size<cmd_res_size:
18         data=client.recv(1024)
19         received_data+=data
20         received_size+=len(data)
21     else:
22         print("數據接收完畢")
23         print(received_data.decode())
24 
25 client.close()

小Tips：

 1 import hashlib
 2 
 3 m=hashlib.md5
 4 
 5 m.updata(b'abc')    
 6 m.updata(b'123')
 7 
 8 #m.updata(b'abc123')    #二者結果一毛同樣，so能夠採用這種方式讀取文件計算md5值
 9 
10 print(m.hexdigest)    #16進制

參數們：

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)

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

　　套接字的文件描述符

2、IO多路複用

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

Linux

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

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多路複用。

對於select方法：

句柄列表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 秒，以後返回三個空列表，若是監聽的句柄有變化，則直接執行。

利用select監聽終端操做

import select
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實現僞同時處理多個Socket客戶端請求

服務器端：

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)  # flag爲0則爲非堵塞模式不然, 套接字將設置爲阻塞模式（默認值）。
# 在非阻塞模式下, 若是recv()調用沒有發現任何數據或者send()調用沒法當即發送數據, 那麼將引起socket.error異常。在阻塞模式下, 這些調用在處理以前都將被阻塞。

inputs = [sk1,]     # 本身也要監測呀,由於sk1自己也是個socket客戶端

while True:
    readable_list, writeable_list, error_list = select.select(inputs, [], inputs, 1)    # 交給select去監控

    for r in readable_list:
        if sk1 == r:    # 當客戶端第一次鏈接服務端時
            print('鏈接上服務器')
            request, address = r.accept()
            request.setblocking(0)      # 設置爲非堵塞
            inputs.append(request)      # 將新鏈接加入input中監聽

        else:   # 當客戶端鏈接上服務端以後，再次發送數據時
            received = r.recv(1024)     # 當正常接收客戶端發送的數據時，沒有數據時會堵塞在這
            if received:
                print('received data:', received)
            else:   # 當客戶端關閉程序時
                inputs.remove(r)

sk1.close()

小Tips：

1）堵塞與非堵塞

一輛從 A 開往 B 的公共汽車上，路上有不少點可能會有人下車。司機不知道哪些點會有哪些人會下車，對於須要下車的人，如何處理更好？ 

1. 司機過程當中定時詢問每一個乘客是否到達目的地，如有人說到了，那麼司機停車，乘客下車。 ( 相似阻塞式 ) 
2. 每一個人告訴售票員本身的目的地，而後睡覺，司機只和售票員交互，到了某個點由售票員通知乘客下車。 ( 相似非阻塞 ) 
很顯然，每一個人要到達某個目的地能夠認爲是一個線程，司機能夠認爲是 CPU 。在阻塞式裏面，每一個線程須要不斷的輪詢，上下文切換，以達到找到目的地的結果。而在非阻塞方式裏，每一個乘客 ( 線程 ) 都在睡覺 ( 休眠 ) ，只在真正外部環境準備好了才喚醒，這樣的喚醒確定不會阻塞。 

非阻塞的原理：

把整個過程切換成小的任務，經過任務間協做完成。 
由一個專門的線程來處理全部的 IO 事件，並負責分發。 

事件驅動模型機制：

事件到的時候觸發，而不是同步的去監視事件。

2）事件驅動模型

目前大部分的UI編程都是事件驅動模型，如不少UI平臺都會提供onClick()事件，這個事件就表明鼠標按下事件。事件驅動模型大致思路以下：
1. 有一個事件（消息）隊列；
2. 鼠標按下時，往這個隊列中增長一個點擊事件（消息）；
3. 有個循環，不斷從隊列取出事件，根據不一樣的事件，調用不一樣的函數，如onClick()、onKeyDown()等；
4. 事件（消息）通常都各自保存各自的處理函數指針，這樣，每一個消息都有獨立的處理函數；

客戶端：

import socket

ip_port = ('127.0.0.1',8002)
sk = socket.socket()
sk.connect(ip_port)

while True:
    inp = input('please input:')
    sk.sendall(inp.encode('utf-8'))
sk.close()

3、socket server模塊

SocketServer內部使用 IO多路複用以及「多線程」和「多進程」，從而實現併發處理多個客戶端請求的Socket服務端。即：每一個客戶端請求鏈接到服務器時，Socket服務端都會在服務器是建立一個「線程」或者「進程」專門負責處理當前客戶端的全部請求。

實現多併發知足條件：

1.你必須本身建立一個請求處理類，而且這個類要繼承BaseRequestHandler,而且還有重寫父親類裏的handle()
2.你必須實例化TCPServer ，而且傳遞server ip 和你上面建立的請求處理類給這個TCPServer
3.server.handle_request() 只處理一個請求
4.server.serve_forever() 處理多個一個請求，永遠執行

import socketserver

class MyTCPHandler(socketserver.BaseRequestHandler):    #條件1，本身建立一個類
    def handle(self):    #重寫父類中的handle()
        while True:
            try:
                self.data = self.request.recv(1024).strip()
                print("{} wrote:".format(self.client_address[0]))
                print(self.data)
                self.request.send(self.data.upper())
            except ConnectionResetError as e:
                print("err",e)
                break
if __name__ == "__main__":
    HOST, PORT = "localhost", 9999
    # Create the server, binding to localhost on port 9999
    server = socketserver.ThreadingTCPServer((HOST, PORT), MyTCPHandler)    #實例化ThreadingTCPServer，並傳遞(端口號+IP)，和請求類名
    server.serve_forever()    #關閉