[ python ] 網絡編程（2）

黏包問題

 這樣一個實例

import socket
import subprocess

sk_server = socket.socket() # 建立 socket對象
sk_server.bind(('localhost', 8080)) # 創建socket
sk_server.listen(5) # 開啓監聽
conn, addr = sk_server.accept() # 接收客戶端信息
while True:
    command = conn.recv(1024).decode()
    cmd_res = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) # 執行命令
    stdout = cmd_res.stdout.read()
    stderr = cmd_res.stderr.read()
    result = stdout if stdout else stderr
    print('result:', result)
    conn.sendall(result)    # 發送命令結果

server.py

import socket

sk_client = socket.socket()
sk_client.connect(('localhost', 8080))
while True:
    cmd = input('>>>').strip()
    if not cmd: continue
    sk_client.sendall(cmd.encode())
    result = sk_client.recv(1024).decode('gbk')
    print(result)

client.py

 

 

運行起來，咱們在客戶端輸入 tasklist (windows查看全部進程)，而後在輸入 dir(查看當前目錄信息)

 執行完 tasklist 後，再次執行 dir 時，發現輸出結果是 tasklist 未顯示出來的部分。這種狀況，就稱之爲 黏包。

【注意：只有TCP有粘包現象，UDP永遠不會粘包】

 

黏包成因

tcp協議的拆包機制：

當發送端緩衝區的長度大於網卡的MTU時，tcp會將此次發送的數據拆成幾個數據包發送出去。
MTU是Maximum Transmission Unit的縮寫。意思是網絡上傳送的最大數據包。MTU的單位是字節。 大部分網絡設備的MTU都是1500。若是本機的MTU比網關的MTU大，大的數據包就會被拆開來傳送，這樣會產生不少數據包碎片，增長丟包率，下降網絡速度。

面向流的通訊特色和Nagle算法
TCP（transport control protocol，傳輸控制協議）是面向鏈接的，面向流的，提供高可靠性服務。
收發兩端（客戶端和服務器端）都要有一一成對的socket，所以，發送端爲了將多個發往接收端的包，更有效的發到對方，使用了優化方法（Nagle算法），將屢次間隔較小且數據量小的數據，合併成一個大的數據塊，而後進行封包。
這樣，接收端，就難於分辨出來了，必須提供科學的拆包機制。 即面向流的通訊是無消息保護邊界的。
對於空消息：tcp是基於數據流的，因而收發的消息不能爲空，這就須要在客戶端和服務端都添加空消息的處理機制，防止程序卡住，而udp是基於數據報的，即使是你輸入的是空內容（直接回車），也能夠被髮送，udp協議會幫你封裝上消息頭髮送過去。
可靠黏包的tcp協議：tcp的協議數據不會丟，沒有收完包，下次接收，會繼續上次繼續接收，己端老是在收到ack時纔會清除緩衝區內容。數據是可靠的，可是會粘包。

udp和tcp一次發送數據長度的限制

    用UDP協議發送時，用sendto函數最大能發送數據的長度爲：65535- IP頭(20) – UDP頭(8)＝65507字節。用sendto函數發送數據時，若是發送數據長度大於該值，則函數會返回錯誤。（丟棄這個包，不進行發送） 

    用TCP協議發送時，因爲TCP是數據流協議，所以不存在包大小的限制（暫不考慮緩衝區的大小），這是指在用send函數時，數據長度參數不受限制。而實際上，所指定的這段數據並不必定會一次性發送出去，若是這段數據比較長，會被分段發送，若是比較短，可能會等待和下一次數據一塊兒發送。

補充說明

 

 

會發生黏包的兩種狀況：

1. 發送方的緩存機制：發送端須要等待緩衝區滿才發送出去，形成黏包
2. 接收方的緩存機制：接收方不能及時接收緩衝區的包，形成多個包接收

 

總結：

