第14章 Tornado 开发 BBS 社区系统

项目长啥样：BBS 问答社区

类似 SegmentFault / Stack Overflow：用户登录 → 提问 → 别人回答 → 提问者第一时间收到通知。今天的目标不是写完，而是把 BBS 最重要的两件事跑通：

问题 + 回答的数据模型（一对多）+ 提问 / 列表页
长轮询（Long Polling）—— Tornado 异步的最佳示范

💡 为什么用 SQLite + 简化版

原版用 MySQL + Redis 做完整 BBS（含验证码、权限、排名）3000+ 行代码——一节课写不完。今天保留架构，把存储换成 SQLite，砍掉非核心模块，只把"长轮询"这一独门技术演示透。

1 骨架

建项目目录

SHELL

$ mkdir -p /home/bbs/templates
$ cd /home/bbs
$ touch app.py db.py init_db.py
$ touch templates/index.html templates/detail.html templates/login.html

/home/bbs/ ├── app.py # Tornado 主程序：所有 Handler ├── db.py # SQLite 操作 ├── init_db.py # 一次性初始化数据库 ├── bbs.db # 数据库文件（init 后产生） └── templates/ ├── index.html # 首页（问题列表） ├── detail.html # 问题详情页（含答案 + 长轮询） └── login.html # 登录页

2 模型

用户 + 问题 + 答案

init_db.py

PYTHON /home/bbs/init_db.py

import sqlite3

conn = sqlite3.connect("bbs.db")
c = conn.cursor()

# ============ 用户 ============
c.execute("DROP TABLE IF EXISTS users")
c.execute("""
    CREATE TABLE users (
        id       INTEGER PRIMARY KEY AUTOINCREMENT,
        username TEXT UNIQUE NOT NULL,
        password TEXT NOT NULL
    )
""")

# ============ 问题 ============
c.execute("DROP TABLE IF EXISTS questions")
c.execute("""
    CREATE TABLE questions (
        id       INTEGER PRIMARY KEY AUTOINCREMENT,
        title    TEXT NOT NULL,
        content  TEXT NOT NULL,
        author   TEXT NOT NULL,
        created  DATETIME DEFAULT CURRENT_TIMESTAMP
    )
""")

# ============ 答案：外键指向问题 ============
c.execute("DROP TABLE IF EXISTS answers")
c.execute("""
    CREATE TABLE answers (
        id          INTEGER PRIMARY KEY AUTOINCREMENT,
        question_id INTEGER NOT NULL,
        content     TEXT NOT NULL,
        author      TEXT NOT NULL,
        created     DATETIME DEFAULT CURRENT_TIMESTAMP,
        FOREIGN KEY (question_id) REFERENCES questions(id)
    )
""")

# 测试数据
c.execute("INSERT INTO users (username, password) VALUES (?, ?)",
          ("alice", "123"))
c.execute("INSERT INTO users (username, password) VALUES (?, ?)",
          ("bob", "123"))

c.execute("INSERT INTO questions (title, content, author) VALUES (?, ?, ?)",
          ("Tornado 的长轮询怎么用？", "看了官方文档没看明白，求大神", "alice"))

conn.commit(); conn.close()
print("✅ 数据库初始化完成")

db.py

PYTHON /home/bbs/db.py

import sqlite3

DB = "bbs.db"


def conn():
    c = sqlite3.connect(DB)
    c.row_factory = sqlite3.Row
    return c


def find_user(username, password):
    with conn() as c:
        return c.execute(
            "SELECT * FROM users WHERE username=? AND password=?",
            (username, password)
        ).fetchone()


def list_questions():
    with conn() as c:
        return c.execute(
            "SELECT * FROM questions ORDER BY created DESC"
        ).fetchall()


def get_question(qid):
    with conn() as c:
        return c.execute(
            "SELECT * FROM questions WHERE id=?", (qid,)
        ).fetchone()


def add_question(title, content, author):
    with conn() as c:
        cur = c.execute(
            "INSERT INTO questions (title, content, author) VALUES (?, ?, ?)",
            (title, content, author)
        )
        return cur.lastrowid    # 返回新建问题的 ID


def list_answers(qid, after_id=0):
    # 关键参数 after_id：只返回 id > after_id 的答案
    # 这是长轮询的核心——客户端告诉服务器"我已经有 id≤after_id 的了"
    with conn() as c:
        return c.execute(
            "SELECT * FROM answers WHERE question_id=? AND id>? ORDER BY id",
            (qid, after_id)
        ).fetchall()


