龙卷风TCP服务器/客户端进程通信(Tornado TCP Server / Client proc

我想设置通信之间的数字龙卷风的处理每个作为网络服务器，即，使用作用tornado.web.RequestHandler 。我们的想法是，我想的过程之间的全网状网络。我有4个过程，我想使用它们之间建立一个持续的永久通信tornado.tcpserver和tornado.tcpclient ：

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

然而，我在我的龙卷风文档中看到的例子是新来的TCP编程： http://www.tornadoweb.org/en/stable/iostream.html在实现类tornado.iostream.IOStream一旦插座建立所有的通信完成，然后再关闭套接字。的例子中，通过块驱动代码与回调各自执行其通信的职责。

但是是有可能打开一个TCP连接，并有BaseIOStream.read_until_close()闲置，并呼吁只有当客户端写入服务器？

换句话说，客户端和服务器保持连接，当客户端写入服务器它在某种程度上中断龙卷风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位整数。

你问，“当客户端写入到它在某种程度上中断龙卷风IOLoop来调用read（）的服务器？” 这是龙卷风的IOLoop是，这就是我们所说的“异步”：许多龙卷风协同程序或回调可以等待网络事件，而当他们正在等待事件发生IOLoop唤醒他们。这是发生了什么，无论你在代码中看到“产量”上面。