Python select
Python的select()方法直接調用操作系統的IO接口,它監控sockets,open files, and pipes(所有帶fileno()方法的文件句柄)何時變成readable 和writeable, 或者通信錯誤,select()使得同時監控多個連接變的簡單,並且這比寫一個長循環來等待和監控多客戶端連接要高效,因為select直接通過操作系統提供的C的網絡接口進行操作,而不是通過Python的解釋器。
注意:Using Python’s file objects with select() works for Unix, but is not supported under Windows.
二、select socket
接下來通過socket server例子要以了解select 是如何通過單進程實現同時處理多個非阻塞的socket連接的
2.1 socket server 開始監聽
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
import
select
import
socket
import
queue
import
sys
# Create a TCP/IP socket
server
=
socket.socket()
# set noblocking
server.setblocking(
False
)
# Bind the socket to the port
server_address
=
(
'localhost'
,
9999
)
print
(sys.stderr,
'starting up on %s port %s'
%
server_address)
server.bind(server_address)
# Listen for incoming connections
server.listen()
|
2.2 3個通信列表
select()方法接收並監控3個通信列表, 第一個是所有的輸入的data,就是指外部發過來的數據,第2個是監控和接收所有要發出去的data(outgoing data),第3個監控錯誤信息,接下來我們需要創建2個列表來包含輸入和輸出信息來傳給select()。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
import
select
import
socket
import
queue
import
sys
# Create a TCP/IP socket
server
=
socket.socket()
# set noblocking
server.setblocking(
False
)
# Bind the socket to the port
server_address
=
(
'localhost'
,
9999
)
print
(sys.stderr,
'starting up on %s port %s'
%
server_address)
server.bind(server_address)
# Listen for incoming connections
server.listen()
# 所有連接進來的對象都放在inputs
inputs
=
[server, ]
# 自己也要監控,因為server本身也是個對象
# 需要發送數據的對象
outputs
=
[]
|
2.3 添加一個隊列
所有客戶端的進來的連接和數據將會被server的主循環程序放在上面的list中處理,我們現在的server端需要等待連接可寫(writable)之后才能過來,然后接收數據並返回(因此不是在接收到數據之后就立刻返回),因為每個連接要把輸入或輸出的數據先緩存到queue里,然后再由select取出來再發出去。
Connections are added to and removed from these lists by the server main loop. Since this version of the server is going to wait for a socket to become writable before sending any data (instead of immediately sending the reply), each output connection needs a queue to act as a buffer for the data to be sent through it.
1
2
|
# 對外發送數據的隊列,記錄到字典中
message_queues
=
{}
|
2.4 主循環
1
2
3
|
while
True
:
readable, writable, exceptional
=
select.select(inputs, outputs, inputs)
# 如果沒有任何fd就緒,那程序就會一直阻塞在這里
|
當你把inputs,outputs,exceptional(這里跟inputs共用)傳給select()后,它返回3個新的list,我們上面將他們分別賦值為readable,writable,exceptional, 所有在readable list中的socket連接代表有數據可接收(recv),所有在writable list中的存放着你可以對其進行發送(send)操作的socket連接,當連接通信出現error時會把error寫到exceptional列表中。
2.5 Readable list
Readable list 中的socket 可以有3種可能狀態,第一種是如果這個socket是main "server" socket,它負責監聽客戶端的連接,如果這個main server socket出現在readable里,那代表這是server端已經ready來接收一個新的連接進來了,為了讓這個main server能同時處理多個連接,在下面的代碼里,我們把這個main server的socket設置為非阻塞模式。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
for
s
in
readable:
# 每一個s就是有個socket
if
s
is
server:
# 別忘記,上面我們server自己也當做一個fd放在了inputs列表里,傳給了select,如果這個s是server,代表server這個fd就緒了,
# 就是有活動了, 什么情況下它才有活動? 當然 是有新連接進來的時候
# 新連接進來了,接受這個連接
conn, client_addr
=
s.accept()
print
(
"new connection from"
, client_addr)
conn.setblocking(
0
)
inputs.append(conn)
# 為了不阻塞整個程序,我們不會立刻在這里開始接收客戶端發來的數據, 把它放到inputs里, 下一次loop時,這個新連接
# 就會被交給select去監聽,如果這個連接的客戶端發來了數據 ,那這個連接的fd在server端就會變成就續的,select就會把這個連接返回,
# 返回到readable 列表里,然后你就可以loop readable列表,取出這個連接,開始接收數據了, 下面就是這么干的
message_queues[conn]
=
queue.Queue()
# 接收到客戶端的數據后,不立刻返回 ,暫存在隊列里,以后發送
|
第二種情況是這個socket是已經建立了的連接,它把數據發了過來,這個時候你就可以通過recv()來接收它發過來的數據,然后把接收到的數據放到queue里,這樣你就可以把接收到的數據再傳回給客戶端了。
1
2
3
4
5
6
7
8
|
else
:
# s不是server的話,那就只能是一個 與客戶端建立的連接的fd了
# 客戶端的數據過來了,在這接收
data
=
s.recv(
1024
)
if
data:
print
(
'received [%s] from %s'
%
(data, s.getpeername()[
0
]))
message_queues[s].put(data)
# 收到的數據先放到queue里,一會返回給客戶端
if
s
not
in
outputs:
outputs.append(s)
# 為了不影響處理與其它客戶端的連接 , 這里不立刻返回數據給客戶端
|
第三種情況就是這個客戶端已經斷開了,所以你再通過recv()接收到的數據就為空了,所以這個時候你就可以把這個跟客戶端的連接關閉了。
1
2
3
4
5
6
7
8
9
10
11
12
13
|
else
:
# 如果收不到data代表什么呢? 代表客戶端斷開了
print
(
"client [%s] closed"
, s)
if
s
in
outputs:
# 既然客戶端都斷開了,我就不用再給它返回數據了,
# 所以這時候如果這個客戶端的連接對象還在outputs列表中,就把它刪掉
outputs.remove(s)
inputs.remove(s)
# 這個連接必然在inputs中,也刪掉
s.close()
# 關閉的連接在隊列中也刪除
del
message_queues[s]
|
2.6 writable list
對於writable list中的socket,也有幾種狀態,如果這個客戶端連接在跟它對應的queue里有數據,就把這個數據取出來再發回給這個客戶端,否則就把這個連接從output list中移除,這樣下一次循環select()調用時檢測到outputs list中沒有這個連接,那就會認為這個連接還處於非活動狀態
1
2
3
4
5
6
7
8
9
10
11
|
for
s
in
writable:
try
:
next_msg
=
message_queues[s].get_nowait()
except
queue.Empty:
# 沒有數據了,該連接對象隊列為空,停止檢測
print
(
'output queue for [%s] is empty'
%
s.getpeername()[
0
])
outputs.remove(s)
else
:
print
(
'send %s to %s'
%
(next_msg, s.getpeername()[
0
]))
s.send(next_msg)
|
2.7 exceptional condition
最后,如果在跟某個socket連接通信過程中出了錯誤,就把這個連接對象在inputs\outputs\message_queue中都刪除,再把連接關閉掉
1
2
3
4
5
6
7
8
9
10
|
for
s
in
exceptional:
print
(
'handling exceptional condition for'
, s.getpeername()[
0
])
# 從inputs中刪除
inputs.remove(s)
if
s
in
outputs:
outputs.remove(s)
s.close()
# 刪除隊列
del
message_queues[s]
|
注: getpeername() / getsocketname
getpeername可以獲得服務器的地址信息和端口號,正好和getsockname獲得本機地址信息和端口號完全相反
三、完整事例
select server
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
# -*- coding: UTF-8 -*-
import
select
import
socket
import
queue
import
sys
# Create a TCP/IP socket
server
=
socket.socket()
# set noblocking
server.setblocking(
False
)
# Bind the socket to the port
server_address
=
(
'localhost'
,
9999
)
print
(sys.stderr,
'starting up on %s port %s'
%
server_address)
server.bind(server_address)
# Listen for incoming connections
server.listen()
# 所有連接進來的對象都放在inputs
inputs
=
[server, ]
# 自己也要監控,因為server本身也是個對象
# 需要發送數據的對象
outputs
=
[]
# 對外發送數據的隊列,記錄到字典中
message_queues
=
{}
while
True
:
readable, writable, exceptional
=
select.select(inputs, outputs, inputs)
# 如果沒有任何fd就緒,那程序就會一直阻塞在這里
for
s
in
readable:
# 每一個s就是有個socket
if
s
is
server:
# 別忘記,上面我們server自己也當做一個fd放在了inputs列表里,傳給了select,如果這個s是server,代表server這個fd就緒了,
# 就是有活動了, 什么情況下它才有活動? 當然 是有新連接進來的時候
# 新連接進來了,接受這個連接
conn, client_addr
=
s.accept()
print
(
"new connection from"
, client_addr)
conn.setblocking(
0
)
inputs.append(conn)
# 為了不阻塞整個程序,我們不會立刻在這里開始接收客戶端發來的數據, 把它放到inputs里, 下一次loop時,這個新連接
# 就會被交給select去監聽,如果這個連接的客戶端發來了數據 ,那這個連接的fd在server端就會變成就續的,select就會把這個連接返回,
# 返回到readable 列表里,然后你就可以loop readable列表,取出這個連接,開始接收數據了, 下面就是這么干的
message_queues[conn]
=
queue.Queue()
# 接收到客戶端的數據后,不立刻返回 ,暫存在隊列里,以后發送
else
:
# s不是server的話,那就只能是一個 與客戶端建立的連接的fd了
# 客戶端的數據過來了,在這接收
data
=
s.recv(
1024
)
if
data:
print
(
'received [%s] from %s'
%
(data, s.getpeername()[
0
]))
message_queues[s].put(data)
# 收到的數據先放到queue里,一會返回給客戶端
if
s
not
in
outputs:
outputs.append(s)
# 為了不影響處理與其它客戶端的連接 , 這里不立刻返回數據給客戶端
else
:
# 如果收不到data代表什么呢? 代表客戶端斷開了
print
(
"client [%s] closed"
, s)
if
s
in
outputs:
# 既然客戶端都斷開了,我就不用再給它返回數據了,
# 所以這時候如果這個客戶端的連接對象還在outputs列表中,就把它刪掉
outputs.remove(s)
inputs.remove(s)
# 這個連接必然在inputs中,也刪掉
s.close()
# 關閉的連接在隊列中也刪除
del
message_queues[s]
for
s
in
writable:
try
:
next_msg
=
message_queues[s].get_nowait()
except
queue.Empty:
# 沒有數據了,該連接對象隊列為空,停止檢測
print
(
'output queue for [%s] is empty'
%
s.getpeername()[
0
])
outputs.remove(s)
else
:
print
(
'send %s to %s'
%
(next_msg, s.getpeername()[
0
]))
s.send(next_msg)
for
s
in
exceptional:
print
(
'handling exceptional condition for'
, s.getpeername()[
0
])
# 從inputs中刪除
inputs.remove(s)
if
s
in
outputs:
outputs.remove(s)
s.close()
# 刪除隊列
del
message_queues[s]
|
client
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
# -*- coding: UTF-8 -*-
import
socket
HOST
=
'localhost'
# The remote host
PORT
=
9999
# The same port as used by the server
s
=
socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
while
True
:
msg
=
bytes(
input
(
">>:"
), encoding
=
"utf8"
)
s.sendall(msg)
data
=
s.recv(
1024
)
# print(data)
print
(
'Received'
,
repr
(data))
|
python select模塊詳解
要理解select.select模塊其實主要就是要理解它的參數, 以及其三個返回值。
select()方法接收並監控3個通信列表, 第一個是所有的輸入的data,就是指外部發過來的數據,第2個是監控和接收所有要發出去的data(outgoing data),第3個監控錯誤信息
在網上一直在找這個select.select的參數解釋, 但實在是沒有, 哎...自己硬着頭皮分析了一下。
readable, writable, exceptional = select.select(inputs, outputs, inputs)
select 函數的參數其實很好理解, 前提是我們對unix 網絡編程有了解. select 模型是unix 系統中的網絡模型, python 將其封裝了,因此我們使用起來就比較方便, 但是面試官就不會這么覺得了(最近被面試逼瘋了, 考慮問題都從面試官的角度考慮), 先說下unix 系統中的select 模型吧, 參數原型:
int select(int maxfdpl, fd_set * readset, fd_set *writeset, fd_set *exceptset, const struct timeval * tiomeout)
第一個是最大的文件描述符長度
第二個是監聽的可讀集合
第三個是監聽的可寫集合
第四個是監聽的異常集合
第五個是時間限制
對struct fd_set結構體操作的宏
FD_SETSIZE 容量,指定fd_array數組大小,默認為64,也可自己修改宏
FD_ZERO(*set) 置空,使數組的元素值都為3435973836,元素個數為0.
FD_SET(s, *set) 添加,向 struct fd_set結構體添加套接字s
FD_ISSET(s, *set) 判斷,判斷s是否為 struct fd_set結構體中的一員
FD_CLR(s, *set) 刪除,從 struct fd_set結構體中刪除成員s
因為此模型主要是在網絡中應用, 我們不考慮文件, 設備, 單從套接字來考慮, 可讀條件如下:
可寫條件如下:
我看C 示例的時候, 看的有點懵逼, 應該需要跑一遍代碼就好, python 就簡單了, 直接調用封裝好的select , 其底層處理好了文件描述符的相關讀寫監聽(回頭再研究下), 我們在Python 中只需這么寫:
can_read, can_write, _ = select.select(inputs, outputs, None, None)
第一個參數是我們需要監聽可讀的套接字, 第二個參數是我們需要監聽可寫的套接字, 第三個參數使我們需要監聽異常的套接字, 第四個則是時間限制設置.
如果監聽的套接字滿足了可讀可寫條件, 那么所返回的can,read 或是 can_write就會有值了, 然后我們就可以利用這些返回值進行隨后的操作了。相比較unix 的select模型, 其select函數的返回值是一個整型, 用以判斷是否執行成功.
第一個參數就是服務器端的socket, 第二個是我們在運行過程中存儲的客戶端的socket, 第三個存儲錯誤信息。
重點是在返回值, 第一個返回的是可讀的list, 第二個存儲的是可寫的list, 第三個存儲的是錯誤信息的
list。
網上所有關於select.select的代碼都是差不多的, 但是有些不能運行, 或是不全。我自己重新寫了一份能運行的程序, 做了很多注釋, 好好看看就能搞懂