def add_answer(qid, content, author):
    with conn() as c:
        cur = c.execute(
            "INSERT INTO answers (question_id, content, author) VALUES (?, ?, ?)",
            (qid, content, author)
        )
        return cur.lastrowid

3 主功能

登录 / 提问 / 列表

新建 /home/bbs/app.py。先把基础的几个 Handler 写出来：

PYTHON /home/bbs/app.py（基础部分）

import os
import tornado.ioloop
import tornado.web
import db


# ============ 基类：判断登录 ============
# 所有要登录的 Handler 都继承它
class BaseHandler(tornado.web.RequestHandler):

    def get_current_user(self):
        # Tornado 约定：重写这个方法返回当前用户
        # self.get_secure_cookie 读取签名 cookie，防伪造
        username = self.get_secure_cookie("user")
        return username.decode() if username else None


# ============ 登录 ============
class LoginHandler(BaseHandler):

    def get(self):
        self.render("login.html", error="")

    def post(self):
        username = self.get_argument("username")
        password = self.get_argument("password")
        if db.find_user(username, password):
            # 把用户名写进签名 cookie（用 secret cookie，防篡改）
            self.set_secure_cookie("user", username)
            self.redirect("/")
        else:
            self.render("login.html", error="账号或密码错误")


class LogoutHandler(BaseHandler):
    def get(self):
        self.clear_cookie("user")
        self.redirect("/login")


# ============ 首页：问题列表 + 提问表单 ============
class IndexHandler(BaseHandler):

    @tornado.web.authenticated           # ★ 自动拦截未登录
    def get(self):
        questions = db.list_questions()
        self.render("index.html", questions=questions,
                    user=self.current_user)

    @tornado.web.authenticated
    def post(self):
        # 提问
        title = self.get_argument("title")
        content = self.get_argument("content")
        db.add_question(title, content, self.current_user)
        self.redirect("/")


# ============ 问题详情页 ============
class DetailHandler(BaseHandler):

    @tornado.web.authenticated
    def get(self, qid):
        question = db.get_question(qid)
        answers = db.list_answers(qid)
        self.render("detail.html", q=question, answers=answers,
                    user=self.current_user)

    @tornado.web.authenticated
    def post(self, qid):
        # 回答某个问题
        content = self.get_argument("content")
        db.add_answer(int(qid), content, self.current_user)
        self.redirect("/q/" + qid)


# ============ Application + 启动（先放着，下一步加路由） ============
def make_app():
    return tornado.web.Application(
        [
            (r"/",           IndexHandler),
            (r"/login",      LoginHandler),
            (r"/logout",     LogoutHandler),
            (r"/q/(\d+)",    DetailHandler),
        ],
        template_path=os.path.join(os.path.dirname(__file__), "templates"),
        cookie_secret="change-me-in-production",    # secure_cookie 必填
        login_url="/login",                          # @authenticated 重定向到这里
        debug=True,
    )


if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    print("启动：http://0.0.0.0:8888")
    tornado.ioloop.IOLoop.current().start()

💬 Tornado 的 4 个新概念

set_secure_cookie / get_secure_cookie — 签名 Cookie，防篡改（要靠 cookie_secret）
get_current_user() — 重写它告诉 Tornado "登录用户怎么取"
self.current_user — 拿当前用户（Tornado 自动调用上面那方法）
@authenticated 装饰器 — 没登录直接跳到 login_url

三个模板（关键代码块）

新建 templates/login.html：

HTML templates/login.html

<!DOCTYPE html>
<html><head><meta charset="UTF-8"><title>BBS 登录</title></head>
<body style="font-family:sans-serif; max-width:360px; margin:80px auto;">
    <h2>💬 BBS 登录</h2>
    <form method="post">
        <input name="username" placeholder="用户名"
               style="width:100%; padding:8px; margin:5px 0;">
        <input name="password" type="password" placeholder="密码"
               style="width:100%; padding:8px; margin:5px 0;">
        <button style="width:100%; padding:10px; background:#3B6B9A; color:#fff;">登录</button>
    </form>
    {% if error %}<p style="color:#B83B2E">❌ {{ error }}</p>{% end %}
    <p style="color:#888">测试账号：alice / 123 或 bob / 123</p>
