Python SocketServer：如何向多个客户端发送数据？

Question

我正在尝试用Python写一个小程序，这个程序是一个SocketServer，目的是把收到的消息发送给所有连接的客户端。不过我遇到了一些问题，我不知道怎么在服务器端存储客户端的信息，也不知道怎么同时给多个客户端发送消息。而且，每当有超过一个客户端连接时，我的程序就会出错，每当一个客户端发送超过一条消息时也会出错……

这是我到目前为止写的代码：

        print str(self.client_address[0])+' connected.'
    def handle(self):
        new=1
        for client in clients:
            if client==self.request:
                new=0
        if new==1:
            clients.append(self.request)
        for client in clients:
            data=self.request.recv(1024)
            client.send(data)

class Host:
    def __init__(self):
        self.address = ('localhost', 0)
        self.server = SocketServer.TCPServer(self.address, EchoRequestHandler)
        ip, port = self.server.server_address
        self.t = threading.Thread(target=self.server.serve_forever)
        self.t.setDaemon(True)
        self.t.start()
        print ''
        print 'Hosted with IP: '+ip+' and port: '+str(port)+'. Clients can now connect.'
        print ''
    def close(self):
        self.server.socket.close()

class Client:
    name=''
    ip=''
    port=0
    def __init__(self,ip,port,name):
        self.name=name
        self.hostIp=ip
        self.hostPort=port
        self.s=socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.s.connect((self.hostIp, self.hostPort))
    def reco(self):
        self.s=socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.s.connect((self.hostIp, self.hostPort))
    def nick(self,newName):
        self.name=newName
    def send(self,message):
        message=self.name+' : '+message
        len_sent=self.s.send(message)
        response=self.s.recv(len_sent)
        print response
        self.reco()
    def close(self):
        self.s.close()

显然，我对自己在做什么完全没有头绪，所以任何帮助都非常感谢。
提前谢谢你们！

补充一下：我在Windows Vista上使用的是Python 2.7。

Answer 1

为什么要使用SocketServer？一个简单的客户端不能满足你的需求吗？

import socket

HOST = ''
PORT = 8000
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind((HOST, PORT))
sock.listen(5)
while True:
    conn, addr = sock.accept()
    print 'connecting to', addr
    while True:
        data = conn.recv(1024)
        if not data:
            break
        conn.send(data)

Answer 2

你可以使用 socketserver 来向所有连接的客户端发送消息。不过，这个功能并不是现成的，需要通过扩展一些已经提供的类来实现。在下面的例子中，我们使用了 ThreadingTCPServer 和 StreamRequestHandler 这两个类来实现。它们提供了一个基础，但仍然需要一些修改才能达到你想要的效果。文档中会解释每个函数、类和方法的作用，帮助你完成任务。

服务器

#! /usr/bin/env python3
import argparse
import pickle
import queue
import select
import socket
import socketserver


def main():
    """Start a chat server and serve clients forever."""
    parser = argparse.ArgumentParser(description='Execute a chat server demo.')
    parser.add_argument('port', type=int, help='location where server listens')
    arguments = parser.parse_args()
    server_address = socket.gethostbyname(socket.gethostname()), arguments.port
    server = CustomServer(server_address, CustomHandler)
    server.serve_forever()


class CustomServer(socketserver.ThreadingTCPServer):

    """Provide server support for the management of connected clients."""

    def __init__(self, server_address, request_handler_class):
        """Initialize the server and keep a set of registered clients."""
        super().__init__(server_address, request_handler_class, True)
        self.clients = set()

    def add_client(self, client):
        """Register a client with the internal store of clients."""
        self.clients.add(client)

    def broadcast(self, source, data):
        """Resend data to all clients except for the data's source."""
        for client in tuple(self.clients):
            if client is not source:
                client.schedule((source.name, data))

    def remove_client(self, client):
        """Take a client off the register to disable broadcasts to it."""
        self.clients.remove(client)


