AI专业者开发社区 1 号成员
  •  0 回帖  •  10 浏览  •  5 个月前
 

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)


</span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> Handle inputs</span>
<span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 循环判断是否有客户端连接进来, 当有客户端连接进来时select 将触发</span>
<span style="color: rgba(0, 0, 255, 1)">for</span> s <span style="color: rgba(0, 0, 255, 1)">in</span><span style="color: rgba(0, 0, 0, 1)"> readable:
    </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 判断当前触发的是不是服务端对象, 当触发的对象是服务端对象时,说明有新客户端连接进来了</span>
    <span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 表示有新用户来连接</span>
    <span style="color: rgba(0, 0, 255, 1)">if</span> s <span style="color: rgba(0, 0, 255, 1)">is</span><span style="color: rgba(0, 0, 0, 1)"> server:
        </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> A "readable" socket is ready to accept a connection</span>
        connection, client_address =<span style="color: rgba(0, 0, 0, 1)"> s.accept()
        </span><span style="color: rgba(0, 0, 255, 1)">print</span> (<span style="color: rgba(128, 0, 0, 1)">'</span><span style="color: rgba(128, 0, 0, 1)">connection from</span><span style="color: rgba(128, 0, 0, 1)">'</span><span style="color: rgba(0, 0, 0, 1)">, client_address)
        </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> this is connection not server</span>



            connection.setblocking(0)
# 将客户端对象也加入到监听的列表中, 当客户端发送消息时 select 将触发
            inputs.append(connection)


        </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> Give the connection a queue for data we want to send</span>
        <span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 为连接的客户端单独创建一个消息队列，用来保存客户端发送的消息</span>
        message_queues[connection] =<span style="color: rgba(0, 0, 0, 1)"> Queue.Queue()
    </span><span style="color: rgba(0, 0, 255, 1)">else</span><span style="color: rgba(0, 0, 0, 1)">:
        </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 有老用户发消息, 处理接受</span>
        <span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 由于客户端连接进来时服务端接收客户端连接请求，将客户端加入到了监听列表中(input_list), 客户端发送消息将触发</span>
        <span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 所以判断是否是客户端对象触发</span>
        data = s.recv(1024<span style="color: rgba(0, 0, 0, 1)">)
        </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 客户端未断开</span>
        <span style="color: rgba(0, 0, 255, 1)">if</span> data != <span style="color: rgba(128, 0, 0, 1)">''</span><span style="color: rgba(0, 0, 0, 1)">:
            </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> A readable client socket has data</span>
            <span style="color: rgba(0, 0, 255, 1)">print</span> (<span style="color: rgba(128, 0, 0, 1)">'</span><span style="color: rgba(128, 0, 0, 1)">received "%s" from %s</span><span style="color: rgba(128, 0, 0, 1)">'</span> %<span style="color: rgba(0, 0, 0, 1)"> (data, s.getpeername()))
            </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 将收到的消息放入到相对应的socket客户端的消息队列中</span>



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


            </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> Remove message queue</span>
            <span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 移除对应socket客户端对象的消息队列</span>
            <span style="color: rgba(0, 0, 255, 1)">del</span><span style="color: rgba(0, 0, 0, 1)"> message_queues[s]