# coding: utf-8 import select import socket import Queue from time import sleep # Create a TCP/IP server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setblocking(False) # Bind the socket to the port server_address = ('localhost', 8090) print ('starting up on %s port %s' % server_address) server.bind(server_address) # Listen for incoming connections server.listen(5) # Sockets from which we expect to read inputs = [server] # Sockets to which we expect to write # 處理要發送的消息 outputs = [] # Outgoing message queues (socket: Queue) message_queues = {} while inputs: # Wait for at least one of the sockets to be ready for processing print ('waiting for the next event') # 開始select 監聽, 對input_list 中的服務器端server 進行監聽 # 一旦調用socket的send, recv函數,將會再次調用此模塊 readable, writable, exceptional = select.select(inputs, outputs, inputs) # Handle inputs # 循環判斷是否有客戶端連接進來, 當有客戶端連接進來時select 將觸發 for s in readable: # 判斷當前觸發的是不是服務端對象, 當觸發的對象是服務端對象時,說明有新客戶端連接進來了 # 表示有新用戶來連接 if s is server: # A "readable" socket is ready to accept a connection connection, client_address = s.accept() print ('connection from', client_address) # this is connection not server connection.setblocking(0) # 將客戶端對象也加入到監聽的列表中, 當客戶端發送消息時 select 將觸發 inputs.append(connection) # Give the connection a queue for data we want to send # 為連接的客戶端單獨創建一個消息隊列,用來保存客戶端發送的消息 message_queues[connection] = Queue.Queue() else: # 有老用戶發消息, 處理接受 # 由於客戶端連接進來時服務端接收客戶端連接請求,將客戶端加入到了監聽列表中(input_list), 客戶端發送消息將觸發 # 所以判斷是否是客戶端對象觸發 data = s.recv(1024) # 客戶端未斷開 if data != '': # A readable client socket has data print ('received "%s" from %s' % (data, s.getpeername())) # 將收到的消息放入到相對應的socket客戶端的消息隊列中 message_queues[s].put(data) # Add output channel for response # 將需要進行回復操作socket放到output 列表中, 讓select監聽 if s not in outputs: outputs.append(s) else: # 客戶端斷開了連接, 將客戶端的監聽從input列表中移除 # Interpret empty result as closed connection print ('closing', client_address) # Stop listening for input on the connection if s in outputs: outputs.remove(s) inputs.remove(s) s.close() # Remove message queue # 移除對應socket客戶端對象的消息隊列 del message_queues[s] # Handle outputs # 如果現在沒有客戶端請求, 也沒有客戶端發送消息時, 開始對發送消息列表進行處理, 是否需要發送消息 # 存儲哪個客戶端發送過消息 for s in writable: try: # 如果消息隊列中有消息,從消息隊列中獲取要發送的消息 message_queue = message_queues.get(s) send_data = '' if message_queue is not None: send_data = message_queue.get_nowait() else: # 客戶端連接斷開了 print "has closed " except Queue.Empty: # 客戶端連接斷開了 print "%s" % (s.getpeername()) outputs.remove(s) else: # print "sending %s to %s " % (send_data, s.getpeername) # print "send something" if message_queue is not None: s.send(send_data) else: print "has closed " # del message_queues[s] # writable.remove(s) # print "Client %s disconnected" % (client_address) # # Handle "exceptional conditions" # 處理異常的情況 for s in exceptional: print ('exception condition on', s.getpeername()) # Stop listening for input on the connection inputs.remove(s) if s in outputs: outputs.remove(s) s.close() # Remove message queue del message_queues[s] sleep(1)