class CustomHandler(socketserver.StreamRequestHandler):

    """Allow forwarding of data to all other registered clients."""

    def __init__(self, request, client_address, server):
        """Initialize the handler with a store for future date streams."""
        self.buffer = queue.Queue()
        super().__init__(request, client_address, server)

    def setup(self):
        """Register self with the clients the server has available."""
        super().setup()
        self.server.add_client(self)

    def handle(self):
        """Run a continuous message pump to broadcast all client data."""
        try:
            while True:
                self.empty_buffers()
        except (ConnectionResetError, EOFError):
            pass

    def empty_buffers(self):
        """Transfer data to other clients and write out all waiting data."""
        if self.readable:
            self.server.broadcast(self, pickle.load(self.rfile))
        while not self.buffer.empty():
            pickle.dump(self.buffer.get_nowait(), self.wfile)

    @property
    def readable(self):
        """Check if the client's connection can be read without blocking."""
        return self.connection in select.select(
            (self.connection,), (), (), 0.1)[0]

    @property
    def name(self):
        """Get the client's address to which the server is connected."""
        return self.connection.getpeername()

    def schedule(self, data):
        """Arrange for a data packet to be transmitted to the client."""
        self.buffer.put_nowait(data)

    def finish(self):
        """Remove the client's registration from the server before closing."""
        self.server.remove_client(self)
        super().finish()


if __name__ == '__main__':
    main()

---

当然，你还需要一个能够和服务器通信的客户端，并且这个客户端需要使用和服务器相同的协议。因为我们使用的是Python，所以决定利用 pickle 模块来帮助服务器和客户端之间的数据传输。也可以使用其他数据传输方法（比如JSON、XML等），但使用pickle和unpickle数据对于这个程序来说已经足够了。文档中也有相关说明，所以理解起来应该不会太难。需要注意的是，服务器的命令可能会打断用户的数据输入。

客户端

#! /usr/bin/env python3
import argparse
import cmd
import pickle
import socket
import threading


def main():
    """Connect a chat client to a server and process incoming commands."""
    parser = argparse.ArgumentParser(description='Execute a chat client demo.')
    parser.add_argument('host', type=str, help='name of server on the network')
    parser.add_argument('port', type=int, help='location where server listens')
    arguments = parser.parse_args()
    client = User(socket.create_connection((arguments.host, arguments.port)))
    client.start()


class User(cmd.Cmd, threading.Thread):

    """Provide a command interface for internal and external instructions."""

    prompt = '>>> '

    def __init__(self, connection):
        """Initialize the user interface for communicating with the server."""
        cmd.Cmd.__init__(self)
        threading.Thread.__init__(self)
        self.connection = connection
        self.reader = connection.makefile('rb', -1)
        self.writer = connection.makefile('wb', 0)
        self.handlers = dict(print=print, ping=self.ping)

    def start(self):
        """Begin execution of processor thread and user command loop."""
        super().start()
        super().cmdloop()
        self.cleanup()

    def cleanup(self):
        """Close the connection and wait for the thread to terminate."""
        self.writer.flush()
        self.connection.shutdown(socket.SHUT_RDWR)
        self.connection.close()
        self.join()

    def run(self):
        """Execute an automated message pump for client communications."""
        try:
            while True:
                self.handle_server_command()
        except (BrokenPipeError, ConnectionResetError):
            pass

    def handle_server_command(self):
        """Get an instruction from the server and execute it."""
        source, (function, args, kwargs) = pickle.load(self.reader)
        print('Host: {} Port: {}'.format(*source))
        self.handlers[function](*args, **kwargs)

    def preloop(self):
        """Announce to other clients that we are connecting."""
        self.call('print', socket.gethostname(), 'just entered.')

    def call(self, function, *args, **kwargs):
        """Arrange for a handler to be executed on all other clients."""
        assert function in self.handlers, 'You must create a handler first!'
        pickle.dump((function, args, kwargs), self.writer)

    def do_say(self, arg):
        """Causes a message to appear to all other clients."""
        self.call('print', arg)

    def do_ping(self, arg):
        """Ask all clients to report their presence here."""
        self.call('ping')

    def ping(self):
        """Broadcast to all other clients that we are present."""
        self.call('print', socket.gethostname(), 'is here.')

    def do_exit(self, arg):
        """Disconnect from the server and close the client."""
        return True

    def postloop(self):
        """Make an announcement to other clients that we are leaving."""
        self.call('print', socket.gethostname(), 'just exited.')