</body></html>

新建 templates/index.html：

HTML templates/index.html

<!DOCTYPE html>
<html><head><meta charset="UTF-8"><title>BBS 社区</title></head>
<body style="font-family:sans-serif; max-width:720px; margin:30px auto;">

    <div style="display:flex; justify-content:space-between;">
        <h2>💬 BBS 问答社区</h2>
        <span>👤 {{ user }} | <a href="/logout">退出</a></span>
    </div>

    <!-- 提问表单 -->
    <form method="post" style="border:1px solid #ddd; padding:14px; margin: 10px 0;">
        <h3>✍️ 我要提问</h3>
        <input name="title" placeholder="标题" required
               style="width:100%; padding:6px;">
        <textarea name="content" placeholder="详细描述" required
                  style="width:100%; padding:6px; height:80px; margin-top:5px;"></textarea>
        <button style="margin-top:6px; padding:6px 14px; background:#D97B2B; color:#fff; border:none;">发布</button>
    </form>

    <!-- 问题列表 -->
    <h3>🗒️ 最新问题</h3>
    {% for q in questions %}
        <div style="padding:10px 0; border-bottom:1px solid #eee;">
            <a href="/q/{{ q['id'] }}"><strong>{{ q['title'] }}</strong></a>
            <div style="color:#888; font-size:.9em;">
                {{ q['author'] }} · {{ q['created'] }}
            </div>
        </div>
    {% end %}
</body></html>

⚠️ 注意 Tornado 模板用 {% end %}

Tornado 模板里循环结束是 {% end %}，不要写 {% endfor %}——会报错。

4 长轮询

压轴：用长轮询实现"答案即时推送"

💬 三种"实时推送"对比

方式	原理	优劣
短轮询	客户端每隔几秒发请求问"有新内容吗"	简单但浪费请求
长轮询	请求挂着，有新内容才返回	一次响应一条更新，对服务器友好
WebSocket	双向长连接，服务器主动推	最强，但代码改动多

长轮询是 WebSocket 普及前的主流方案，Tornado 异步特性让它实现起来非常自然——挂起请求不占线程，扛上万连接不费力。

长轮询的工作流程

浏览器服务器 | | |----- /q/1/poll?after=0 ----->| | | | （挂着，等新答案... ） | | | | [B 用户提交答案] | | |<----- 返回新答案 JSON -------| | | | （马上发起下一次轮询） | |----- /q/1/poll?after=5 ----->| | |

① 服务端：长轮询 Handler

在 app.py 里追加两个 Handler——一个负责"挂起等待"，一个负责"提交答案后唤醒"：

PYTHON app.py（追加长轮询部分）

import tornado.gen
import datetime
import json
from tornado.locks import Condition


# ============ 全局：每个问题对应一个 Condition（信号灯） ============
# 提交答案时通知它，让所有挂起的轮询请求醒过来
question_conditions = {}     # {qid: Condition}


def get_condition(qid):
    if qid not in question_conditions:
        question_conditions[qid] = Condition()
    return question_conditions[qid]


# ============ 长轮询 Handler：客户端定期请求这里 ============
class PollHandler(BaseHandler):

    @tornado.web.authenticated
    async def get(self, qid):
        qid = int(qid)
        after_id = int(self.get_argument("after", "0"))

        # 1. 先查一次：是否已有 after_id 之后的新答案
        new_answers = db.list_answers(qid, after_id)

        # 2. 如果还没有新答案，挂起等待 —— ★ Tornado 的核心能力
        if not new_answers:
            condition = get_condition(qid)
            try:
                # wait 在被 notify 之前一直阻塞（但是异步阻塞，不占线程！）
                # timeout=30 兜底——最多等 30 秒就返回，避免请求挂太久
                await condition.wait(timeout=datetime.timedelta(seconds=30))
            except tornado.gen.TimeoutError:
                pass     # 超时也没事，返回空就行

            # 醒了之后再查一次
            new_answers = db.list_answers(qid, after_id)

        # 3. 把答案转 JSON 返回
        result = [{"id": a["id"],
                   "content": a["content"],
                   "author": a["author"],
                   "created": a["created"]} for a in new_answers]
        self.write({"answers": result})


