使用 input() 时，使用线程在后台接收消息时出现管道损坏错误

Question

我正在创建一个Python聊天应用程序，与我的客户端一起，我希望能够将

input()

发送到服务器，同时也从服务器获取消息，基本上接收消息，同时还能够发送input（），我已经通过使用线程接收消息“实现”了这一点，但是一旦客户端使用“！quit”（退出命令），我就会遇到管道损坏错误。

这是我制作的客户端代码：

import socket
import threading
import sys
import time

class Client:
    # data buffer size
    DATA_BUFFER_SIZE = 1024

    def __init__(self, host, port):
        # create client socket then connect to server
        self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.client.connect((host, port))
        # receive messages thread
        # daemon is true so it runs in the background
        self.receive_messages = threading.Thread(target=self.receive_messages, daemon=True)
        self.receive_messages.start()

    def close(self, send_message=True):
        if send_message:
            # send !quit to the server so it knows we are quitting the program
            self.client.send('!quit'.encode())
            # wait for the server to send us 'ack' so then we know that the server knows we are about to quit
            # receive function waits for a message from the server, so it wont continue executing code, it will wait for data to be received
            print(self.client.recv(Client.DATA_BUFFER_SIZE))
        # finally, quit
        print('closing client')
        self.client.close()
        # exit program with error code 0, meaning no error
        sys.exit(0)

    # receive messages function for receive messages thread
    # this lets the client receive new messages while also allowing the user to send messages
    def receive_messages(self):
        while True:
            time.sleep(1)
            self.client.send('ack'.encode())
            msg = self.client.recv(Client.DATA_BUFFER_SIZE)
            if msg and msg.decode() == '!ack':
                break

    # loop function
    def loop(self):
        try:
            while True:
                msg = input('message -> ')
                # send input to the server
                self.client.send(msg.encode())
                # receive message from server
                print(self.client.recv(Client.DATA_BUFFER_SIZE))
                if msg == '!quit':
                    # we've sent '!quit' already, so set send_message to false
                    self.close(send_message=False)
        except KeyboardInterrupt:
            self.close()

def main():
    # create client and loop
    client = Client('172.31.196.42', 7777)
    client.loop()

if __name__ == '__main__':
    main()

这可能没有帮助，但如果需要，这里是服务器代码：

import threading
import socket
import sys

class Server:
  # buffer size for data that should be sent around
  DATA_BUFFER_SIZE = 1024
  
  # connection class to make it easier handling multiple clients
  class Connection:
    def __init__(self, server):
      self.conn, self.addr = server.accept()
      print(f'accepted new connection from {self.addr}')
      self.user = 'Anonymous'

    def close(self):
      print(f'closing connection to user {self.user}, addr: {self.addr}')
      # close connection
      self.conn.close()
  
  def __init__(self, port):
    # get ip of host computer
    host = socket.gethostbyname(socket.gethostname())
    # create server socket
    self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    self.server.bind((host, port))
    self.server.listen()
    print(f'server binded to {host}:{port}')
    self.running = True
    # clients
    self.clients = []
    # accept connections thread
    # this thread will run in the background so the server will accept new clients while also handling current clients
    # daemon means it will run in the background, so we will set daemon to True
    self.accept_connections = threading.Thread(target=self.accept_connections, daemon=True)
    self.accept_connections.start()

  # close function
  def close(self):
    print('closing server')
    # loop through all the client connections, then close them
    for client in self.clients:
      client.close()
    # close the main server
    self.server.close()

  def close_client(self, client_id):
    # first close the connection
    self.clients[client_id].close()
    # then pop the client of the list
    self.clients.pop(client_id)

  # function that accepts connections in the background
  def accept_connections(self):
    while True:
      self.clients.append(Server.Connection(self.server))

  # broadcast function - sends a message to all clients
  def broadcast(self, msg):
    for client in self.clients:
      # we must receive data before sending data
      client.conn.recv(Server.DATA_BUFFER_SIZE)
      client.conn.send(msg)

  def loop(self):
    # try except statement to catch when the server host presses CTRL+C, then close the server
    try:
      while True:
        for client in self.clients:
          # recieve message from client
          msg = client.conn.recv(Server.DATA_BUFFER_SIZE).decode()
          print(f'received message from {client.user}: {msg}')
          client.conn.send('ack'.encode())
          if msg == '!quit':
            # the message is !quit, meaning we will now destroy the client out of the giant list of clients
            # close the client
            # self.clients.index(client) finds the id or index of the actual client
            self.close_client(self.clients.index(client))
          else:
            # broadcast the message to all clients
            self.broadcast(f'{client.user}: {msg}'.encode())
    except KeyboardInterrupt:
      # close server
      self.close()
      # exit program with code 0 meaning the program is fine
      sys.exit(0)

def main():
  # start server on port 7777 and loop
  server = Server(7777)
  server.loop()

if __name__ == '__main__':
  main()

Answer 1

当网络连接的一侧尝试写入而另一侧已关闭连接时，通常会发生

Broken Pipe

错误。当客户端向服务器发送

!quit

命令，然后服务器或客户端在正确处理或确认所有待处理消息之前关闭连接时，可能会发生这种情况。

我更改了接收消息代码如下：

def receive_messages(self):
    while True:
        try:
            time.sleep(1)
            self.client.send('ack'.encode())
            msg = self.client.recv(Client.DATA_BUFFER_SIZE)
            if msg and msg.decode() == '!ack':
                break
        except OSError:
            break #Handling the case if client socket is closed