python - 如何使用非阻塞套接字

Question

我正在尝试编写非阻塞服务器/客户端脚本。

首先，这是我的代码：

服务器.py ->

import socket
import select
import threading

class ChatServer(threading.Thread):
    """
    SERVER THREAD
    """

    MAX_WAITING_CONNECTION = 10
    RECV_HEADER_LENGTH = 10

    def __init__(self, host, port):
        """
        Initialize new ChatServer

        :param host: Binding Host IP
        :param port: Binding Port number
        """


        threading.Thread.__init__(self)
        self.host = host
        self.port = port
        self.connections = [] ## Will keep active client connections.
        self.clients = {}
        self.running = True

    def _bind_socket(self):
        """
        Creates the server socket and binds it to the given host and port
        """
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.server_socket.bind((self.host, self.port))
        self.server_socket.listen(self.MAX_WAITING_CONNECTION)
        self.connections.append(self.server_socket)

    def _send(self, sock, client_message):
        """
        Prefixes each message with a 4-byte length before sending.

        :param sock: the incoming sock
        :param msg: the massage to send
        """
        user = self.clients[sock]
        client_message = user['header'] + user['data'] + client_message['header'] + client_message['data']
        sock.send(client_message)

    def _receive(self, sock):
        try:
            ## Bytes type header
            message_header = sock.recv(self.RECV_HEADER_LENGTH)

            if not len(message_header):
                return False

            message_length = int(message_header.decode('utf-8').strip())
            ## Bytes type data
            return {"header": message_header, "data": sock.recv(message_length)}
        except Exception as e:
            print('exception occur')
            print(e)
            return False

    def _broadcast(self, sending_client_socket, client_message):
        """
        Breadcasts a message to all the clients different from both the server itself and
        the client sending the message.
        :param client_socket: the socket of the client sending the message
        :param client_message: the message to broadcast ({'header': <bytes header>, 'data': <bytes data message>})
        """

        for sock in self.clients:
            is_not_the_server = sock != self.server_socket
            is_not_the_client_sending = sock != sending_client_socket ## sending client socket

            if is_not_the_server and is_not_the_client_sending:
                try:
                    user = self.clients[sending_client_socket]
                    print(f"Type client_message: {type(client_message)}")
                    print(f"Type user: {type(user)}")
                    sending_message = user['header'] + user['data'] + client_message['header'] + client_message['data']
                    sock.send(sending_message)
                except socket.error:
                    ## handles a possible disconnection of client "sock" by ..
                    sock.close()
                    self.connections.remove(sock) ## removing sock form active connections.
                    del self.clients[sock]

    def _log(self, sock, message):
        user = self.clients[sock]
        print(f"Received message from {user['data'].decode('utf-8')}: {message['data'].decode('utf-8')}")

    def _run(self):
        """
        Actually runs the server.
        """
        while self.running:
            ## Get the list of sockets which are ready to be read through select non-blocking calls
            ## The select has a timeout of 60 seconds

            try:
                ready_to_read, ready_to_write, in_error = select.select(self.connections, [], self.connections)
            except socket.error as e:
                print(f"General Error: {e}")
                continue
            else:
                for sock in ready_to_read:
                    ## if socket is server socket.
                    if sock == self.server_socket:
                        try:
                            client_socket, client_address = self.server_socket.accept()
                        except socket.error as e:
                            print(f"General Error: {e}")
                            break
                        else:
                            user = self._receive(client_socket)
                            if user is False:
                                continue
                            self.connections.append(client_socket)
                            self.clients[client_socket] = user
                            print(f"Accepted new connection from {client_address[0]}:{client_address[1]}..")
                    else:
                        message = self._receive(sock) ## Get client message
                        if message is False:
                            print(f"Closed connection from {self.clients[sock]['data'].decode('utf-8')}")
                            self.connections.remove(sock)
                            del self.clients[sock]
                            continue

                        self._log(sock, message)
                        print(message)
                        self._broadcast(sock, message)

                for sock in in_error:
                    self.connections.remove(sock)
                    del self.clients[sock]
        self.stop()

    def run(self):
        """
        Given a host and a port, binds the socket and runs the server.
        """
        self._bind_socket()
        self._run()

    def stop():
        """
        Stops the server by setting the "running" flag before closing
        the socket connection.
        """
        self.running = False
        self.server_socket.close()


if __name__ == '__main__':

    _HOST = '127.0.0.1'
    _PORT = 6667

    chat_server = ChatServer(_HOST, _PORT)
    chat_server.start()
    chat_server.join()

还有我的 client.py ->

import socket
import select
import errno
import threading
import sys

RECV_HEADER_LENGTH = 10

