45網絡編程_UDP

?

?

?

UDP編程：

?

>netstat -anp tcp | find "9998"?? #win下

#echo "123abc" | nc -u 127.0.0.1 9998

?

應用：

無連接協議，基于如下假設：網絡足夠好、消息不會丟包、包不會亂序；

1、音頻、視頻傳輸，一般丟些包，問題不大，最多丟些圖像、聽不清說話，再次說話即可；

2、海量采集數據，如傳感器發來的數據，丟幾十、幾百條數據也沒關系；

3、DNS協議，數據內容小，一個包就能查詢到結果，不存在亂序、丟包，重新請求解析；

?

注：

即使在LAN，也不能保證不丟包，且包的到達不一定有序；

一般來說，UDP性能優于TCP，但可靠性要求高的場合還是選擇用TCP；

QUIC，quick udp internet connection，google，是谷歌制定的一種基于UDP的低時延的互聯網傳輸層協議。在2016年11月國際互聯網工程任務組(IETF)召開了第一次QUIC工作組會議，受到了業界的廣泛關注。這也意味著QUIC開始了它的標準化過程，成為新一代傳輸層協議；

?

?

UDP服務端編程步驟：

創建socket對象，sock=socket.socket(type=socket.SOCK_DGRAM);

綁定ip和port，bind()方法；

傳輸數據：

???????? recvfrom(bufsize[,flags])，接收數據，獲取一個二元組(string,address)；

???????? sendto(string,address)，發送數據，發送某信息給某地址；

釋放資源；

?

例：

import socket

?

sock = socket.socket(type=socket.SOCK_DGRAM)

?

addr = ('127.0.0.1', 9998)

sock.bind(addr)

?

data, clientaddr = sock.recvfrom(1024)

print(clientaddr)

?

msg = 'ack: {}'.format(data.decode())

sock.sendto(msg.encode(), clientaddr)

輸出：

('127.0.0.1', 9999)

?

例，ChatServerUdp：

ver1：

class ChatServerUdp:

??? def __init__(self, ip='127.0.0.1', port=9998):

??????? self.sock = socket.socket(type=socket.SOCK_DGRAM)

??????? self.addr = (ip, port)

??????? self.event = threading.Event()

??????? self.clients = set()?? #集合，去重，client主動退出后要清此數據結構

?

??? def start(self):

??????? self.sock.bind(self.addr)

??????? threading.Thread(target=self._recv, name='recv').start()

?

??? def stop(self):

??????? for c in self.clients:?? #業務中udp的server關閉時不會通知client

??????????? self.sock.sendto(b'end', c)

??????? self.sock.close()?? #udp的socket關閉很快，不會有很多垃圾

??????? self.event.set()

?

??? def _recv(self):?? #_recv中使用多線程場景，在一對多情況下，server發送消息和接收消息出現不匹配時，用另一線程單獨處理發送數據，否則接收和發送是同步，只有等發送完才能繼續再次接收

??????? while not self.event.is_set():

??????????? data, client = self.sock.recvfrom(1024)

??????????? data = data.strip().decode()

?

??????????? if data == 'quit':

??????????????? self.clients.remove(client)

??????????????? continue?? #關鍵，接收下個client的消息

??????????? self.clients.add(client)

??????????? print(self.clients)

?

??????????? msg = 'ack: {}'.format(data)

??????????? for c in self.clients:

??????????????? self.sock.sendto(msg.encode(), c)

?

if __name__ == '__main__':

??? cs = ChatServerUdp()

??? cs.start()

??? myutils.show_threads()

?

例：

ver2：

增加ack和heartbeat機制；

心跳即一端定時發往另一端信息，一般每次發的數據越少越好，心跳時間間隔約定好就行，ack響應，一端收到另一端的消息后返回的信息；

心跳包設計：

c主動，一般由client發hb-->server，server并不需要發ack-->client，只需要記錄client還活著就行；

s主動，server發hb掃一遍client，一般需要client發ack響應來表示活著，server沒收到ack就斷開與client連接，server移除其信息，這種實現較為復雜，用的少；

c-s雙向，用的更少；

class ChatServerUdp:

??? def __init__(self, ip='127.0.0.1', port=9998, interval=10):

??????? self.sock = socket.socket(type=socket.SOCK_DGRAM)

??????? self.addr = (ip, port)

??????? self.event = threading.Event()

??????? self.clients = {}

??????? self.interval = interval

?

??? def start(self):

??????? self.sock.bind(self.addr)

??????? threading.Thread(target=self._recv, name='recv').start()

?

??? def stop(self):

??????? for c in self.clients:

??????????? self.sock.sendto(b'end', c)

??????? self.sock.close()

??????? self.event.set()

?

??? def _recv(self):

??????? while not self.event.is_set():

??????????? lostset = set()

??????????? data, client = self.sock.recvfrom(1024)

??????????? data = data.strip().decode()

?