# coding: utf-8 import socket messages = ['This is the message ', 'It will be sent ', 'in parts ', ] server_address = ('localhost', 8090) # Create aTCP/IP socket socks = [socket.socket(socket.AF_INET, socket.SOCK_STREAM), socket.socket(socket.AF_INET, socket.SOCK_STREAM), ] # Connect thesocket to the port where the server is listening print ('connecting to %s port %s' % server_address) # 連接到服務器 for s in socks: s.connect(server_address) for index, message in enumerate(messages): # Send messages on both sockets for s in socks: print ('%s: sending "%s"' % (s.getsockname(), message + str(index))) s.send(bytes(message + str(index)).decode('utf-8')) # Read responses on both sockets for s in socks: data = s.recv(1024) print ('%s: received "%s"' % (s.getsockname(), data)) if data != "": print ('closingsocket', s.getsockname()) s.close()
一、select介紹
select()的機制中提供一fd_set的數據結構,實際上是一long類型的數組, 每一個數組元素都能與一打開的文件句柄(不管是Socket句柄,還是其他文件或命名管道或設備句柄)建立聯系,建立聯系的工作由程序員完成, 當調用select()時,由內核根據IO狀態修改fd_set的內容,由此來通知執行了select()的進程哪一Socket或文件可讀或可寫。主要用於Socket通信當中。
總結:select主要用於socket通信當中,能監視我們需要的文件描述變化。
二、非阻塞式I/O編程特點
2.1、如果一個發現I/O有輸入,讀取的過程中,另外一個也有了輸入,這時候不會產生任何反應.這就需要你的程序語句去用到select函數的時候才知道有數據輸入。
2.2、程序去select的時候,如果沒有數據輸入,程序會一直等待,直到有數據為止,也就是程序中無需循環和sleep。
Select在Socket編程中還是比較重要的,可是對於初學Socket的人來說都不太愛用Select寫程序,他們只是習慣寫諸如connect、accept、recv或recvfrom這樣的阻塞程序(所謂阻塞方式block,顧名思義,就是進程或是線程執行到這些函數時必須等待某個事件的發生,如果事件沒有發生,進程或線程就被阻塞,函數不能立即返回)。
可是使用Select就可以完成非阻塞(所謂非阻塞方式non-block,就是進程或線程執行此函數時不必非要等待事件的發生,一旦執行肯定返回,以返回值的不同來反映函數的執行情況,如果事件發生則與阻塞方式相同,若事件沒有發生,則返回一個代碼來告知事件未發生,而進程或線程繼續執行,所以效率較高)方式工作的程序,它能夠監視我們需要監視的文件描述符的變化情況——讀寫或是異常。
返回值:准備就緒的描述符數,若超時則返回0,若出錯則返回-1。
三、示例
示例1:模擬select,同時監聽多個端口