class ChatClient(threading.Thread):


    def __init__(self, host, port):
        """
        Initialize new ChatClient

        :param host: Connect Host IP
        :param port: Connect Port number
        """

        threading.Thread.__init__(self)
        self.host = host
        self.port = port
        self.username = input("Username: ")
        self.running = True

    def _send(self, sock, message):
        sock.send(message.encode('utf-8'))  

    def _connect(self):
        """
        Connecting to the ChatServer
        """
        self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.client_socket.connect((self.host, self.port))
        self.client_socket.setblocking(0)
        self.username_header = f"{len(self.username):<{RECV_HEADER_LENGTH}}"
        self._send(self.client_socket, self.username_header+self.username)

    def prompt(self) :
        sys.stdout.write(f"#{self.username}$ ")
        sys.stdout.flush()

    def _run(self):
        """
        Actually run client.
        """
        while self.running:
            reading_sockets, writing_sockets, exceptional_sockets = select.select([self.client_socket], [self.client_socket], [])
            for sock in reading_sockets:
                if sock == self.client_socket:
                    username_header = self.client_socket.recv(RECV_HEADER_LENGTH)
                    if not len(username_header):
                        print("Connection closed by the server.")
                        sys.exit()
                    username_length = int(username_header.decode("utf-8").strip())
                    username = self.client_socket.recv(username_length).decode("utf-8")
                    message_header = self.client_socket.recv(RECV_HEADER_LENGTH)
                    message_length = int(message_header.decode('utf-8').strip())
                    message = self.client_socket.recv(message_length).decode('utf-8')
                    print(f"#{username}$ {message}")
            for sock in writing_sockets:
                self.prompt()
                message = input()
                print(len(message))
                if not message:
                    continue
                message_header = f"{len(message):<{RECV_HEADER_LENGTH}}"
                self._send(sock, message_header+message)
        self.stop()

    def run(self):
        """
        Given a host and a port, binds the socket and runs the server.
        """
        self._connect()
        self._run()

    def stop():
        """
        Stops the server by setting the "running" flag before closing
        the socket connection.
        """
        self.running = False
        self.client_socket.close()


if __name__ == '__main__':

    _HOST = '127.0.0.1'
    _PORT = 6667

    chat_server = ChatClient(_HOST, _PORT)
    chat_server.start()
    chat_server.join()

现在我的问题出在我认为的 client.py 上。在 _run 函数中，我将 select reading_socket 和 writing_socket 用于同一个套接字。

当我运行此代码时，会阻塞 read_socket 的循环。因为在 for 循环 for writing_sockets 中保留了我的外壳，并且永远不会释放，甚至会出现另一次按摩。所以我想等待用户输入，但同时阅读其他消息并在 shell 上打印。我正在使用python3.7。我怎样才能做到这一点？

标签： pythonpython-3.xsocketsblockingnonblocking

Answer 1

所以我想等待用户输入，但同时阅读其他消息并在 shell 上打印。我正在使用python3.7。我怎样才能做到这一点？

通过确保仅在实际准备好用户输入sys.stdin时才读取；sys.stdin这样你的input()电话就不会阻塞。您可以通过将sys.stdin第一个参数中的一个套接字传递给select(). （注意：这在 Windows 下不起作用，因为 Microsoft 明智地决定他们的实现select()不支持 selection-on- stdin。在 Windows 下，您将不得不使用单独的线程来执行阻塞读取stdin，以及 inter -某种线程消息传递，以将数据从stdin回读取到网络线程，并且开始工作是一个巨大的痛苦）

以下是我如何修改您的_run(self)方法以获得您想要的行为（在 MacOS/X 下测试）：

def _run(self):
    """
    Actually run client.
    """
    while self.running:
        reading_sockets, writing_sockets, exceptional_sockets = select.select([self.client_socket, sys.stdin], [], [])
        for sock in reading_sockets:
            if sock == self.client_socket:
                username_header = self.client_socket.recv(RECV_HEADER_LENGTH)
                if not len(username_header):
                    print("Connection closed by the server.")
                    sys.exit()
                username_length = int(username_header.decode("utf-8").strip())
                username = self.client_socket.recv(username_length).decode("utf-8")
                message_header = self.client_socket.recv(RECV_HEADER_LENGTH)
                message_length = int(message_header.decode('utf-8').strip())
                message = self.client_socket.recv(message_length).decode('utf-8')
                print(f"#{username}$ {message}")
            elif sock == sys.stdin:
               self.prompt()
               message = input()
               print(len(message))
               if not message:
                   continue
               message_header = f"{len(message):<{RECV_HEADER_LENGTH}}"
               self._send(self.client_socket, message_header+message)
    self.stop()

请注意，我添加到sys.stdin调用的 read-sockets 参数（select()这样select()当它们的数据准备好从只要有缓冲区空间来接受更多传出数据，也就是说，它几乎总是会立即返回，这将旋转您的事件循环并导致您的客户端程序使用接近 100% 的核心，即不是你想要的）。stdinself.client_socketselect()self.client_socket

我还将您的read-from-stdin代码修改为仅在 readable-socket 为时调用sys.stdin，因为除非它有数据可以提供给您，否则尝试从 stdin 读取是没有意义的；最后，我让您self._send()在 TCP 套接字上发送呼叫，而不是尝试将字节发送回stdin（因为 stdin 是只读/输入的，因此向它发送字节没有任何意义）。