Tornado TCP 服务器/客户端进程通信

Question

我想在多个 Tornado 进程之间建立通信，每个进程都充当 Web 服务器，即使用tornado.web.RequestHandler。这个想法是我想要在进程之间建立一个完全网状的网络。我有 4 个进程，我想使用 tornado.tcpserver 和 tornado.tcpclient 在它们之间建立持续的永久通信：

T1---T2
| \  /| 
|  \/ |
| / \ |
T3---T4

我是 TCP 编程的新手，但是在我在 tornado 文档中看到的示例中：http://www.tornadoweb.org/en/stable/iostream.html 在tornado.iostream.IOStream 类的实现下，一旦建立了套接字，所有的通信就完成了，然后套接字就关闭了。该示例通过带有回调的块驱动代码，每个回调都执行其通信职责。

但是，是否可以打开 TCP 连接并让 BaseIOStream.read_until_close() 空闲并仅在客户端写入服务器时调用？

换句话说，客户端和服务器保持连接，当客户端写入服务器时，它会以某种方式中断 Tornado IOLoop 以调用 read()？

或者我的想法被误导了，这样做的方法是每次我需要进程进行通信时我建立一个新的 TCP 连接，完成工作然后终止连接？似乎每次建立这个新连接都会包含很多开销，而不是让连接保持打开状态......

Answer 1

这是一个基本的实现。 （我不能保证它的生产质量！）将它保存到一个文件并执行类似这样的操作，每个都在不同的终端窗口中：

> python myscript.py 10001 10002 10003
> python myscript.py 10002 10003 10001
> python myscript.py 10003 10001 10002

第一个参数是监听端口，其余参数是其他服务器的端口。

import argparse
import logging
import os
import random
import socket
import struct

from tornado import gen
from tornado.ioloop import IOLoop
from tornado.iostream import IOStream, StreamClosedError
from tornado.tcpclient import TCPClient
from tornado.tcpserver import TCPServer
from tornado.options import options as tornado_options


parser = argparse.ArgumentParser()
parser.add_argument("port", type=int, help="port to listen on")
parser.add_argument("peers", type=int, nargs="+", help="peers' ports")
opts = parser.parse_args()

# This is just to configure Tornado logging.
tornado_options.parse_command_line()
logger = logging.getLogger(os.path.basename(__file__))
logger.setLevel(logging.INFO)

# Cache this struct definition; important optimization.
int_struct = struct.Struct("<i")
_UNPACK_INT = int_struct.unpack
_PACK_INT = int_struct.pack

tcp_client = TCPClient()


@gen.coroutine
def client(port):
    while True:
        try:
            stream = yield tcp_client.connect('localhost', port)
            logging.info("Connected to %d", port)

            # Set TCP_NODELAY / disable Nagle's Algorithm.
            stream.set_nodelay(True)

            while True:
                msg = ("Hello from port %d" % opts.port).encode()
                length = _PACK_INT(len(msg))
                yield stream.write(length + msg)
                yield gen.sleep(random.random() * 10)

        except StreamClosedError as exc:
            logger.error("Error connecting to %d: %s", port, exc)
            yield gen.sleep(5)


loop = IOLoop.current()

for peer in opts.peers:
    loop.spawn_callback(client, peer)


class MyServer(TCPServer):
    @gen.coroutine
    def handle_stream(self, stream, address):
        logging.info("Connection from peer")
        try:
            while True:
                # Read 4 bytes.
                header = yield stream.read_bytes(4)

                # Convert from network order to int.
                length = _UNPACK_INT(header)[0]

                msg = yield stream.read_bytes(length)
                logger.info('"%s"' % msg.decode())

                del msg  # Dereference msg in case it's big.

        except StreamClosedError:
            logger.error("%s disconnected", address)


server = MyServer()
server.listen(opts.port)

loop.start()

请注意，我们不调用 read_until_close，所以我们需要一些方法来知道消息何时被完全接收。我在每条消息的开头使用一个 32 位整数来执行此操作，该整数对消息其余部分的长度进行编码。

您问，“当客户端写入服务器时，它会以某种方式中断 Tornado IOLoop 以调用 read()？”这就是 Tornado 的 IOLoop 的用途，也就是我们所说的“异步”：许多 Tornado 协程或回调可以等待网络事件，当它们等待的事件发生时，IOLoop 会唤醒它们。这就是您在上面的代码中看到“yield”时发生的情况。

Answer 2

但是，是否可以打开 TCP 连接并让 BaseIOStream.read_until_close() 空闲并仅在客户端写入服务器时调用？

不知道龙卷风。但是，就 TCP 而言，一旦建立连接（服务器和客户端保持状态为“已建立”），服务器和客户端可以交换数据，直到有人希望关闭连接或发生网络问题导致消息发送不要到达另一端。

换句话说，客户端和服务器保持连接，当客户端写入服务器时，它会以某种方式中断 Tornado IOLoop 以调用 read()？

是的。应该是这样。

或者我的想法被误导了，这样做的方法是每次我需要进程进行通信时，我建立一个新的 TCP 连接，完成工作然后终止连接？

没有。每次数据交换都不需要重新发起 TCP 连接