Ben Darnell

它应该只是默认的工作。下面是一个简单的例子:

from tornado.ioloop import IOLoop
from tornado.web import RequestHandler, Application
import aiohttp

class MyHandler(RequestHandler):
    async def get(self):
        async with aiohttp.ClientSession() as session:
            async with session.get("https://www.google.com/robots.txt") as resp:
                self.write(await resp.text())

app = Application([('/', MyHandler)])
app.listen(8080)
IOLoop.current().start()

不,龙卷风不支持这一点。有一些技术 SCM_RIGHTS 将文件描述符传输到其他进程,但这将给您提供另一个进程中的原始套接字,而不是Tornado WebSocket对象(并且没有支持的方法为该套接字构造WebSocket对象)。

Tornado的推荐方法是每个CPU运行一个进程,并通过将它们放在负载均衡器后面或使用so-reuseport让它们共享流量。在Tornado中发送数据是不阻塞的;您必须确保自己的代码也是不阻塞的(使用异步接口或线程池)。

IOLoop.add_callback

generateMessageToSockets

yield [con.write_message(msg) for con in HandlerWebSocket.connections]
yield gen.moment

gen.sleep

1. 要使用pytest fixture,必须将其作为函数的参数列出:

   def test_http_client(event_loop, http_client):
   

2. AsynchHttpClient不可调用;它具有 fetch 方法:

   resp = event_loop.run_until_complete(http_client.fetch(url))
   

您的代码中发生的事情是您正在调用fixture,而不是让pytest初始化它,并通过 url 作为其 tornado_server 争论。

同时考虑使用 pytest-asyncio 或 pytest-tornado ,允许您使用 await 而不是 run_until_complete (这是在Tornado或Asyncio中使用Pytest的常用方法):

@pytest.mark.asyncio
async def test_http_client(http_client):
    url = 'http://httpbin.org/get'
    resp = await http_client.fetch(url)
    assert resp.code == 200

支持多个 Applications 对于多进程模式,必须将“高级”模式与 add_sockets 方法:

sockets1 = tornado.netutil.bind_sockets(8080)
sockets2 = tornado.netutil.bind_sockets(8081)
tornado.process.fork_processes(0)
server1 = HTTPServer(app1)
server2 = HTTPServer(app2)
server1.add_sockets(sockets1)
server2.add_sockets(sockets2)
IOLoop.current().start()

在调用 fork_processes ,然后创建服务器并向其添加套接字。

不,事件循环未被阻塞。 post() 在主体已完全加载之前不运行。要增量处理主体,请参见 @stream_request_body 装饰者。