# ============ 重写 DetailHandler 的 post：提交答案后通知所有等待者 ============
###  把前面 DetailHandler.post 改成下面这样：
class DetailHandler(BaseHandler):

    @tornado.web.authenticated
    def get(self, qid):
        question = db.get_question(qid)
        answers = db.list_answers(int(qid))
        self.render("detail.html", q=question, answers=answers,
                    user=self.current_user)

    @tornado.web.authenticated
    def post(self, qid):
        qid = int(qid)
        content = self.get_argument("content")
        db.add_answer(qid, content, self.current_user)

        # ★ 关键：唤醒所有挂起的轮询请求
        # notify_all() 让所有 await condition.wait() 的协程立刻继续执行
        get_condition(qid).notify_all()

        self.redirect("/q/" + str(qid))


# ★ 最后别忘把新路由加到 Application 的列表里：
###     (r"/q/(\d+)/poll", PollHandler),

② 客户端：detail.html 用 JS 不停轮询

HTML templates/detail.html

<!DOCTYPE html>
<html><head><meta charset="UTF-8"><title>{{ q['title'] }}</title></head>
<body style="font-family:sans-serif; max-width:720px; margin:30px auto;">

    <a href="/">← 返回列表</a> · 👤 {{ user }}

    <h2>{{ q['title'] }}</h2>
    <p style="color:#888">{{ q['author'] }} · {{ q['created'] }}</p>
    <p>{{ q['content'] }}</p>

    <h3>💡 答案 (<span id="count">{{ len(answers) }}</span>)</h3>

    <div id="answers">
        {% for a in answers %}
            <div data-id="{{ a['id'] }}"
                 style="padding:10px; border-bottom:1px solid #eee;">
                <strong>{{ a['author'] }}</strong> ·
                <span style="color:#888">{{ a['created'] }}</span>
                <p>{{ a['content'] }}</p>
            </div>
        {% end %}
    </div>

    <!-- 回答表单 -->
    <form method="post" style="margin-top:20px;">
        <textarea name="content" placeholder="我来回答..."
                  required style="width:100%; height:80px; padding:6px;"></textarea>
        <button style="padding:8px 18px; background:#3B6B9A; color:#fff; border:none;">
            提交答案
        </button>
    </form>

    <script>
        // ★ 长轮询前端逻辑
        // 找当前已经收到的最大答案 ID（页面渲染时已经显示了一些）
        let lastId = Math.max(0,
            ...Array.from(document.querySelectorAll('#answers [data-id]'))
                    .map(d => +d.dataset.id));

        async function poll() {
            try {
                // 请求服务器：告诉它我已经有 lastId 之前的了
                // 服务器没新内容时，这个请求会挂着等几十秒
                const resp = await fetch('/q/{{ q["id"] }}/poll?after=' + lastId);
                const data = await resp.json();

                for (const a of data.answers) {
                    // 把每条新答案插到列表底部
                    const div = document.createElement('div');
                    div.dataset.id = a.id;
                    div.style = 'padding:10px; border-bottom:1px solid #eee; background: #FEF3E2;';
                    div.innerHTML = `<strong>${a.author}</strong> ·
                                     <span style="color:#888">${a.created}</span>
                                     <p>${a.content}</p>`;
                    document.getElementById('answers').appendChild(div);
                    document.getElementById('count').textContent =
                        document.querySelectorAll('#answers [data-id]').length;
                    lastId = Math.max(lastId, a.id);
                }
            } catch (e) { console.error(e); }

            // 不管成功失败，立刻发起下一轮——这就是"长轮询"
            poll();
        }

        poll();   // 页面打开就启动
    </script>
</body></html>

🎉 演示效果（开两个浏览器窗口）

窗口 A 用 alice 登录，进问题详情页

窗口 B 用 bob 登录，进同一个问题详情页 → 提交一个答案

→ 窗口 A 不刷新就能看到新答案出现（高亮黄色）。这就是长轮询的效果。

🎯 长轮询的核心：异步挂起 + 信号唤醒

挂起用 await condition.wait() —— Tornado 异步专长
唤醒用 condition.notify_all() —— 提交答案时调一下
挂起期间不占线程，一台机器扛上万长轮询连接没问题——这就是 Tornado 的护城河

📋 本节小结

BBS 的"立柱"

用户认证：set_secure_cookie + @authenticated 装饰器
问答 CRUD：questions 表 + answers 表（外键关联）
长轮询：Condition.wait + notify_all 实现"答案即时推送"