黏包現象只發生在tcp協議中：
1. 從表面上看，黏包問題主要是由於發送方和接收方的緩存機制、tcp協議面向流通訊特色；
2. 實際上，主要仍是由於接收方不知道消息之間的界限，不知道一次性提取多少字節的數據所形成的。

 

黏包的解決方案：

黏包的根源在於，接收端不知道發送端將要傳送的字節流的長度，因此解決黏包的方法就是圍繞，如何讓發送端在發送數據前，把本身將要發送的字節流總大小讓接收端知道，而後接收端來一個死循環收完全部數據。

 

 解決方案一：

import socket
import subprocess

sk_server = socket.socket()
sk_server.bind(('localhost', 8080))
sk_server.listen(5)
conn, addr = sk_server.accept()

while True:
    command = conn.recv(1024).decode()
    cmd_res = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    stdout = cmd_res.stdout.read()
    stderr = cmd_res.stderr.read()
    result = stdout if stdout else stderr
    res_size = len(result)
    conn.sendall(str(res_size).encode())
    response = conn.recv(1024)
    conn.sendall(result)

server.py

import socket

sk_client = socket.socket()
sk_client.connect(('localhost', 8080))

while True:
    command = input('>>>').strip()
    if not command: continue
    sk_client.sendall(command.encode())
    res_size = sk_client.recv(1024).decode()
    sk_client.sendall(b'000')
    revice_size = 0
    while revice_size != int(res_size):
        data = sk_client.recv(1024)
        revice_size += len(data)
        print(data.decode('gbk'))

client.py

 

 

使用這種方式存在的問題：
    程序的運行速度遠快於網絡傳輸速度，因此在發送一段字節前，先用 send 去發送該字節流長度，這種方式會放大網絡延遲帶來的性能損耗

 

解決方案二：

使用 struct模塊，這個模塊能夠把要發送的數據長度轉換成固定長度的字節。這樣客戶端每次接收消息以前只要先接受這個固定長度字節的內容看一看接下來要接收的信息大小，那麼最終接收的數據只要達到這個值就中止，就能恰好很少很多的接收完整的數據了。

 

struct 模塊

該模塊能夠把一個類型，如數字，轉成固定長度的bytes

In [1]: import struct

In [2]: s = struct.pack('i', 111111)    # 使用 pack 方法將int類型轉換爲固定的 4 個字節

In [3]: s
Out[3]: b'\x07\xb2\x01\x00'

In [4]: len(s)    # 固定的 4 個字節
Out[4]: 4

In [5]: struct.unpack('i', s)    # 使用 unpack 方法將 4 個字節還原爲字符，類型爲元組
Out[5]: (111111,)

 

 

使用 struct 解決黏包

藉助 struct 模塊，咱們知道長度數字能夠被轉換成一個標準大小的 4 個字節數字。所以能夠利用這個特色來預先發送數據長度。

 

 

import socket, struct
import subprocess

sk_server = socket.socket()
sk_server.bind(('localhost', 8080))
sk_server.listen(5)
conn, addr = sk_server.accept()
while True:
    command = conn.recv(1024).decode()
    res_cmd = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    stdout = res_cmd.stdout.read()
    stderr = res_cmd.stderr.read()
    result = stdout if stdout else stderr
    res_size = len(result)
    conn.sendall(struct.pack('i', res_size))
    conn.sendall(result)

server.py

import socket
import struct

sk_client = socket.socket()
sk_client.connect(('localhost', 8080))

while True:
    command = input('>>>').strip()
    if not command: continue
    sk_client.sendall(command.encode())
    res = sk_client.recv(4)
    res_size = struct.unpack('i', res)[0]
    print(res_size)
    revice_size = 0
    while revice_size != res_size:
        data = sk_client.recv(1024)
        revice_size += len(data)
        print(data.decode('gbk'))

client.py

 

 

這裏還能夠將報頭作成字典字典裏包含將要發送的真實數據的詳細信息，而後json序列化，而後用 struct 將序列化後的數據長度打包成4個字節.

 

 

這種方式用於須要初始化較多的信息

import socket, struct, json
import subprocess