??????????? current = datetime.datetime.now().timestamp()

??????????? if data == '^hb^' or data == 'reg':

??????????????? print('hb')

??????????????? self.clients[client] = current

??????????????? continue

??????????? elif data == 'quit':

??????????????? self.clients.pop(client, None)

??????????????? logging.info('{} leaving'.format(client))

?????? ?????????continue

?

??????????? self.clients[client] = current

??????????? print(self.clients)

?

??????????? msg = 'ack: {} {}\n{}\n'.format(*client, data)

??????????? logging.info(msg)

??????????? for c, stamp in self.clients.items():

??????????????? if current - stamp > self.interval:

??????????????????? lostset.add(c)

??????????????? else:

??????????????????? self.sock.sendto(msg.encode(), c)

?

??????????? for c in lostset:

??????????????? self.clients.pop(c)

?

if __name__ == '__main__':

??? cs = ChatServerUdp()

??? cs.start()

??? myutils.show_threads()

?

UDP客戶端編程步驟：

創建socket對象，socket.socket(type=socket.SOCK_DGRAM)；

發送數據：sendto(string,address)，發送某信息給某地址；

釋放資源；

?

udp客戶端編程中，只能在sendto()后，才能recvfrom()，否則OSError；

?

例：

sock = socket.socket(type=socket.SOCK_DGRAM)

?

addr = ('127.0.0.1', 9998)

data = 'test_data'.encode()

?

sock.sendto(data, addr)?? #方式1，使用sendto()和recvfrom()，建議用此種方式

data, saddr = sock.recvfrom(1024)?? #也可用recv()，只不過不知道誰發的消息了

print(data, saddr)

sock.close()

?

# sock.connect(addr)?? #方式2，用connect()連接后才能用send()或sendto()，沒有connect()連接只能用sendto()；此方式可能會有問題，client-->server正常，server-->client，server上連client的端口不對了

# sock.send(data)

# data, saddr = sock.recvfrom(1024)

# print(data, saddr)

# sock.close()

?

例：

addr = ('127.0.0.1', 9998)

event = threading.Event()

?

def recv1(sock:socket.socket, event:threading.Event):

??? while not event.is_set():

??????? data, saddr = sock.recvfrom(1024)

??????? logging.info('recv: {} ; from: {}'.format(data, saddr))

?

sock1 = socket.socket(type=socket.SOCK_DGRAM)

sock1.sendto('udp client1 send'.encode(), addr)

threading.Thread(target=recv1, args=(sock1, event)).start()?? #recvfrom()必須在sendto()或connect()之后，否則OSError，即recvfrom()操作之前應該先sendto()或connect()；如果用connect()，則遠端必須有服務

?

def recv2(sock:socket.socket, event:threading.Event):

??? while not event.is_set():

??????? data, saddr2 = sock.recvfrom(1024)

??????? logging.info('recv: {} ; from: {}'.format(data, saddr2))

?

sock2 = socket.socket(type=socket.SOCK_DGRAM)

sock2.connect(addr)

threading.Thread(target=recv2, args=(sock2, event)).start()

threading.Event().wait(5)

sock2.sendto('udp client2 send'.encode(), addr)

event.wait(2)

sock2.send('udp client2.1 send'.encode())

?

while True:

??? if input('>>> ').strip() == 'quit':

??????? sock1.close()

??????? sock2.close()

??????? event.wait(3)

??????? break

logging.info('end')

?

例，ChatClientUdp：

注：此代碼有問題，在發送hb后一直阻塞

class ChatClientUdp:

??? def __init__(self, ip='127.0.0.1', port=9998, interval=5):

??????? self.sock = socket.socket(type=socket.SOCK_DGRAM)

??????? self.addr = (ip, port)

??????? self.event = threading.Event()

??????? self.interval = interval

??????? self.sock.connect(self.addr)

??????? self._sendhb()

?

??? def start(self):

??????? # self.sock.send(b'reg')

??????? threading.Thread(target=self._sendhb, name='hb', daemon=True).start()

??????? # threading.Thread(target=self._recv, name='recv').start()

??????? # self._recv()

?

??? def stop(self):

??????? self.send()

??????? self.sock.close()

??????? self.event.wait(2)

??????? self.event.set()

?

??? def _sendhb(self):

??????? while not self.event.wait(5):

??????????? self.sock.sendto(b'^hb^', self.addr)

?

??? def send(self, msg:str='quit'):

??????? self.sock.sendto(msg.encode(), self.addr)

?

??? def _recv(self):

??????? while not self.event.is_set():

??????????? data, addr = self.sock.recvfrom(1024)

??????????? logging.info('recv {} from {}'.format(data, addr))

?

cc = ChatClientUdp()

cc.start()

while True:

??? data = input('plz input string>>> ')

??? if data == 'quit':

??????? cc.stop()

??????? break

??? else:

??????? cc.send(data)

logging.info('end')

?