Python SocketServer：发送给多个客户？

Question

好吧，我试图用socketserver构建一个小的python prgram，该插座服务器应该将其收到的消息发送给所有已连接的客户端。我卡住了，我不知道如何将客户存储在Serveride上，也不知道如何发送给多个客户。哦，我的程序每次都会失败，每次客户端连接，并且每次客户端发送一条消息更多...

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

        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

你想看 异步 这里。您在客户端呼叫的套接字操作正在阻止（在收到某些数据或超时之前返回），这使得很难收听从主机发送的消息，并让客户端实例求解数据发送到同时。 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!

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等），但是能够腌制和取消分类数据可以很好地满足该程序的需求。再次包含文档，因此不难弄清楚发生了什么。请注意，服务器命令可以中断用户数据输入。

客户

#! /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

为什么要使用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 4

要同时考虑多个客户，您将不得不添加 SocketServer.ForkingMixIn 或者 ThreadingMixIn.