sk_server = socket.socket()
sk_server.bind(('localhost', 8080))
sk_server.listen(5)

conn, addr = sk_server.accept()
while True:
    command = conn.recv(1024).decode()
    cmd_res = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    stdout = cmd_res.stdout.read()
    stderr = cmd_res.stderr.read()
    result = stdout if stdout else stderr
    headers = {'res_size': len(result)} # 將head信息組合成 字典類型
    head_json = json.dumps(headers) # 轉換爲 json 類型
    head_json_bytes = bytes(head_json, encoding='utf-8')
    conn.send(struct.pack('i', len(head_json_bytes)))   # 首先發送 head 信息的大小
    conn.send(head_json_bytes)  # 再次發送 head 信息
    conn.send(result)   # 最後 發送 執行命令的結果集合

server.py

import socket, struct, json

sk_client = socket.socket()
sk_client.connect(('localhost', 8080))
while True:
    cmd = input('>>>').strip()
    if not cmd: continue
    sk_client.send(cmd.encode())
    res_size = struct.unpack('i', sk_client.recv(1024))[0]  # 首先獲取 head 大小
    head_json = sk_client.recv(res_size).decode()   # 經過 head 大小獲取 head 信息
    head_dict = json.loads(head_json)   # 轉爲 字典 類型
    data_len = head_dict['res_size']    # 取出 結果集 大小
    revice_size = 0
    while revice_size != data_len:  # 循環接收 結果集
        data = sk_client.recv(1024)
        revice_size += len(data)
        print(data.decode('gbk'))

client.py

 

 

FTP小做業：上傳下載文件

import socket
import json
import struct
import os


class MyServer:

    request_queue_size = 5

    def __init__(self, ip_port, bind_activate=True):
        self.socket = socket.socket()
        self.ip_port = ip_port
        if bind_activate:
            try:
                self.activate()
            except:
                self.server_close()
                raise

    def activate(self):
        self.socket.bind(self.ip_port)
        self.socket.listen(self.request_queue_size)

    def get_resquest(self):
        return self.socket.accept()

    def server_close(self):
        self.socket.close()

    def run(self):
        while True:
            self.conn, self.client_addr = self.get_resquest()
            print('from client:', self.client_addr)
            while True:
                try:
                    head_struct = self.conn.recv(4)
                    head_len = struct.unpack('i', head_struct)[0]
                    head_json = self.conn.recv(head_len).decode()
                    head_dict = json.loads(head_json)
                    command = head_dict['command']
                    if hasattr(self, command):
                        func = getattr(self, command)
                        func(head_dict)
                except Exception:
                    break

    def put(self, args):
        filename = args['filename']
        file_size = args['file_size']
        file_path = os.path.join('file_upload', filename)
        recv_size = 0
        with open(file_path, 'wb') as f:
            while recv_size != file_size:
                recv_data = self.conn.recv(1024)
                recv_size += len(recv_data)
                f.write(recv_data)

if __name__ == '__main__':
    ftp_server = MyServer(('localhost', 8080))
    ftp_server.run()

server.py

import socket
import json, struct
import os, sys