import socket import select sk1 = socket.socket() sk1.bind(('0.0.0.0', 8001)) sk1.listen() sk2 = socket.socket() sk2.bind(('0.0.0.0', 8002)) sk2.listen() sk3 = socket.socket() sk3.bind(('0.0.0.0', 8003)) sk3.listen() inputs = [sk1, sk2, sk3, ] while True: r_list, w_list, e_list = select.select(inputs,[],inputs,1) for sk in r_list: # conn表示每一個連接對象 conn, address = sk.accept() conn.sendall(bytes('hello', encoding='utf-8')) conn.close() for sk in e_list: inputs.remove(sk) 解釋: # select內部自動監聽sk1,sk2,sk3三個對象,監聽三個句柄是否發生變化,把發生變化的元素放入r_list中。 # 如果有人連接sk1,則r_list = [sk1] # 如果有人連接sk1和sk2,則r_list = [sk1,sk2] # select中第1個參數表示inputs中發生變化的句柄放入r_list。 # select中第2個參數表示[]中的值原封不動的傳遞給w_list。 # select中第3個參數表示inputs中發生錯誤的句柄放入e_list。 # 參數1表示1秒監聽一次 # 當有用戶連接時,r_list里面的內容[<socket.socket fd=220, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('0.0.0.0', 8001)>]