if __name__ == '__main__':
    main()

Answer 3

你可以看看asyncore这个东西。你在客户端调用的那些socket操作是阻塞的，也就是说它们在接收到数据或者超时之前不会返回，这样就很难同时监听从主机发送过来的消息，同时让客户端实例把要发送的数据排队。asyncore的作用就是帮你处理这种基于超时的轮询循环。

这里有一段代码示例——如果有什么不清楚的地方请告诉我：

from __future__ import print_function

import asyncore
import collections
import logging
import socket


MAX_MESSAGE_LENGTH = 1024


class RemoteClient(asyncore.dispatcher):

    """Wraps a remote client socket."""

    def __init__(self, host, socket, address):
        asyncore.dispatcher.__init__(self, socket)
        self.host = host
        self.outbox = collections.deque()

    def say(self, message):
        self.outbox.append(message)

    def handle_read(self):
        client_message = self.recv(MAX_MESSAGE_LENGTH)
        self.host.broadcast(client_message)

    def handle_write(self):
        if not self.outbox:
            return
        message = self.outbox.popleft()
        if len(message) > MAX_MESSAGE_LENGTH:
            raise ValueError('Message too long')
        self.send(message)


class Host(asyncore.dispatcher):

    log = logging.getLogger('Host')

    def __init__(self, address=('localhost', 0)):
        asyncore.dispatcher.__init__(self)
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.bind(address)
        self.listen(1)
        self.remote_clients = []

    def handle_accept(self):
        socket, addr = self.accept() # For the remote client.
        self.log.info('Accepted client at %s', addr)
        self.remote_clients.append(RemoteClient(self, socket, addr))

    def handle_read(self):
        self.log.info('Received message: %s', self.read())

    def broadcast(self, message):
        self.log.info('Broadcasting message: %s', message)
        for remote_client in self.remote_clients:
            remote_client.say(message)


class Client(asyncore.dispatcher):

    def __init__(self, host_address, name):
        asyncore.dispatcher.__init__(self)
        self.log = logging.getLogger('Client (%7s)' % name)
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.name = name
        self.log.info('Connecting to host at %s', host_address)
        self.connect(host_address)
        self.outbox = collections.deque()

    def say(self, message):
        self.outbox.append(message)
        self.log.info('Enqueued message: %s', message)

    def handle_write(self):
        if not self.outbox:
            return
        message = self.outbox.popleft()
        if len(message) > MAX_MESSAGE_LENGTH:
            raise ValueError('Message too long')
        self.send(message)

    def handle_read(self):
        message = self.recv(MAX_MESSAGE_LENGTH)
        self.log.info('Received message: %s', message)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    logging.info('Creating host')
    host = Host()
    logging.info('Creating clients')
    alice = Client(host.getsockname(), 'Alice')
    bob = Client(host.getsockname(), 'Bob')
    alice.say('Hello, everybody!')
    logging.info('Looping')
    asyncore.loop()

这段代码会产生以下输出：

INFO:root:Creating host
INFO:root:Creating clients
INFO:Client (  Alice):Connecting to host at ('127.0.0.1', 51117)
INFO:Client (    Bob):Connecting to host at ('127.0.0.1', 51117)
INFO:Client (  Alice):Enqueued message: Hello, everybody!
INFO:root:Looping
INFO:Host:Accepted client at ('127.0.0.1', 55628)
INFO:Host:Accepted client at ('127.0.0.1', 55629)
INFO:Host:Broadcasting message: Hello, everybody!
INFO:Client (  Alice):Received message: Hello, everybody!
INFO:Client (    Bob):Received message: Hello, everybody!