python在同一连接上套接字多条消息

Question

我的问题更笼统而不是具体。我想实现一个简单的客户机-服务器应用程序，只需将消息从客户机传递到服务器并从服务器获取确认。

我想知道在使用sockets时需要考虑什么，我需要实现自己的通信接口并在同一个连接上管理消息传递，还是为每个消息创建一个新的连接？

（请假设当前的消息小于缓冲区大小）

代码如下：

服务器.py

server_info = (HOST, PORT)
sock = socket.socket(family=AF_INET, type=SOCK_STREAM)
sock.bind(server_info)
sock.listen(NUMBER_OF_SOCKETS)
try:
    while True:
        connection, client_address = sock.accept()
        try:
            while True:
                data = connection.recv(BUFFER_SIZE)
                print('message received: {data}'.format(data=data))
                connection.send("ok")
        finally:
            connection.close()

客户端.py

server_info = (HOST, PORT)
sock = socket.socket(family=AF_INET, type=SOCK_STREAM)
sock.connect(server_info)
try:
    print("connection established")
    while True:
        print("Please enter a message you want to pass to the server")
        msg = raw_input()

        print('sending "{message}"'.format(message=msg))
        sock.send(msg)

        while True:
            data = sock.recv(constants.BUFFER_SIZE)
            print('received "{data}"'.format(data=data))
            break

finally:
    print('closing socket')
    sock.close()

这段代码使我能够在服务器端接收多条消息，并从客户端发送多条消息。这是正确的方法吗？为此，我必须在客户端创建两个无限循环，关闭连接如何？当我发送0字节的消息时，服务器和客户端都会卡住。

非常感谢！

Answer 1

在双向通信中，默认情况下，客户端可以知道何时完成发送，但无法知道是否完成接收。而且，服务器也不知道客户端是否已经完成发送。

代码：

def recv_end(the_socket):
    End='SERVER WRONG MARKER'
    total_data=[];data='';got_end=False
    while True:
            data=the_socket.recv(8192)
            if not data: break
            if End in data:
                total_data.append(data[:data.find(End)])
                got_end=True
                break
            total_data.append(data)
            if len(total_data)>1:
                #check if end_of_data was split
                last_pair=total_data[-2]+total_data[-1]
                if End in last_pair:
                    total_data[-2]=last_pair[:last_pair.find(End)]
                    total_data.pop()
                    got_end=True
                    break
    return (got_end,''.join(total_data))

def basic_server(sock):
    got=[]
    got_end,data = recv_end(sock)
    if not got_end:  
        sock.send('ERROR:no end!') #<--- not possible w/close()
    else: sock.sendall(data*2)
    sock.shutdown(1)
    sock.close()

import socket
Port=4444
def start_server():
    sock=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    sock.bind(('',Port))
    sock.listen(5)
    print 'started on',Port
    while True:
        newsock,address=sock.accept()
        basic_server(newsock)

def send_data(data):
    sock=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    sock.connect(('localhost',Port))
    print 'connected'
    sock.sendall(data+'CLIENT WRONG MARKER')
    print 'sent',data
    sock.shutdown(1)
    print 'shutdown'
    result=[]
    while True:
       got=sock.recv(2)
       if not got: break
       result.append(got)
    sock.close()
    return ''.join(result)

if __name__=='__main__':
    start_server()

您可以在数据前面放置一个字节计数，或者有一个结束标记，这样服务器就可以知道它是否得到了所有字节。

然而，这带来了一个问题。如果字节计数错误或结束标记从未到达怎么办？使用socket.close()时，服务器无法告诉客户端“奇怪。你已经把数据发送给我了，但是我没有得到所有的数据”，因为客户端发送完后，客户端连接没有保持打开状态。

