Python中socket程式設計
1、Socket介紹:
Python中提供socket.py標準庫,非常底層的介面庫。
Socket是一種通用的網路程式設計介面,和網路層次沒有一一對應關係。
跨作業系統的。緊密結合tcp和udp來使用的。
介面簡單,但是背後的原理不簡單,ip加tcp,通過插兩端。通過socket通道;連線程式。
建立關聯。
apc庫。
加埠是因為應用程式太多了。繫結ip地址,作為假埠。
埠是由誰管理的
一般都是tcp和udp程式設計。Socket基本的,chatserver。
協議族:AF 表示address family ,用於socket()的第一個引數。
名稱 |
含義 |
AF_INET |
Ipv4 |
AF_INET6 |
Ipv6 |
AF_UNIX |
Unix domain socket,windows沒有 |
第三個本機使用效率不錯的,通用的話就是應該進一步考慮了。
Socket型別:
名稱 |
含義 |
Sock_STREAM |
面向連線的流套接字,預設值,tcp協議 |
SOCK_DGRAM |
無連線的資料報文套接字,UDP協議 |
2、tcp程式設計
Tcp程式設計是IO密集型。多執行緒處理的問題。
Server端:1、要有socket介面。2、找ip地址和埠繫結。3、監聽埠。4、accept,接受socket,建立小的socket端。直接和應用程式連線在一起的。5、讀取使用者資料。6、寫,傳送資料。7、資料完成後斷開。
Client端:1、要有socket端,主動連線別人。2、connect建立連線,有socket,和埠和ip。
3、寫,傳送資料。4、讀取伺服器端資料。5、資料完成後關閉了。
伺服器端沒有響應了,tcp協議管理。
Socket會佔用描述符,每一個都會建立一個檔案描述符。客戶端看到只有的是一個。
Import socket
Server = socket.socket() socket介面。
Server.bind(ipaddr)
Server.bind((‘0.0.0.0’,9999))繫結
Server.listen()監聽
S2,iP2 = server.accept()
S1.recv(1024)緩衝區大小。
S1.send(b’ack’)
decode()解碼
encode()編碼
建立socket物件,
一個ip和一個埠只能被一個程式使用。埠只能進行一次監聽,繫結,再次監聽或者繫結的話就會報錯。
使用完畢後必須進行關閉。
應用:
簡單的實現:解決中文的情況,編解碼的時候全部註明統一編碼和解碼。
import socket
server =socket.socket()
server.bind(('127.0.0.1',99))
server.listen(99)
s1,ip = server.accept()
print(s1)
print(ip)
while True:
data = s1.recv(1024)
print(data)
s1.send('ack{}'.format(data.decode('gbk')).encode('gbk'))
s1.close()
server.close()
<socket.socket fd=192, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 99), raddr=('127.0.0.1', 50149)>
('127.0.0.1', 50149)
b'\xd6\xd0\xb9\xfa'
客服端和伺服器端建立間接,需要建立一條socket通道,每次建立連線,listen的埠都會和客戶端建立的新的埠,因為連線埠就會阻塞,所以需要建立新的埠。隱士的還是看到連線的埠還是原來的socket。
應用:寫一個群聊程式
1)第一步:import threading
import logging
import socket
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class ChatServer:
def __init__(self,ip='127.0.0.1',port=999):
self.addr = (ip , port)
self.socket = socket.socket()
def start(self):
self.socket.bind(self.addr)
self.socket.listen()
threading.Thread(target=self.accept,name='accept').start()
def accept(self):
while True:
s,ip = self.socket.accept()
logging.info(s)
logging.info(ip)
threading.Thread(target=self.connt,name='connt',args=(s,)).start()
def connt(self,sockets):
while True:
data = sockets.recv(1024)
logging.info(data)
sockets.send('ack-{}'.format(data.decode()).encode())
def stop(self):
self.socket.close()
cs = ChatServer()
cs.start()
2)第二步
把所有的客戶端的ip和埠保留在一個容器裡面,一個客戶端傳送訊息到伺服器端,伺服器端,進行訊息的轉發等。
import threading
import logging
import socket
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class ChatServer:
def __init__(self,ip='127.0.0.1',port=999):
self.addr = (ip , port)
self.socket = socket.socket()
self.cliens = {}
def start(self):
self.socket.bind(self.addr)
self.socket.listen()
threading.Thread(target=self.accept,name='accept').start()
def accept(self):
while True:
s,ip = self.socket.accept()
logging.info(s)
logging.info(ip)
self.cliens[ip] = s
threading.Thread(target=self.connt,name='connt',args=(s,)).start()
def connt(self,sockets):
while True:
data = sockets.recv(1024)
logging.info(data)
sockets.send('ack-{}'.format(data.decode('gbk')).encode('gbk'))
for s in self.cliens.values():
s.send('ack1-{}'.format(data.decode('gbk')).encode('gbk'))
def stop(self):
for s in self.cliens.values():
s.close()
self.socket.close()
cs = ChatServer()
cs.start()
其他方法:
名稱 |
含義 |
Socket.recv(bufsize[,flags]) |
獲取資料,預設阻塞的方式 |
Socket.recvfrom(bufsize[,flags]) |
獲取資料,返回一個二元組(bytes,address) |
Socket.recv_into(buffer[,nbytes[,flags]]) |
獲取nbytes的資料後,儲存到buffer中,如果nbytes沒有指定或0,將buffer大小的資料存入buffer中,返回接受的位元組數 |
Socket.recvfrom_into(buffer[,nbytes[,flags]]) |
獲取資料,返回一個二元組(bytes,address)到buffer中 |
Socket.send(bytes[,flags]) |
TCP 傳送資料 |
Socket.sendall(bytes[,flags]) |
TCP傳送全部資料,成功返回None |
Socket.sendto(string[,flag],address) |
UDP傳送資料 |
Socket.sendfile(file,offset=0,count=None) |
傳送一個檔案直到EOF,使用高效能的os.sendfile機制,返回傳送的位元組數,如果win下不支援sendfile,或者不是普通檔案,使用send()傳送檔案,offset告訴其實位置,3.5版本開始。 |
Makefile
Socket.makefile(mode=’r’,buffering=None,*,encoding=None,errors=None,newline=None)建立一個與該套接字相關連的檔案物件,將recv方法看做讀方法,將send方法看做是寫方法。
異常不捕獲會導致當前執行緒異常退出,不捕獲直接到最外層,也就是主執行緒。
3、客戶端tcp程式設計
import socket
raddr = ('127.0.0.1',999)
client = socket.socket()
client.connect(raddr)
while True:
data = client.recv(1024)
print(data)
if data.strip() == b'quit':
break
client.send(b'ack')
client.close()
import threading
import socket
import logging
import datetime
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class ChatClient:
def __init__(self,ip='127.0.0.1',port=8080):
self.clients = socket.socket()
self.raddr = (ip,port)
self.event = threading.Event()
def start(self):
self.clients.connect(self.raddr)
self.send('I am ok')
threading.Thread(target=self.recive,name='receive').start()
def recive(self):
while not self.event.is_set():
data = self.clients.recv(1024)
# logging.info(data)
if data.strip() == b'quit':
break
message = '{:%Y/%m/%d %H:%M:%S}{}:{}\n{}\n'.format(datetime.datetime.now(),*self.raddr,data.strip())
logging.info(message)
def send(self,message:str):
data = '{}\n'.format(message.strip()).encode()
self.clients.send(data)
def stop(self):
self.event.set()
self.clients.close()
def main():
cc = ChatClient()
cc.start()
while True:
cmd = input('>>>>')
if cmd.strip() == 'quit':
cc.stop()
break
cc.send(cmd)
logging.info(threading.enumerate())
if __name__ == '__main__':
main()
4、udp程式設計
同一個協議下繫結同一個埠,才會有埠衝突。Udp不會真的連線。
Import socket
Server=socket.socket(type=)
Server.bind(laddr)繫結本地自己用的。
Server.recv(1024)
Data ,raddr = Server.recvfrom(1024)
Server.sendto(b’back’,raddr);後面的ip可以是不存在的,都會發送出去的。
Server.connect(raddr)後面才可以使用send。一般都是客戶端向服務端連線用的。
伺服器端程式碼:
import socket
import threading
import logging
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class ChatServer:
def __init__(self,ip='127.0.0.1',port=9999):
self.addr = (ip,port)
self.sockets = socket.socket(type=socket.SOCK_DGRAM)
self.event = threading.Event()
def start(self):
self.sockets.bind(self.addr)
threading.Thread(target=self.recv,name='recv').start()
def recv(self):
while not self.event.is_set():
data,laddr = self.sockets.recvfrom(1024)
logging.info(data)
logging.info(laddr)
msg = 'ack.{}from{}{}'.format(data.decode(),*laddr)
masg1 = msg.encode()
logging.info(msg)
self.sockets.sendto(masg1,laddr)
def stop(self):
self.sockets.close()
self.event.set()
def main():
cs = ChatServer()
cs.start()
while True:
cmd = input('>>>>')
if cmd.strip() == 'quit':
cs.stop()
break
logging.info(threading.enumerate())
if __name__ == '__main__':
main()
客戶端程式碼:
import threading
import socket
import logging
import datetime
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class ChatUdpClient:
def __init__(self,ip='127.0.0.1',port=8080):
self.addr = (ip,port)
self.cucsocket = socket.socket(type=socket.SOCK_DGRAM)
self.event = threading.Event()
def start(self):
self.cucsocket.connect(self.addr)
threading.Thread(target=self.recive,name='recive').start()
def recive(self):
while not self.event.is_set():
data,raddr =self.cucsocket.recvfrom(1024)
logging.info(data)
logging.info(raddr)
message = '{}from{}{}'.format(data.decode(),*raddr)
def send(self,message:str):
self.cucsocket.sendto(message.encode(), self.addr)
def stop(self):
self.cucsocket.close()
self.event.set()
def main():
cuc = ChatUdpClient()
cuc.start()
while True:
cmd = input('>>>')
if cmd.strip() == 'quit':
cuc.stop()
break
cuc.send(cmd)
logging.info(threading.enumerate())
if __name__ == '__main__':
main()
ack機制和心跳heartbeat。
心跳機制:
1)一般來說客戶端定時發往伺服器端,伺服器端並不需要ack回覆客戶端,只是需要記錄客戶端活著就可以了。(嚴格考慮時間的問題)
2)伺服器端定時發往客戶端,一般需要客戶端ack響應來表示活著,如果沒有收到ack的客戶端,服務端移除其資訊,這種實現複雜,用的較少。
3)也可以是雙向都發心跳包的,用的情況下較少。
為True的時候就不進入迴圈了。
心跳包客戶端程式碼:
import threading
import socket
import logging
import datetime
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class ChatUdpClient:
def __init__(self,ip='127.0.0.1',port=8080):
self.addr = (ip,port)
self.cucsocket = socket.socket(type=socket.SOCK_DGRAM)
self.event = threading.Event()
def start(self):
self.cucsocket.connect(self.addr)
threading.Thread(target=self.sen_hb,name='hb').start()
threading.Thread(target=self.recive,name='recive').start()
def recive(self):
while not self.event.is_set():
data,raddr =self.cucsocket.recvfrom(1024)
logging.info(data)
logging.info(raddr)
message = '{}from{}{}'.format(data.decode(),*raddr)
def send(self,message:str):
self.cucsocket.sendto(message.encode(), self.addr)
def sen_hb(self):
self.send('hb')
def stop(self):
self.cucsocket.close()
self.event.set()
def main():
cuc = ChatUdpClient()
cuc.start()
while True:
cmd = input('>>>')
if cmd.strip() == 'quit':
cuc.stop()
break
cuc.send(cmd)
logging.info(threading.enumerate())
if __name__ == '__main__':
main()
心跳包伺服器端程式碼:
import threading
import socket
import logging
import datetime
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class ChatUdpServer:
def __init__(self,ip='127.0.0.1',port=8080,interval=10):
self.addr = (ip,port)
self.udpsocket = socket.socket(type=socket.SOCK_DGRAM)
self.event = threading.Event()
self.interval = interval
self.clients = {}
def start(self):
self.udpsocket.bind(self.addr)
threading.Thread(target=self.revice,name='recive').start()
def revice(self):
while not self.event.is_set():
lset = set()
data,raddr = self.udpsocket.recvfrom(1024)
logging.info(data)
logging.info(raddr)
current = datetime.datetime.now().timestamp()
if data.strip() == b'hb':
self.clients[raddr]=current
continue
elif data.strip() == b'quit': #有可能發出來的資料不在clients。
self.clients.pop(raddr,None)
logging.info('{}leaving'.format(raddr))
continue
self.clients[raddr] = current
message = '{}form{}{}'.format(data.decode(),*raddr)
for c ,stamp in self.clients.items():
if current - stamp >self.interval:
lset.add(c)
else:
self.udpsocket.sendto(message.encode(), raddr)
for c in lset:
self.clients.pop(c)
def stop(self):
self.event.set()
self.udpsocket.close()
def main():
cus = ChatUdpServer()
cus.start()
while True:
cmd = input('>>>>')
if cmd == 'quit':
cus.stop()
break
logging.info(threading.enumerate())
if __name__ == '__main__':
main()
正在迭代字典的時候不能進行pop。。
將其新增到set中,之後再進行pop。
Udp協議應用:
是無連線協議,基於以下假設:網路足夠好 訊息不會丟包,包不會亂序。
但是,即使在區域網,也不能保證不丟包,而且包到達不一定有序。
應用場景在視訊、音訊傳輸,一般來說,丟些包,問題不大,最多丟影象,聽不清話語,可以重新發話語來解決,海量採集資料,例如感測器發來的資料,丟十幾,幾百條沒有太大問題,DNS協議,資料內容小,一個包能 查詢到結果,不存在亂序,丟包,重新請求解析。
Udp效能優於tcp,但是可靠性場所適用於tcp。
Udp廣域網。
二、Socketserver
Socket程式設計過於底層,Python中對api進行封裝的就是socketserver模組,是網路程式設計框架,便於企業快速開發;
類的繼承關係
Socketserver簡化了網路伺服器的編寫。
有四個同步類:TCPserver,UDPserver,Unixstreamserver,Unixdatagramserver
2個Mixin類:ForkingMixin和threadingMixin類,用來支援非同步
Class forKingUDPserver(forKingMixin,UDPserver):pass
Class forKingTCPserver(forKingMixin,TCPPserver):pass
Class ThreadingUDPserver(ThreadingMixin,UDPserver):pass
Class ThreadingTCPserver(ThreadingMixin,TCPserver):pass
class BaseServer:
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 finish_request(self, request, client_address):
"""Finish one request by instantiating RequestHandlerClass."""
self.RequestHandlerClass(request, client_address, self)
baserequesthandler類:
他是和使用者連線的使用者請求處理類的基類,定義為baserequesthandler(request,client_address,server)
伺服器端server例項接受使用者的請求偶,最後會例項化這個類。
被初始化以後,送入三個構造引數,request,client_address,server 本身。
以後就可以在baserequesthandler類的例項上使用以下屬性:
Self.request是和客戶端的連線的socket物件,
Self.server是TCPserver本身
Self.client_address是客戶端地址。
這個類在初始化的過程中,會依次呼叫3個方法,子類可以覆蓋這些方法
class BaseRequestHandler:
"""Base class for request handler classes.
This class is instantiated for each request to be handled. The
constructor sets the instance variables request, client_address
and server, and then calls the handle() method. To implement a
specific service, all you need to do is to derive a class which
defines a handle() method.
The handle() method can find the request as self.request, the
client address as self.client_address, and the server (in case it
needs access to per-server information) as self.server. Since a
separate instance is created for each request, the handle() method
can define other arbitrary instance variables.
"""
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
import threading
import socketserver
class Myhandler(socketserver.BaseRequestHandler):
def handle(self):
print('--------')
print(self.server)
print(self.request)
print(self.client_address)
print(self.__dict__)
print(self.server.__dict__)
print(threading.enumerate())
print(threading.current_thread())
addr = ('127.0.0.1',8080)
server = socketserver.ThreadingTCPServer(addr,Myhandler)
server.serve_forever()
import threading
import socketserver
import logging
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class Myhandler(socketserver.BaseRequestHandler):
def handle(self):
print('--------')
print(self.server)
print(self.request)
print(self.client_address)
print(self.__dict__)
print(self.server.__dict__)
print(threading.enumerate())
print(threading.current_thread())
for i in range(2):
data = self.request.recv(1024)
logging.info(data)
addr = ('127.0.0.1',8080)
server = socketserver.ThreadingTCPServer(addr,Myhandler)
server.serve_forever()
類命名:
程式設計介面:
Socketserver.baseserver(server_address,RequestHandlerclass)
需要提供伺服器繫結的地址資訊,和用於請求處理請求的requesthandlerclass類。
Requesthandlerclass類必須是baserequesthandler類的子類,在baseserver中程式碼如下:
建立、傳埠、handler。
ThreadingTCPServer。多執行緒,非同步的,同時處理多個連線,
TCPServer TCP的,序列的。同步的,一個處理完畢後,才能處理下一個。只有主執行緒。
建立伺服器需要幾個步驟:
1)從baseRequestHandler類派生出子類,並覆蓋其handler()方法來建立請求處理程式類,此方法將處理傳入處理。
2)例項化一個伺服器類,傳參伺服器的地址和請求處理類。
3)呼叫伺服器例項的handle_request()或server_forever()方法。
4)呼叫server_close()關閉套接字。
實現echoserver
import threading
import socketserver
import logging
import sys
FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT,level=logging.INFO)
class EchoHandler(socketserver.BaseRequestHandler):
clients = {}
def setup(self):
self.event = threading.Event()
self.clients[self.client_address] = self.request
def finish(self):
self.event.set()
def handle(self):
while not self.event.is_set():
data = self.request.recv(1024)
logging.info(data)
if data == b'' or data =='quit':
break
msg = '{}'.format(data.decode())
for c in self.clients.values():
c.send(msg.encode())
addr = ('127.0.0.1',8080)
server = socketserver.ThreadingTCPServer(addr,EchoHandler)
# server.serve_forever()
t = threading.Thread(target=server.serve_forever,name='encho')
t.start()
if __name__ == '__main__':
try:
while True:
cmd = input('>>>')
if cmd.strip() == 'quit':
server.server_close()
break
logging.info(threading.enumerate())
except Exception as e:
logging.info(e)
finally:
print('exit')
sys.exit(0)
解決客戶端主動斷開連線伺服器端報錯的方式:客戶端主動斷開,會導致recv方法會立即返回一個空bytes,並沒有同事丟擲異常,當迴圈到recv這一句的時候就會丟擲異常,所以,可以通過判斷data資料是否為空客戶端是否斷開。
總結:
為每一個連線提供requesthandlerclass類例項,依次呼叫setup、handler、finish方法,且使用了try..finally結構,保證finish方法一定被呼叫、這些方法一次執行完畢,如果想維持這個連線和客戶端通訊,就需要在handler函式中迴圈。
所持socketserver模組提供不同的類,但是程式設計介面一樣的,即使是多程序、多執行緒的類也是一樣,大大減少了程式設計的難度。