使用 Tornado 创建可扩展的应用程序

在快速发展的 Web 开发领域，可扩展性是开发人员构建可处理不断增加的流量和不断增长的用户群的应用程序的关键问题。Tornado 是一个 Python Web 框架和异步网络库，其设计充分考虑了可扩展性。本博文将探讨如何使用 Tornado 创建可扩展的应用程序，重点介绍其异步 I/O、非阻塞 HTTP 服务器和其他功能，这些功能使其成为构建高性能 Web 应用程序的强大工具。

是什么让 Tornado 具有可扩展性？

Tornado 能够以最小的开销处理数千个同时连接，这要归功于其非阻塞异步架构。与使用多线程或进程进行并发的传统 Web 框架不同，Tornado 的事件驱动特性使其能够高效地管理 I/O 绑定操作，而不会阻塞主线程。这对于需要实时功能的应用程序尤其有益，例如聊天应用程序、流媒体服务或处理大量请求的 API。

Tornado 可扩展性的关键概念

1. 异步 I/O：Tornado 的核心基于事件循环，这使其能够异步处理 I/O 操作。这意味着 Tornado 无需等待操作完成（例如数据库查询、HTTP 请求），而是可以继续处理其他任务，从而使应用程序响应更快。
2. 非阻塞 HTTP 服务器：Tornado 包含一个高性能、非阻塞 HTTP 服务器，可以同时管理数千个连接。此服务器是构建可扩展 Web 服务的关键组件。
3. 协程和 async/await：Tornado 完全支持 Python 的 async 和 await 语法，使编写异步代码变得更加容易。协程是可以暂停执行（使用 await）以等待 I/O 操作的函数，同时允许其他代码运行。
4. WebSocket 支持：Tornado 的 WebSocket 处理程序允许您轻松构建实时应用程序。WebSocket 在客户端和服务器之间提供持久连接，从而实现低延迟、双向通信。
5. 负载平衡：Tornado 可以在负载平衡器后面运行，这样您就可以在应用程序的多个实例之间分配流量，从而进一步增强可扩展性。

演示 1：构建可扩展的 HTTP API

在此演示中，我们将使用 Tornado 创建一个简单的 HTTP API，它可以高效地处理多个同时的请求。

import tornado.ioloop
import tornado.web
import asyncio


class MainHandler(tornado.web.RequestHandler):
    async def get(self):
        await asyncio.sleep(1)  # Simulating a time-consuming operation
        self.write("Hello, world!")


def make_app():
    return tornado.web.Application([
        (r"/", MainHandler),
    ])


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

说明：

这个简单的 Tornado 应用程序定义了一个处理 GET 请求的路由 (/)。MainHandler 使用 asyncio.sleep 模拟延迟，演示 Tornado 如何同时处理其他请求。即使有这种延迟，服务器也可以管理多个连接而不会阻塞。

测试 API：

您可以通过运行脚本并向 http://localhost:8888/ 发出并发请求来测试此 API。可以使用 curl、ab（Apache Benchmark）或 wrk 等工具来模拟高流量。

演示 2：实现 WebSockets 以进行实时通信

对于聊天或实时通知等实时应用程序，WebSockets 提供了可扩展的解决方案。以下是使用 Tornado 实现基本 WebSocket 服务器的方法。

import tornado.ioloop
import tornado.web
import tornado.websocket


class EchoWebSocket(tornado.websocket.WebSocketHandler):
    def open(self):
        print("WebSocket opened")


    def on_message(self, message):
        self.write_message(u"You said: " + message)


    def on_close(self):
        print("WebSocket closed")


def make_app():
    return tornado.web.Application([
        (r"/websocket", EchoWebSocket),
    ])


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

说明：

此 Tornado 应用程序设置了一个 WebSocket 服务器，该服务器会回显客户端发送的任何消息。EchoWebSocket 类处理 WebSocket 连接，管理打开、消息和关闭事件。

测试 WebSocket：

您可以使用 WebSocket 客户端（如 websocat）或基于浏览器的工具（如 WebSocket King）测试此 WebSocket 服务器。连接到 ws://localhost:8888/websocket 并发送消息以查看回显响应。

演示 3：使用负载均衡器和多个实例进行扩展

为了进一步扩展 Tornado 应用程序，您可以在负载均衡器后面运行服务器的多个实例。这种方法将传入流量分布到多个服务器，从而防止任何单个服务器成为瓶颈。

示例设置：

1. 运行多个 Tornado 实例ances：在不同的端口或机器上启动 Tornado 应用程序的多个实例。
2. 配置负载均衡器：使用 Nginx、HAProxy 或 AWS ELB 等负载均衡器在这些实例之间分配请求。

Nginx 配置示例：

http {
    upstream tornado_app {
        server 127.0.0.1:8888;
        server 127.0.0.1:8889;
        server 127.0.0.1:8890;
    }


    server {
        listen 80;


        location / {
            proxy_pass http://tornado_app;
        }
    }
}

此 Nginx 配置将请求分发到在端口 8888、8889 和 8890 上运行的三个 Tornado 实例。

以下是使用 Tornado 创建可扩展应用程序的其他演示，这些演示超出了前面提供的示例：

演示 4：实施速率限制以处理高流量

速率限制对于防止滥用和确保您的应用程序在重负载下保持响应至关重要。在此演示中，我们将使用 Tornado 实现基本的速率限制机制。

import tornado.ioloop
import tornado.web
import time


