欢迎光临
我们一直在努力

python select 模块

https://www.cnblogs.com/bigberg/p/8044581.html

一、前言

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))

赞(0) 打赏
未经允许不得转载:乌西塔 » python select 模块

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

觉得文章有用就打赏一下文章作者

微信扫一扫打赏