使用socket.shutdown(1）服务器仍然可以告诉客户端有问题，并采取适当的措施。

shutdown命令有三个选项：

在上面的代码中，重点放在1上，以消除关闭操作中的隐式发送。请注意，在send_data中，close操作与关机的距离（相对而言）较远。这允许服务器告诉客户端任何临别赠言。

只需运行代码启动服务器。出于演示目的，服务器一次设置为recv only2 bytes（应该是8192）。要向它发送数据，请导入它（调用shut_srv或其他名称），并为客户端调用send_data。

data=('a1234','b1234','c1234','d1234','e1234') for d in data: print shut_srv.send_data(d)

你会得到这样的反应：connected sent a1234 shutdown ERROR:no end! connected sent b1234 shutdown ERROR:no end! connected sent c1234 shutdown ERROR:no end! connected sent d1234 shutdown ERROR:no end! connected sent e1234 shutdown ERROR:no end!

如果你做的记号一样。响应应该是：connected sent a123456789 shutdown a1234a1234 connected sent b1234 shutdown b1234b1234 connected sent c1234 shutdown c1234c1234 connected sent d1234 shutdown d1234d1234 connected sent e1234 shutdown e1234e1234

Answer 2

添加两种类型的服务器客户端一种是多进程的，另一种是异步的，它们做的事情几乎相同，异步的更健壮，请阅读此处的原因： Threads vs. Async。

我的例子： 使用多进程：

import multiprocessing
import socket
import time

HOST = "0.0.0.0"
PORT = 9000


def handle(connection, address):

    try:
        while True:
            data = connection.recv(1024)
            connection.sendall(data + ' server time {}'.format(time.time()))
    except:
        pass
    finally:
        connection.close()


class Server(object):

    def __init__(self, hostname, port):
        self.hostname = hostname
        self.port = port

    def start(self):
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.bind((self.hostname, self.port))
        self.socket.listen(1)

        while True:
            conn, address = self.socket.accept()
            process = multiprocessing.Process(
                target=handle, args=(conn, address))
            process.daemon = True
            process.start()


if __name__ == "__main__":
    server = Server(HOST, PORT)
    try:
        print 'start'
        server.start()
    except:
        print 'something wrong happened, a keyboard break ?'
    finally:
        for process in multiprocessing.active_children():
            process.terminate()
            process.join()
    print 'Goodbye'

委托人：

    import sys
import threading
import time
import socket

SOCKET_AMOUNT = 100
HOST = "localhost"
PORT = 9000


def myclient(ip, port, message):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((ip, port))
    sock.sendall(message)
    result = sock.recv(1024)
    print result + ' final clnt time {}'.format(time.time())
    sock.close()

if __name__ == "__main__":
    thread_list = []
    for i in range(SOCKET_AMOUNT):
        msg = "Thread #{}, clnt time {}".format(i, time.time())
        client_thread = threading.Thread(
            target=myclient, args=(HOST, PORT, msg))
        thread_list.append(client_thread)
        client_thread.start()

    waiting = time.time()
    [x.join() for x in thread_list]
    done = time.time()
    print 'DONE {}. Waiting for {} seconds'.format(done, done-waiting)

下一个服务器要健壮得多！！！数据不会丢失！！！ 服务器：

import asyncore
import socket
import time
import logging
import json


class Server(asyncore.dispatcher):

    def __init__(self, host, port):

        self.logger = logging.getLogger('SERVER')
        asyncore.dispatcher.__init__(self)
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.set_reuse_addr()
        self.bind(('', port))
        self.listen(confjson.get('SERVER_QUEUE_SIZE', None))
        self.logger.debug('binding to {}'.format(self.socket.getsockname()))

    def handle_accept(self):
        socket, address = self.accept()
        self.logger.debug('new connection accepted')
        EchoHandler(socket)


class EchoHandler(asyncore.dispatcher_with_send):

    def handle_read(self):

        msg = self.recv(confjson.get('RATE', None))
        self.out_buffer = msg
        self.out_buffer += ' server recieve: {}'.format(time.time())
        if not self.out_buffer:
            self.close()


if __name__ == "__main__":

    logging.basicConfig(level=logging.DEBUG,
                        format='%(name)s: %(message)s',
                        )
    with open('config.json', 'r') as jfile:
        confjson = json.load(jfile)
    try:
        logging.debug('Server start')
        server = Server(confjson.get('HOST', None),
                        confjson.get('PORT', None))
        asyncore.loop()
    except:
        logging.error('Something happened,\n'
                      'if it was not a keyboard break...\n'
                      'check if address taken, '
                      'or another instance is running. Exit')
    finally:
        logging.debug('Goodbye')

异步客户端：

import asyncore
import socket
import time
import logging
import json


class Client(asyncore.dispatcher_with_send):

    def __init__(self, host, port, message, pk):
        self.logger = logging.getLogger('CLIENT')
        asyncore.dispatcher.__init__(self)
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.host = host
        self.port = port
        self.connect((host, port))
        self.out_buffer = message
        self.clientID = pk
        self.logger.debug('Connected #{}'.format(self.clientID))

    def handle_close(self):
        self.close()

    def handle_read(self):
        rec_msg = self.recv(confjson.get('RATE', None))
        self.logger.debug('#{}, {} back at client {}'.format(self.clientID,
                                                             rec_msg,
                                                             time.time()
                                                             )
                          )
        self.close()


if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG,
                        format='%(name)s: %(message)s',
                        )

    with open('config.json', 'r') as jfile:
        confjson = json.load(jfile)
    clients = []
    for idx in range(confjson.get('SOCKET_AMOUNT', None)):
        msg = "Start: {}".format(time.time())
        clients.append(Client(confjson.get('HOST', None),
                              confjson.get('PORT', None),
                              msg,
                              idx)
                       )
    start = time.time()
    logging.debug(
        'Starting async loop for all connections, unix time {}'.format(start))
    asyncore.loop()
    logging.debug('{}'.format(time.time() - start))

还有一个小配置文件：

{
    "HOST": "127.0.0.1",
    "PORT": 5007,
    "RATE": 8096,
    "SERVER_QUEUE_SIZE": 16,
    "SOCKET_AMOUNT": 100
}

Answer 3

我也遇到了同样的问题，请尝试以下服务器：

import socket
import select
open_client_sockets = []
server_socket = socket.socket()
server_socket.bind(('0.0.0.0', 8001))

server_socket.listen(1)
(new_socket1, address1) = server_socket.accept()
open_client_sockets.append(new_socket1)

while True:
    rlist, wlist, xlist = select.select(open_client_sockets, open_client_sockets, [])
    for current_socket in rlist:
        data = current_socket.recv(4096)
        if data != '':
            print "given data: ", str(data)