今天主題是實現併發伺服器,實現方法有多種版本,先從簡單的單進程代碼實現到多進程,多線程的實現,最終引入一些高級模塊來實現併發TCP伺服器。 說到TCP,想起吐槽大會有個段子提到三次握手,也只有程式猿(媛)能get。 UDP伺服器數據傳輸不可靠,這裡就忽略了。 >>: 簡單的單進程TCP伺服器 假代碼 ...
今天主題是實現併發伺服器,實現方法有多種版本,先從簡單的單進程代碼實現到多進程,多線程的實現,最終引入一些高級模塊來實現併發TCP伺服器。
說到TCP,想起吐槽大會有個段子提到三次握手,也只有程式猿(媛)能get。
UDP伺服器數據傳輸不可靠,這裡就忽略了。
>>:
簡單的單進程TCP伺服器
假代碼:
#創建tcp伺服器套接字
#綁定埠
#設置正常情況退出的伺服器下,埠可以重用
#設置監聽,變為主動監聽
# 等待客戶端的鏈接,返回新的socket和地址
#關閉tcp伺服器套接字
from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDR
#創建tcp伺服器套接字
server_socket = socket(AF_INET,SOCK_STREAM)
#綁定埠
server_socket.bind(("",9999))
#設置正常情況退出的伺服器下,埠可以重用
server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
#設置監聽,變為主動監聽
server_socket.listen(5)
while True:
# 等待客戶端的鏈接,返回新的socket和地址
new_socket,new_address = server_socket.accept()
#接收數據,並且發送數據
try:
while True:
recv_data = new_socket.recv(1024)
#當有客戶端關閉後,recv解除阻塞,並且返回長度為0
if len(recv_data) > 0:
recv_content = recv_data.decode("gb2312")
print("收到:%s的信息是:%s" % (str(new_address),recv_content))
new_socket.send("thank you!".encode("gb2312"))
else:
print("客戶端%s已經關閉" % (str(new_address)))
break
finally:
new_socket.close()
print("關閉%s客戶端" % (str(new_address)))
#關閉tcp伺服器套接字
server_socket.close()
多進程TCP伺服器
from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDR
from multiprocessing import Process
#在子進程中接收消息
def recv_data(new_socket,new_address):
while True:
recv_data = new_socket.recv(1024)
# 當有客戶端關閉後,recv解除阻塞,並且返回長度為0
if len(recv_data) > 0:
recv_content = recv_data.decode("gb2312")
print("收到:%s的信息是:%s" % (str(new_address), recv_content))
new_socket.send("thank you!".encode("gb2312"))
else:
print("客戶端%s已經關閉" % (str(new_address)))
break
#關閉與客戶端的連接
print("關閉與客戶端的連接")
new_socket.close()
def main():
#創建tcp伺服器套接字
server_socket = socket(AF_INET,SOCK_STREAM)
#綁定埠
server_socket.bind(("",8888))
#設置正常情況退出的伺服器下,埠可以重用
server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
#設置監聽,變為被動連接
server_socket.listen(3)
try:
while True:
# 等待客戶端的鏈接,返回新的socket和地址
new_socket,new_address = server_socket.accept()
#接收數據,並且發送數據
Process(target=recv_data,args=(new_socket,new_address)).start()
#因為主進程和子進程不共用數據
#如果我們直接關閉new_socket,只是關閉主進程的new_socket,而子進程的不受影響
new_socket.close()
finally:
#關閉tcp伺服器套接字
server_socket.close()
if __name__ == "__main__":
main()
多進程TCP伺服器
from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDR
from multiprocessing import Process
#在子進程中接收消息
def recv_data(new_socket,new_address):
while True:
recv_data = new_socket.recv(1024)
# 當有客戶端關閉後,recv解除阻塞,並且返回長度為0
if len(recv_data) > 0:
recv_content = recv_data.decode("gb2312")
print("收到:%s的信息是:%s" % (str(new_address), recv_content))
new_socket.send("thank you!".encode("gb2312"))
else:
print("客戶端%s已經關閉" % (str(new_address)))
break
#關閉與客戶端的連接
print("關閉與客戶端的連接")
new_socket.close()
def main():
#創建tcp伺服器套接字
server_socket = socket(AF_INET,SOCK_STREAM)
#綁定埠
server_socket.bind(("",8888))
#設置正常情況退出的伺服器下,埠可以重用
server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
#設置監聽,變為被動連接
server_socket.listen(3)
try:
while True:
# 等待客戶端的鏈接,返回新的socket和地址
new_socket,new_address = server_socket.accept()
#接收數據,並且發送數據
Process(target=recv_data,args=(new_socket,new_address)).start()
#因為主進程和子進程不共用數據
#如果我們直接關閉new_socket,只是關閉主進程的new_socket,而子進程的不受影響
new_socket.close()
finally:
#關閉tcp伺服器套接字
server_socket.close()
if __name__ == "__main__":
main()
多線程TCP伺服器
from socket import socket, AF_INET,SOCK_STREAM,SOL_SOCKET,SO_REUSEADDR
from threading import Thread
#接收消息
def recv_data(new_socket,new_address):
while True:
recv_data = new_socket.recv(1024)
# 當有客戶端關閉後,recv解除阻塞,並且返回長度為0
if len(recv_data) > 0:
recv_content = recv_data.decode("gb2312")
print("收到:%s的信息是:%s" % (str(new_address), recv_content))
new_socket.send("thank you!".encode("gb2312"))
else:
print("客戶端%s已經關閉" % (str(new_address)))
break
def main():
#創建tcp伺服器套接字
server_socket = socket(AF_INET,SOCK_STREAM)
#綁定埠
server_socket.bind(("",9999))
#設置正常情況退出的伺服器下,埠可以重用
server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
#設置監聽,變為被動連接
server_socket.listen(3)
try:
while True:
# 等待客戶端的鏈接,返回新的socket和地址
new_socket,new_address = server_socket.accept()
#接收數據,並且發送數據
Thread(target=recv_data,args=(new_socket,new_address)).start()
finally:
#關閉tcp伺服器套接字
server_socket.close()
if __name__ == "__main__":
main()
多任務協程實現 ——
greenlet和gevent
#coding=utf-8
from greenlet import greenlet
import time
def test1():
while True:
print "---A--"
gr2.switch()
time.sleep(0.5)
def test2():
while True:
print "---B--"
gr1.switch()
time.sleep(0.5)
gr1 = greenlet(test1)
gr2 = greenlet(test2)
#切換到gr1中運行
gr1.switch()
-----------------------------------------------
import gevent
#函數
def f(n):
for i in range(n):
print("%s:%s" % (gevent.getcurrent(),i))
f1 = gevent.spawn(f,5)
f2 = gevent.spawn(f,5)
f3 = gevent.spawn(f,5)
#讓主線程等待三個協程執行完畢,否則沒有機會執行
f1.join()
f2.join()
f3.join()
#可以看到,3個greenlet是依次運行而不是交替運行。要讓greenlet交替運行,可以通過gevent.sleep()交出控制權。
--------------------------------------------------
#coding=utf-8
import gevent
def f(n):
for i in range(n):
print gevent.getcurrent(), i
#用來模擬一個耗時操作,註意不是time模塊中的sleep
gevent.sleep(1)
g1 = gevent.spawn(f, 5)
g2 = gevent.spawn(f, 5)
g3 = gevent.spawn(f, 5)
#下麵三行代碼意思:主線程等待各個協成支持完,否則協成沒有機會執行
g1.join()
g2.join()
g3.join()
單進程TCP伺服器 ——
非堵塞式
from socket import AF_INET,socket,SO_REUSEADDR,SOCK_STREAM,SOL_SOCKET
def main():
#創建tcp的socket套接字
server_socket = socket(AF_INET,SOCK_STREAM)
server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
#綁定埠
server_socket.bind(("",9999))
#設置非阻塞,也就是說accept方法不阻塞了,
# 但是在沒有客戶端鏈接且被執行的時候會報錯
#有客戶端鏈接的時候正常執行
server_socket.setblocking(False)
#設置監聽
server_socket.listen(5)
#客戶端列表
client_lists = []
try:
#不斷調用accept
while True:
try:
# print("accept--111")
new_socket,new_address = server_socket.accept()
print("accept--2222")
except Exception as result:
# print(result)
pass
else:
print("新的客戶%s鏈接上" % str(new_address))
#新鏈接的new_sokect預設也是阻塞,也設置為非阻塞後,recv為非阻塞
new_socket.setblocking(False)
client_lists.append((new_socket,new_address))
# print(111)
for client_sokect,client_address in client_lists:
#接收數據
try:
recv_data = client_sokect.recv(1024)
except Exception as result:
# print(result)
pass
else:
# print("正常數據:%s" %recv_data)
if len(recv_data) > 0 :
print("收到%s:%s" % (str(client_address),recv_data))
client_sokect.send("thank you!".encode("gb2312"))
else:
#客戶端已經埠,要把該客戶端從列表中異常
client_lists.remove((client_sokect,new_address))
client_sokect.close()
print("%s已經斷開" % str(new_address))
finally:
#關閉套接字
server_socket.close()
if __name__ == "__main__":
main()
單進程TCP伺服器 ——
select版
select 原理
其他語言(c或者c++)也有使用select實現多任務伺服器。
select 能夠完成一些套接字的檢查,從頭到尾檢查一遍後,標記哪些套接字是否可以收數據,返回的時候,就返回能接收數據的套接字,返回的是列表。select是由操作系統提供的,效率要高些,非常快的方式檢測哪些套接字可以接收數據。select是跨平臺的,在window也可以用。
io多路復用:沒有使用多進程和多線程的情況下完成多個套接字的使用。
from socket import AF_INET,socket,SO_REUSEADDR,SOCK_STREAM,SOL_SOCKET
from select import select
import sys
def main():
#創建tcp的socket套接字
server_socket = socket(AF_INET,SOCK_STREAM)
server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
#綁定埠
server_socket.bind(("",9999))
#設置監聽
server_socket.listen(5)
#客戶端列表
socket_lists = [server_socket,sys.stdin]
wirte_list = []
#是否退出
is_run = False
try:
while True:
#檢測列表client_lists那些socket可以接收數據,
#檢測列表[]那些套接字(socket)可否發送數據
#檢測列表[]那些套接字(socket)是否產生了異常
print("select--111")
#這個select函數預設是堵塞,當有客戶端鏈接的時候解除阻塞,
# 當有數據可以接收的時候解除阻塞,當客戶端斷開的時候解除阻塞
readable, wirteable,excep = select(socket_lists,wirte_list,[])
# print("select--2222")
# print(111)
for sock in wirteable:
#這個會一直發送,因為他是處於已經發的狀態
sock.send("thank you!".encode("gb2312"))
for sock in readable:
#接收數據
if sock == server_socket:
print("sock == server_socket")
#有新的客戶端鏈接進來
new_socket,new_address = sock.accept()
#新的socket添加到列表中,便於下次socket的時候能檢查到
socket_lists.append(new_socket)
elif sock == sys.stdin:
cmd = sys.stdin.readline()
print(cmd)
is_run = cmd
else:
# print("sock.recv(1024)....")
#此時的套接字sock是直接可以取數據的
recv_data = sock.recv(1024)
if len(recv_data) > 0:
print("從[%s]:%s" % (str(new_address),recv_data))
sock.send(recv_data)
#把鏈接上有消息接收的socket添加到監聽寫的列表中
wirte_list.append(sock)
else:
print("客戶端已經斷開")
#客戶端已經斷開,要移除
sock.close()
socket_lists.remove(sock)
#是否退出程式
if is_run:
break
finally:
#關閉套接字
server_socket.close()
if __name__ == "__main__":
main()
單進程TCP伺服器 ——
epoll版
from socket import *
import select
def main():
#創建tcp伺服器套接字
server_socket = socket(AF_INET,SOCK_STREAM)
#設置埠可以重用
server_socket.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
#綁定埠
server_socket.bind(("",9999))
#設置監聽
server_socket.listen(5)
#用epoll設置監聽收數據
epoll = select.epoll()
#把server_socket註冊到epoll的事件監聽中,如果已經註冊過會發生異常
epoll.register(server_socket.fileno(),select.EPOLLIN|select.EPOLLET)
#裝socket列表
socket_lists = {}
#裝socket對應的地址
socket_address = {}
while True:
#返回套接字列表[(socket的文件描述符,select.EPOLLIN)],
# 如果有新的鏈接,有數據發過來,斷開鏈接等都會解除阻塞
print("epoll.poll--111")
epoll_list = epoll.poll()
print("epoll.poll--222")
print(epoll_list)
for fd,event in epoll_list:
#有新的鏈接
if fd == server_socket.fileno():
print("新的客戶fd==%s" % fd)
new_sokect,new_address = server_socket.accept()
#往字典添加數據
socket_lists[new_sokect.fileno()] = new_sokect
socket_address[new_sokect.fileno()] = new_address
#註冊新的socket也註冊到epoll的事件監聽中
epoll.register(new_sokect.fileno(), select.EPOLLIN | select.EPOLLET)
elif event ==select.EPOLLIN:
print("收到數據了")
#根據文件操作符取出對應socket
new_sokect = socket_lists[fd]
address = socket_address[fd]
recv_data = new_sokect.recv(1024)
if len(recv_data) > 0:
print("已經收到[%s]:%s" % (str(address),recv_data.decode("gb2312")))
else:
#客戶端埠,取消監聽
epoll.unregister(fd)
#關閉鏈接
new_sokect.close()
print("[%s]已經下線" % str(address))
#關閉套接字鏈接
server_socket.close()
if __name__ == "__main__":
main()
單進程TCP伺服器 ——
gevent版
gevent原理
greenlet已經實現了協程,但是這個還得人工切換,是不是覺得太麻煩了,莫要捉急,python還有一個比greenlet更強大的並且能夠自動切換任務的模塊gevent
原理------當一個greenlet遇到IO(指的是input output 輸入輸出,比如網路、文件操作等)操作時,比如訪問網路,就自動切換到其他的greenlet,等到IO操作完成,再在適當的時候切換回來繼續執行。
由於IO操作非常耗時,經常使程式處於等待狀態,有了gevent為我們自動切換協程,就保證總有greenlet在運行,而不是等待IO.
import sys
import time
import gevent
from gevent import socket,monkey
monkey.patch_all()
def handle_request(conn):
while True:
data = conn.recv(1024)
if not data:
conn.close()
break
print("recv:", data)
conn.send(data)
def server(port):
s = socket.socket()
s.bind(('', port))
s.listen(5)
while True:
newSocket, addr = s.accept()
gevent.spawn(handle_request, newSocket)
if __name__ == '__main__':
server(7788)
首先基於以上代碼模塊,撒點概念問題:
1.什麼是協程?
協程:存線上程中,是比線程更小的執行單元,又稱微線程,纖程。自帶cpu上下文,操作協程由程式員決定,它可以將一個線程分解為多個微線程,每個協程間共用全局空間的變數,每秒鐘切換頻率高達百萬次。
2. 什麼是計算密集型和IO密集型
計算密集型:要進行大量的計算,消耗cpu資源。如複雜計算,對視頻進行高清解碼等,全靠cpu的運算能力。而計算密集型任務完成多任務切換任務比較耗時,cpu執行任務效率就越低。在python中,多進程適合計算密集型任務。
IO密集型:涉及到網路、磁碟io的任務都是io密集型。cpu消耗少,計算量小,如請求網頁,讀寫文件等。在python中,使用sleep達到IO密集型任務的目的,多線程適合IO密集型任務。
各大實現版本對比:
select:
1)支持跨平臺,最大缺陷是單個進程打開的FD是有限的,由FD_SETSIZE設置,預設是1024;
2)對socket掃描時是線性掃描,及採用輪詢方式,效率低;
3)需要維護一個存放大量FD的數據結構,使得用戶空間和內核空間在傳遞該數據結構時複製開銷大。
poll:
1)poll與select本質上沒有區別,但poll沒有最大連接數的限制;
2)大量的fd數組被整體複製於用戶態和內核地址空間之間,不管這樣的複製是不是有意義;
3)‘水平觸發’,如果報告了fd後,沒有被處理,下次poll時還會再次報告該fd。
epoll:
1)是之前poll和select的增強版,epoll更靈活,沒有描述符限制,能打開的fd遠大於1024(1G的記憶體上能監聽約10萬個埠);
2)‘邊緣出發’,事件通知機制,效率提升,最大的特點在於它只管你活躍的連接,而跟連接總數無關。而epoll對文件描述符的操作模式之一ET是一種高效的工作方式,很大程度減少事件反覆觸發的次數,內核不會發送更多的通知(only once)。