class RateLimitedHandler(tornado.web.RequestHandler):
    last_request_time = {}


    def get(self):
        client_ip = self.request.remote_ip
        current_time = time.time()


        if client_ip in self.last_request_time:
            time_since_last_request = current_time - self.last_request_time[client_ip]
            if time_since_last_request < 2:  # 2 seconds rate limit
                self.set_status(429)
                self.write("Too many requests. Please try again later.")
                return


        self.last_request_time[client_ip] = current_time
        self.write("Request processed successfully.")


def make_app():
    return tornado.web.Application([
        (r"/", RateLimitedHandler),
    ])


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

说明：

此示例实现了一个基本速率限制器，限制客户端每 2 秒发出的请求数不得超过一个。如果客户端超出此限制，则会返回 429 请求过多响应。

测试速率限制器：

使用 curl 等工具或 Web 浏览器向 http://localhost:8888/ 发送多个快速请求。您将观察到快速连续发出的请求会收到 429 响应。

演示 5：使用 Tornado 和 Redis 进行缓存和扩展

缓存是一种有效的方法，通过将频繁访问的数据存储在 Redis 等更快的存储介质中来减少应用程序的负载。此演示展示了如何将 Redis 与 Tornado 集成以缓存响应并提高可扩展性。

import tornado.ioloop
import tornado.web
import aioredis


class CachedHandler(tornado.web.RequestHandler):
    async def get(self):
        redis = await aioredis.create_redis_pool("redis://localhost")
        cached_response = await redis.get("cached_data")


        if cached_response:
            self.write(f"Cached response: {cached_response.decode('utf-8')}")
        else:
            response = "Expensive operation result"
            await redis.set("cached_data", response, expire=10)  # Cache for 10 seconds
            self.write(f"New response: {response}")


        redis.close()
        await redis.wait_closed()


def make_app():
    return tornado.web.Application([
        (r"/", CachedHandler),
    ])


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

说明：

此演示使用 Redis 作为缓存层。CachedHandler 检查响应是否已缓存在 Redis 中。如果是，则返回缓存的响应。否则，将生成新的响应，缓存并返回给客户端。

测试缓存：

运行脚本并向 http://localhost:8888/ 发出请求。第一个请求将生成新的响应，但 10 秒内的后续请求将返回缓存的响应。

演示 6：使用 Tornado 和任务队列进行扩展（使用 Celery）

对于在 Tornado 请求生命周期内无法处理的操作，将任务卸载到 Celery 等任务队列可以帮助保持响应性和可扩展性。在此演示中，我们将 Tornado 与 Celery 集成以处理后台任务。

安装依赖项：

pip install celery tornado

Celery 任务 (tasks.py)：

from celery import Celery


app = Celery('tasks', broker='redis://localhost:6379/0')


@app.task
def long_running_task(data):
    # Simulate a long-running task
    import time
    time.sleep(5)
    return f"Processed {data}"

import tornado.ioloop
import tornado.web
from tasks import long_running_task


class TaskHandler(tornado.web.RequestHandler):
    async def get(self):
        task = long_running_task.delay("example data")
        self.write(f"Task started: {task.id}")


def make_app():
    return tornado.web.Application([
        (r"/start-task", TaskHandler),
    ])


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

说明：

在此示例中，我们使用 Celery 定义一个长时间运行的任务，并从 Tornado 请求处理程序中触发它。 long_running_task 在 Celery 工作器中异步执行，使 Tornado 保持响应。

测试任务队列：

使用 celery -A task worker --loglevel=info 运行 Celery 工作器，然后启动 Tornado 服务器。向 http://localhost:8888/start-task 发出请求以触发任务。响应将表明任务已启动，您可以监视 Celery 工作器的输出以查看其何时完成。

演示 7：使用 Tornado 和 Docker 进行水平扩展

横向扩展涉及添加更多应用程序实例来处理增加的负载。在此演示中，我们将在 Docker 容器中设置多个 Tornado 实例并使用 Docker Compose 来管理它们。

Dockerfile：

FROM python:3.10-slim


WORKDIR /app


COPY . /app


RUN pip install tornado


CMD ["python", "app.py"]

app.py:

import tornado.ioloop
import tornado.web


class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("Hello from Tornado!")


def make_app():
    return tornado.web.Application([
        (r"/", MainHandler),
    ])


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

docker-compose.yml:

version: '3'


services:
  tornado1:
    build: .
    ports:
      - "8001:8888"


  tornado2:
    build: .
    ports:
      - "8002:8888"


  tornado3:
    build: .
    ports:
      - "8003:8888"

说明：

此设置定义了三个 Tornado 实例，它们在单独的 Docker 容器中运行，每个实例监听不同的端口。Docker Compose 管理这些实例，通过添加更多服务，可以轻松实现水平扩展。

测试设置：

使用 docker-compose up --scale tornado=3 构建并运行容器。然后，访问 http://localhost:8001/、http://localhost:8002/ 和 http://localhost:8003/ 以查看来自不同 Tornado 实例的响应。

这些演示进一步介绍了如何使用 Tornado 创建可扩展应用程序，展示了速率限制、使用 Redis 缓存、使用 Celery 卸载任务以及使用 Docker 水平扩展等技术。这些方法中的每一种都有助于确保您的 Tornado 应用程序能够处理不断增加的负载，并在需求增长时继续高效运行。

Tornado 是一个用于构建可扩展、高性能应用程序的强大框架。其异步 I/O 和非阻塞特性使其成为实时服务和处理大量同时连接的理想选择。通过将 Tornado 的功能与负载平衡和高效的资源管理相结合，您可以创建可扩展的应用程序以满足不断增长的用户群的需求。无论您是构建 API、实时通信平台还是任何其他需要高效扩展的服务，Tornado 都能为您提供成功所需的工具。