class MyClient:
    def __init__(self, ip_port, connect=True):
        self.socket = socket.socket()
        self.ip_port = ip_port
        if connect:
            try:
                self.connect()
            except:
                self.client_close()
                raise

    def connect(self):
        self.socket.connect(self.ip_port)

    def client_close(self):
        self.socket.close()

    def run(self):
        while True:
            cmd = input('>>>').strip()
            if not cmd: continue
            cmd_str = cmd.split()[0]
            if hasattr(self, cmd_str):
                func = getattr(self, cmd_str)
                func(cmd)

    def put(self, command):
        if len(command) > 1:
            filename = command.split()[1]
            if os.path.isfile(filename):
                cmd_str = command.split()[0]
                file_size = os.path.getsize(filename)
                head_dict = {'command': cmd_str, 'filename': filename, 'file_size': file_size}
                head_json = json.dumps(head_dict)
                head_json_bytes = bytes(head_json, encoding='utf-8')
                head_json_strcut = struct.pack('i', len(head_json_bytes))
                print(head_json_strcut)
                self.socket.send(head_json_strcut)
                self.socket.send(head_json_bytes)
                with open(filename, 'rb') as f:
                    while True:
                        data = f.read(1024)
                        send_size = f.tell()
                        if not data:
                            print('upload successful.')
                            break
                        self.socket.send(data)
                        self.__progress(send_size, file_size, '上傳中')
            else:
                print('\033[31;1m文件不存在.\033[0m')


        else:
            print('\033[31;1m命令格式錯誤.\033[0m')


    def __progress(self, trans_size, file_size, mode):
        bar_length = 100
        percent = float(trans_size) / float(file_size)
        hashes = '=' * int(percent * bar_length)
        spaces = ' ' * int(bar_length - len(hashes))
        sys.stdout.write('\r%s %.2fM/%.2fM %d%% [%s]'
                         %(mode, trans_size/1048576, file_size/1048576, percent*100, hashes+spaces))


if __name__ == '__main__':
    ftp_client = MyClient(('localhost', 8080))
    ftp_client.run()

client.py

 

 

 

socketserver 模塊

主要類型

    該模塊有4個比較主要的類，其中經常使用的是 TCPServer 和 UDPServer

 

1. TCPServer
2. UDPServer
3. UnixStreamServer: 相似於TCPServer提供面向數據流的套接字鏈接，可是旨在UNIX平臺上可用；
4. UnixDatagramServer: 相似於UDPServer提供面向數據報的套接字鏈接，可是旨在UNIX平臺上可用；

 這四個類型同步地處理請求，也就是說一個請求沒有完成以前是不會處理下一個請求的，這種模式固然不適合生產環境，一個客戶端鏈接就可能拖延全部的執行。因此這個模塊還提供了兩種支持異步處理的類：

1.  ForkingMixIn: 爲每個客戶端請求派生一個新的進程專門處理；
2. ThreadingMixIn: 爲每個客戶端請求派生一個新的線程專門處理；

 繼承自這兩個類型的服務端在處理新的客戶端鏈接時不會阻塞，而是建立新的進/線程專門處理客戶端請求。

 

編程框架

首先從高層面介紹一下使用SocketServer模塊開發多進程/線程 異步服務器的流程：

1. 根據須要選擇一個合適的服務類型，如，面向TCP鏈接的多進程服務器： ForkingTCPServer ；
2. 建立一個請求處理器（request handler）類型，這個類型的 handle()（相似於回調函數）方法中定義如何處理到達的客戶端鏈接。
3. 實例化服務器，傳入服務器綁定的地址和第2步定義的請求處理器類；
4. 調用服務器實例的 handle_request() 或 serve_forever() 方法，一次或屢次處理客戶請求。

 

使用 socketserver 實例：

import socketserver

class MyServer(socketserver.BaseRequestHandler):
    def handle(self):
        while True:
            self.data = self.request.recv(1024).decode()
            print('from client:', self.client_address)
            print(self.data)
            self.request.send(self.data.upper().encode())


if __name__ == '__main__':
    HOST, PORT = 'localhost', 8080
    server = socketserver.ThreadingTCPServer((HOST, PORT), MyServer)
    server.serve_forever()

server.py

import socket

class MyClient:
    def __init__(self, ip_port, connect=True):
        self.client = socket.socket()
        self.ip_port = ip_port
        if connect:
            try:
                self.connect()
            except:
                self.client_close()
                raise

    def connect(self):
        self.client.connect(self.ip_port)

    def client_close(self):
        self.client.close()

    def start(self):
        while True:
            cmd = input('>>>').strip()
            if not cmd: continue
            self.client.send(cmd.encode())
            cmd_upper = self.client.recv(1024).decode()
            print(cmd_upper)


if __name__ == '__main__':
    client = MyClient(('localhost', 8080))
    client.start()

client.py