</span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> Handle outputs</span>
<span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 如果现在没有客户端请求, 也没有客户端发送消息时, 开始对发送消息列表进行处理, 是否需要发送消息</span>
<span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 存储哪个客户端发送过消息</span>
<span style="color: rgba(0, 0, 255, 1)">for</span> s <span style="color: rgba(0, 0, 255, 1)">in</span><span style="color: rgba(0, 0, 0, 1)"> writable:
    </span><span style="color: rgba(0, 0, 255, 1)">try</span><span style="color: rgba(0, 0, 0, 1)">:
        </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 如果消息队列中有消息,从消息队列中获取要发送的消息</span>
        message_queue =<span style="color: rgba(0, 0, 0, 1)"> message_queues.get(s)
        send_data </span>= <span style="color: rgba(128, 0, 0, 1)">''</span>
        <span style="color: rgba(0, 0, 255, 1)">if</span> message_queue <span style="color: rgba(0, 0, 255, 1)">is</span> <span style="color: rgba(0, 0, 255, 1)">not</span><span style="color: rgba(0, 0, 0, 1)"> None:
            send_data </span>=<span style="color: rgba(0, 0, 0, 1)"> message_queue.get_nowait()
        </span><span style="color: rgba(0, 0, 255, 1)">else</span><span style="color: rgba(0, 0, 0, 1)">:
            </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 客户端连接断开了</span>
            <span style="color: rgba(0, 0, 255, 1)">print</span> <span style="color: rgba(128, 0, 0, 1)">"</span><span style="color: rgba(128, 0, 0, 1)">has closed </span><span style="color: rgba(128, 0, 0, 1)">"</span>
    <span style="color: rgba(0, 0, 255, 1)">except</span><span style="color: rgba(0, 0, 0, 1)"> Queue.Empty:
        </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 客户端连接断开了</span>
        <span style="color: rgba(0, 0, 255, 1)">print</span> <span style="color: rgba(128, 0, 0, 1)">"</span><span style="color: rgba(128, 0, 0, 1)">%s</span><span style="color: rgba(128, 0, 0, 1)">"</span> %<span style="color: rgba(0, 0, 0, 1)"> (s.getpeername())
        outputs.remove(s)
    </span><span style="color: rgba(0, 0, 255, 1)">else</span><span style="color: rgba(0, 0, 0, 1)">:
        </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> print "sending %s to %s " % (send_data, s.getpeername)</span>
        <span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> print "send something"</span>
        <span style="color: rgba(0, 0, 255, 1)">if</span> message_queue <span style="color: rgba(0, 0, 255, 1)">is</span> <span style="color: rgba(0, 0, 255, 1)">not</span><span style="color: rgba(0, 0, 0, 1)"> None:
            s.send(send_data)
        </span><span style="color: rgba(0, 0, 255, 1)">else</span><span style="color: rgba(0, 0, 0, 1)">:
            </span><span style="color: rgba(0, 0, 255, 1)">print</span> <span style="color: rgba(128, 0, 0, 1)">"</span><span style="color: rgba(128, 0, 0, 1)">has closed </span><span style="color: rgba(128, 0, 0, 1)">"</span>
        <span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> del message_queues[s]</span>
        <span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> writable.remove(s)</span>
        <span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> print "Client %s disconnected" % (client_address)</span>

<span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> # Handle "exceptional conditions"</span>
<span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> 处理异常的情况</span>
<span style="color: rgba(0, 0, 255, 1)">for</span> s <span style="color: rgba(0, 0, 255, 1)">in</span><span style="color: rgba(0, 0, 0, 1)"> exceptional:
    </span><span style="color: rgba(0, 0, 255, 1)">print</span> (<span style="color: rgba(128, 0, 0, 1)">'</span><span style="color: rgba(128, 0, 0, 1)">exception condition on</span><span style="color: rgba(128, 0, 0, 1)">'</span><span style="color: rgba(0, 0, 0, 1)">, s.getpeername())
    </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> Stop listening for input on the connection</span>



        inputs.remove(s)
if s in outputs:
outputs.remove(s)
s.close()


    </span><span style="color: rgba(0, 128, 0, 1)">#</span><span style="color: rgba(0, 128, 0, 1)"> Remove message queue</span>
    <span style="color: rgba(0, 0, 255, 1)">del</span><span style="color: rgba(0, 0, 0, 1)"> message_queues[s]

sleep(</span>1)</pre>

客户端:

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

写代码过程中遇到了两个问题, 一是如何判断客户端已经关闭了 socket 连接, 后来自己分析了下, 如果关闭了客户端 socket, 那么此时服务器端接收到的 data 就是 '', 加个这个判断。二是如果服务器端关闭了 socket, 一旦在调用 socket 的相关方法都会报错, 不管 socket 是不是用不同的容器存储的 (意思是说 list_1 存储了 socket1, list_2 存储了 socket1, 我关闭了 socket1, 两者都不能在调用这个 socket 了)

服务器端：

客户端：