实战时还要补的

验证码（教材原版用 Pillow + Redis）
真正的密码哈希（passlib 或 bcrypt）
分页 / 标签 / 用户排名
WebSocket 替代长轮询（更高效）
站点装 Nginx 反向代理 + 静态资源

课堂练习

两组共 12 题：第一组项目搭建，第二组长轮询原理。

第一组

BBS 项目搭建（6 题）

Q1单选

Tornado Handler 想拦截未登录用户，最便捷的写法是？

A每个 Handler 里手写 if

B装饰器 @tornado.web.authenticated

C中间件

D抛 Exception

解析：装饰器自动调 get_current_user()，没值就重定向到 Application 的 login_url。

Q2单选

set_secure_cookie 与普通 set_cookie 的区别？

A速度更快

B带签名，防止用户篡改 cookie 内容

C会加密用户名

D没区别

解析：secure cookie 用 cookie_secret 做 HMAC 签名——服务器能识破任何被改过的 cookie。这跟"加密"不一样：内容是可读的，只是无法伪造。

Q3单选

Application 配置里 cookie_secret 的作用是？

A给 cookie 加密

Bsecure cookie 签名用的密钥

C用户密码

Dsession ID

解析：没设它，set_secure_cookie 会报错。生产环境必须用足够复杂的随机串，泄漏等于所有 secure cookie 都失效。

Q4单选

为什么 answers 表要有 question_id 字段？

A装饰用

B排序用

C外键，关联到对应的问题（一对多）

D缓存 ID

解析："一个问题有多条答案"——外键定义在"多"的一方（answers）。SQL 查"某问题的所有答案"靠 WHERE question_id=?。

Q5判断

Tornado 模板里循环结束写 {% endfor %} 比 {% end %} 更明确，应该用前者。

✓对

✕错

解析：反了。Tornado 只认 {% end %}——所有逻辑块共用。Django 才用 {% endfor %}。混用会报模板错误。

Q6判断

tornado.web.authenticated 装饰器只对 GET 请求有效，POST 请求需要单独处理。

✓对

✕错

解析：装饰器加在哪个方法（get / post / put / delete）就对哪个生效——和 HTTP 方法无关。每个要鉴权的方法都得加。

第二组

长轮询原理（6 题）

Q1单选

"长轮询"和"短轮询"最关键的区别是？

A请求更长

B使用 WebSocket

C服务器在没新数据时挂起请求等待，有新数据才返回

D客户端 sleep 久一点再发请求

解析：短轮询是"不停问"，长轮询是"问一次但服务器不急着回"。结果就是大幅减少无效请求。

Q2单选

本章长轮询用 tornado.locks.Condition 干什么？

A建立 WebSocket 连接

B挂起 / 唤醒等待中的协程（信号灯）

C给数据库加锁

D验证用户

解析：Condition 是协程间的"信号灯"——一边 await 等，一边在某个时机 notify_all 通知。这是异步编程的经典模式。

Q3单选

提交新答案后调用 condition.notify_all()，会发生什么？

A关闭所有 cookie

B所有正在 await condition.wait() 的协程立刻继续执行

C清空数据库

D发送邮件

解析：这就是"唤醒"动作。所有挂起等待的轮询请求被叫醒 → 重新查数据库 → 把新答案返回给浏览器。

Q4单选

轮询请求里的 ?after=5 参数有什么用？

A排序

B计数

C告诉服务器"我已有 id≤5 的答案，只要给我新的"

D分页

解析：这是长轮询的关键设计——客户端记住"目前为止的最大 ID"，下次只问之后的。避免重复传输已经有的数据。

Q5单选

为什么前端一收到响应就立刻发下一次 poll()？

A规则要求

BHTTP 是请求-响应模型，连接断了就要重连——这样保持持续等待

C计数

D避免内存泄漏

解析：HTTP 不像 WebSocket 能保持双向连接——每次请求都是一次性的。要"持续等待"就只能"请求-断开-再请求"。

Q6判断

如果同一台服务器上有 1000 个用户在长轮询，意味着 1000 个线程被一直占着等待。

✓对

✕错

解析：反了——这正是 Tornado 异步的优势。await condition.wait() 是异步挂起，连接挂着但不占线程。1000 个长轮询请求可能只占一两个线程。换 Flask 同步框架则真要 1000 个线程，会撑爆机器。