import socket obj = socket.socket() obj.connect(('127.0.0.1', 8001)) content = str(obj.recv(1024), encoding='utf-8') print(content) obj.close() # 客戶端c2.py import socket obj = socket.socket() obj.connect(('127.0.0.1', 8002)) content = str(obj.recv(1024), encoding='utf-8') print(content) obj.close()
示例2:IO多路復用--使用socket模擬多線程,並實現讀寫分離

#使用socket模擬多線程,使多用戶可以同時連接 import socket import select sk1 = socket.socket() sk1.bind(('0.0.0.0', 8001)) sk1.listen() inputs = [sk1, ] outputs = [] message_dict = {} while True: r_list, w_list, e_list = select.select(inputs, outputs, inputs, 1) print('正在監聽的socket對象%d' % len(inputs)) print(r_list) for sk1_or_conn in r_list: #每一個連接對象 if sk1_or_conn == sk1: # 表示有新用戶來連接 conn, address = sk1_or_conn.accept() inputs.append(conn) message_dict[conn] = [] else: # 有老用戶發消息了 try: data_bytes = sk1_or_conn.recv(1024) except Exception as ex: # 如果用戶終止連接 inputs.remove(sk1_or_conn) else: data_str = str(data_bytes, encoding='utf-8') message_dict[sk1_or_conn].append(data_str) outputs.append(sk1_or_conn) #w_list中僅僅保存了誰給我發過消息 for conn in w_list: recv_str = message_dict[conn][0] del message_dict[conn][0] conn.sendall(bytes(recv_str+'好', encoding='utf-8')) outputs.remove(conn) for sk in e_list: inputs.remove(sk)

import socket obj = socket.socket() obj.connect(('127.0.0.1', 8001)) while True: inp = input('>>>') obj.sendall(bytes(inp, encoding='utf-8')) ret = str(obj.recv(1024),encoding='utf-8') print(ret) obj.close()