{}JSJSON

构建生产级 Telegram Bot：Node.js + TypeScript + grammY 完全指南

深入讲解如何用 Node.js 和 TypeScript 构建生产级 Telegram Bot，涵盖 grammY 框架、Webhook 部署、中间件架构、数据库集成、支付接入与性能优化，附完整可运行代码。

前端开发 2026-06-10 20 分钟

Telegram Bot API 是全球最开放的即时通讯机器人平台——零审核费用、无消息条数限制、支持 Webhook 长连接，月活跃 Bot 超过 300 万个。根据 Telegram 官方 2025 年数据，Bot 平台每天处理超过 50 亿条消息，开发者生态增长率连续三年超过 40%。如果你正在构建通知系统、客服机器人、内部工具或 SaaS 产品的消息通道，Telegram Bot 是性价比最高的选择——不需要企业认证、不需要审核周期，注册 Bot Token 五分钟就能开始开发。

本文不是「Hello World」级别的入门教程，而是基于真实生产项目的经验总结——从框架选型、架构设计到部署运维，覆盖你在生产环境中会遇到的每一个工程问题。

📌 记住： Telegram Bot API 的更新频率很高（几乎每月都有新特性），本文基于 Bot API 7.x 版本。如果你在阅读时发现 API 行为有变化，请以 Telegram Bot API 官方文档 为准。

🔧 一、框架选型与项目架构

1.1 为什么选择 grammY

Telegram Bot 的 Node.js 生态有三个主流框架：node-telegram-bot-api（最老牌）、telegraf（最流行）和 grammY（最新）。三者的核心差异在于类型安全性和中间件架构：

维度 node-telegram-bot-api Telegraf grammY
TypeScript 支持 ⚠️ 社区类型，不完整 ✅ 内置，但泛型较弱 ✅ 原生 TS，泛型完整
中间件架构 ❌ 事件回调，无中间件 ✅ Koa 风格中间件 ✅ Koa 风格，更强类型
插件生态 少 中等 丰富（官方维护）
Webhook 支持 手动集成 Express 内置 内置 + 多框架适配
活跃维护 ⚠️ 更新缓慢 ✅ 活跃 ✅ 非常活跃
npm 周下载量 ~50 万 ~25 万 ~15 万（增速最快）

⚡ 关键结论： 如果你的项目使用 TypeScript，grammY 是最佳选择。它的类型系统是三个框架中最完善的——每个中间件的上下文类型都能正确传播，IDE 自动补全覆盖所有 Bot API 方法，编译时就能发现 API 调用错误。

1.2 项目初始化与目录结构

# 初始化项目
mkdir my-telegram-bot && cd my-telegram-bot
npm init -y
npm install grammY drizzle-orm better-sqlite3
npm install -D typescript @types/better-sqlite3 drizzle-kit tsx

# tsconfig.json
npx tsc --init --target ES2022 --module NodeNext --moduleResolution NodeNext

生产级 Telegram Bot 推荐以下目录结构：

my-telegram-bot/
├── src/
│   ├── bot.ts              # Bot 实例创建与配置
│   ├── index.ts            # 入口文件（启动 Bot）
│   ├── middleware/
│   │   ├── auth.ts         # 权限验证中间件
│   │   ├── logger.ts       # 请求日志中间件
│   │   └── rateLimit.ts    # 限流中间件
│   ├── handlers/
│   │   ├── start.ts        # /start 命令
│   │   ├── help.ts         # /help 命令
│   │   └── settings.ts     # 设置菜单
│   ├── scenes/
│   │   └── feedback.ts     # 多步对话场景
│   ├── db/
│   │   ├── schema.ts       # 数据库 Schema
│   │   └── index.ts        # 数据库连接
│   └── utils/
│       ├── keyboards.ts    # 键盘组件
│       └── messages.ts     # 消息模板
├── drizzle.config.ts
├── package.json
└── tsconfig.json

💡 提示： grammY 的插件系统非常强大，很多常见功能（会话管理、限流、国际化）都有官方插件。不要自己造轮子，先查 grammY 插件列表。

🚀 二、核心功能实现

2.1 Bot 实例与中间件管道

grammY 的中间件架构与 Koa 类似——每个中间件是一个 async 函数，通过 next() 传递到下一个中间件。这意味着你可以像搭积木一样组合功能：

// src/bot.ts — Bot 实例与中间件配置
import { Bot, session, GrammyError, HttpError } from 'grammy';
import { hydrateReply, parseMode } from '@grammyjs/parse-mode';
import { conversations } from '@grammyjs/conversations';
import type { Context, SessionData } from 'grammy';
import type { ParseMode } from '@grammyjs/types';

// 自定义 Context 类型，扩展 session 和 reply 的类型
interface BotContext extends Context {
  session: SessionData & {
    userId: number;
    username?: string;
    step?: string;
    data?: Record<string, unknown>;
  };
}

// 创建 Bot 实例
const bot = new Bot<BotContext>(process.env.BOT_TOKEN!);

// 中间件管道（按顺序执行）
bot.use(hydrateReply);                    // 给 reply 添加 parseMode 支持
bot.use(session({                         // 会话管理
  initial: () => ({ userId: 0 }),
}));
bot.use(conversations());                 // 多步对话支持

// 全局错误处理
bot.catch((err) => {
  const ctx = err.ctx;
  console.error(`[Bot Error] 处理 update ${ctx.update.update_id} 时出错:`);
  const e = err.error;
  if (e instanceof GrammyError) {
    console.error('Telegram API 错误:', e.description);
  } else if (e instanceof HttpError) {
    console.error('HTTP 请求错误:', e);
  } else {
    console.error('未知错误:', e);
  }
});

export { bot, type BotContext };

⚠️ 警告： 永远不要在生产环境中使用 bot.start() 的默认轮询模式（Long Polling）。轮询模式会持续占用服务器资源、增加延迟，且在多实例部署时会导致消息重复处理。生产环境应使用 Webhook 模式。

2.2 命令处理与内联键盘

Telegram Bot 最常见的交互模式是命令 + 内联键盘。命令用于触发功能，内联键盘用于提供选项：

// src/handlers/start.ts — /start 命令与内联键盘
import { InlineKeyboard } from 'grammy';
import type { BotContext } from '../bot';

// /start 命令处理
export function registerStartHandler(ctx: BotContext) {
  // 记录用户信息到 session
  ctx.session.userId = ctx.from!.id;
  ctx.session.username = ctx.from?.username;

  const keyboard = new InlineKeyboard()
    .text('📊 查看统计', 'action:stats')
    .text('⚙️ 设置', 'action:settings')
    .row()
    .text('❓ 帮助', 'action:help')
    .url('📖 文档', 'https://docs.example.com');

  return ctx.reply(
    `👋 你好，${ctx.from?.first_name}！\n\n` +
    `欢迎使用 Example Bot。请选择一个操作：`,
    { reply_markup: keyboard }
  );
}

// 处理内联键盘回调
export function registerCallbackHandlers(bot: BotContext extends never ? never : any) {
  // 使用正则匹配回调数据（更灵活）
  bot.callbackQuery(/^action:(.+)$/, async (ctx) => {
    const action = ctx.match[1];

    switch (action) {
      case 'stats':
        await ctx.answerCallbackQuery({ text: '正在加载统计数据...' });
        await ctx.editMessageText('📊 统计数据加载中...');
        break;
      case 'settings':
        await ctx.answerCallbackQuery();
        await showSettingsMenu(ctx);
        break;
      case 'help':
        await ctx.answerCallbackQuery();
        await ctx.editMessageText(
          '❓ **使用帮助**\n\n' +
          '/start - 启动 Bot\n' +
          '/stats - 查看统计\n' +
          '/settings - 打开设置\n' +
          '/feedback - 提交反馈',
          { parse_mode: 'Markdown' }
        );
        break;
    }
  });
}

💡 提示： callbackQuery 数据最大长度为 64 字节。如果需要传递复杂数据，建议将数据存入数据库，回调数据只存 ID。不要用 JSON.stringify 直接塞进回调数据。

2.3 多步对话（Conversations）

很多场景需要多步交互——比如收集用户反馈、引导填写表单。grammY 的 @grammyjs/conversations 插件用 Generator 函数实现了优雅的多步对话：

// src/scenes/feedback.ts — 多步对话场景
import type { Conversation } from '@grammyjs/conversations';
import type { BotContext } from '../bot';
import { InlineKeyboard } from 'grammy';

// 反馈收集对话
export async function feedbackConversation(
  conversation: Conversation<BotContext>,
  ctx: BotContext
) {
  // 第一步：选择反馈类型
  const typeKeyboard = new InlineKeyboard()
    .text('🐛 Bug 报告', 'fb:bug')
    .text('💡 功能建议', 'fb:feature')
    .row()
    .text('💬 其他', 'fb:other');

  await ctx.reply('请选择反馈类型：', { reply_markup: typeKeyboard });
  const typeCtx = await conversation.waitForCallbackQuery(/^fb:/);
  const feedbackType = typeCtx.match[1];
  await typeCtx.answerCallbackQuery();

  // 第二步：输入反馈内容
  await ctx.reply('请输入反馈内容（支持文字和图片）：');
  const contentCtx = await conversation.waitFor(['message:text', 'message:photo']);
  const content = contentCtx.message?.text || '[图片消息]';

  // 第三步：确认提交
  const confirmKeyboard = new InlineKeyboard()
    .text('✅ 确认提交', 'fb:confirm')
    .text('❌ 取消', 'fb:cancel');

  await ctx.reply(
    `📋 **反馈预览**\n\n` +
    `类型：${feedbackType === 'bug' ? '🐛 Bug' : feedbackType === 'feature' ? '💡 建议' : '💬 其他'}\n` +
    `内容：${content}\n\n` +
    `确认提交吗？`,
    { reply_markup: confirmKeyboard, parse_mode: 'Markdown' }
  );

  const confirmCtx = await conversation.waitForCallbackQuery(/^fb:(confirm|cancel)$/);
  if (confirmCtx.match[1] === 'confirm') {
    // 保存到数据库
    await conversation.external(() => {
      saveFeedback({
        userId: ctx.from!.id,
        type: feedbackType as string,
        content,
        createdAt: new Date(),
      });
    });
    await confirmCtx.editMessageText('✅ 反馈已提交，感谢你的反馈！');
  } else {
    await confirmCtx.editMessageText('❌ 已取消。');
  }
}

function saveFeedback(data: any) {
  // 数据库保存逻辑
  console.log('保存反馈:', data);
}

🔐 三、生产环境部署与安全

3.1 Webhook 模式部署

生产环境应使用 Webhook 而非 Long Polling。以下是使用 Node.js 原生 HTTP 服务器的 Webhook 部署方案：

// src/index.ts — Webhook 模式启动
import { webhookCallback } from 'grammy';
import { createServer } from 'node:http';
import { bot } from './bot';

// 注册所有 handler
import './handlers/start';
import './handlers/help';
import './handlers/settings';

// 创建 Webhook 回调处理器
const handleUpdate = webhookCallback(bot, 'std/http');

// 环境变量
const PORT = Number(process.env.PORT) || 3000;
const WEBHOOK_PATH = process.env.WEBHOOK_PATH || '/webhook';
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET!;

// HTTP 服务器
const server = createServer(async (req, res) => {
  // 健康检查端点
  if (req.url === '/health') {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({ status: 'ok', uptime: process.uptime() }));
    return;
  }

  // Webhook 端点
  if (req.url === WEBHOOK_PATH && req.method === 'POST') {
    try {
      // 验证 Webhook 密钥（防止伪造请求）
      const secretToken = req.headers['x-telegram-bot-api-secret-token'];
      if (secretToken !== WEBHOOK_SECRET) {
        res.writeHead(403);
        res.end('Forbidden');
        return;
      }

      // 收集请求体
      const chunks: Buffer[] = [];
      for await (const chunk of req) {
        chunks.push(chunk);
      }
      const body = new TextDecoder().decode(Buffer.concat(chunks));

      // 处理 Update
      const response = await handleUpdate(
        new Request(`http://localhost${WEBHOOK_PATH}`, {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body,
        })
      );

      res.writeHead(response.status, { 'Content-Type': 'application/json' });
      res.end(await response.text());
    } catch (err) {
      console.error('Webhook 处理错误:', err);
      res.writeHead(500);
      res.end('Internal Server Error');
    }
    return;
  }

  res.writeHead(404);
  res.end('Not Found');
});

server.listen(PORT, () => {
  console.log(`🤖 Bot 服务器启动在端口 ${PORT}`);
  console.log(`📡 Webhook 地址: http://localhost:${PORT}${WEBHOOK_PATH}`);
});

// 注册 Webhook（首次启动时调用）
async function registerWebhook() {
  const webhookUrl = `${process.env.PUBLIC_URL}${WEBHOOK_PATH}`;
  try {
    await bot.api.setWebhook(webhookUrl, {
      secret_token: WEBHOOK_SECRET,
      allowed_updates: ['message', 'callback_query', 'inline_query'],
      drop_pending_updates: true,  // 丢弃积压的更新
    });
    console.log(`✅ Webhook 已注册: ${webhookUrl}`);
  } catch (err) {
    console.error('❌ Webhook 注册失败:', err);
    process.exit(1);
  }
}

registerWebhook();

⚠️ 警告： Webhook 的 secret_token 参数非常重要。没有它，任何人都可以向你的 Webhook 端点发送伪造的 Update。Telegram 官方强烈建议启用此参数。

3.2 限流与安全防护

生产级 Bot 必须有防滥用机制。以下是基于内存的简单限流实现：

// src/middleware/rateLimit.ts — 简单限流中间件
import type { BotContext } from '../bot';

interface RateLimitEntry {
  count: number;
  resetAt: number;
}

const userLimits = new Map<number, RateLimitEntry>();

// 每用户每分钟最多 20 条消息
const MAX_REQUESTS = 20;
const WINDOW_MS = 60 * 1000;

export function rateLimit(maxRequests = MAX_REQUESTS, windowMs = WINDOW_MS) {
  return async (ctx: BotContext, next: () => Promise<void>) => {
    const userId = ctx.from?.id;
    if (!userId) return next();

    const now = Date.now();
    const entry = userLimits.get(userId);

    if (!entry || now > entry.resetAt) {
      // 新窗口
      userLimits.set(userId, { count: 1, resetAt: now + windowMs });
      return next();
    }

    if (entry.count >= maxRequests) {
      // 超过限制
      const waitSeconds = Math.ceil((entry.resetAt - now) / 1000);
      await ctx.reply(
        `⚠️ 请求过于频繁，请 ${waitSeconds} 秒后再试。`,
        { parse_mode: 'Markdown' }
      );
      return; // 不调用 next()，中断中间件链
    }

    entry.count++;
    return next();
  };
}

3.3 数据库集成与会话持久化

生产环境中，内存中的 session 数据在进程重启后会丢失。使用 Drizzle ORM + SQLite 实现持久化会话：

// src/db/schema.ts — 数据库 Schema
import { sqliteTable, text, integer } from 'drizzle-orm/sqlite-core';

// 用户表
export const users = sqliteTable('users', {
  id: integer('id').primaryKey(),                // Telegram user ID
  username: text('username'),
  firstName: text('first_name'),
  languageCode: text('language_code'),
  isPremium: integer('is_premium', { mode: 'boolean' }),
  createdAt: integer('created_at', { mode: 'timestamp' }),
  lastActiveAt: integer('last_active_at', { mode: 'timestamp' }),
});

// 会话表（持久化 session 数据）
export const sessions = sqliteTable('sessions', {
  key: text('key').primaryKey(),                 // "bot:userId:chatId"
  data: text('data'),                            // JSON 字符串
  expiresAt: integer('expires_at', { mode: 'timestamp' }),
});

// 反馈表
export const feedbacks = sqliteTable('feedbacks', {
  id: integer('id').primaryKey({ autoIncrement: true }),
  userId: integer('user_id').references(() => users.id),
  type: text('type'),                            // bug / feature / other
  content: text('content'),
  status: text('status').default('pending'),     // pending / reviewed / resolved
  createdAt: integer('created_at', { mode: 'timestamp' }),
});

// src/db/index.ts — 数据库连接
import Database from 'better-sqlite3';
import { drizzle } from 'drizzle-orm/better-sqlite3';
import * as schema from './schema';

const sqlite = new Database(process.env.DATABASE_PATH || './bot.db');

// 开启 WAL 模式（提升并发读写性能）
sqlite.pragma('journal_mode = WAL');

export const db = drizzle(sqlite, { schema });

📌 记住： SQLite 的 WAL（Write-Ahead Logging）模式允许多个读操作和一个写操作并发执行，非常适合 Telegram Bot 这种「读多写少」的场景。不开启 WAL 模式，并发写入时会出现 SQLITE_BUSY 错误。

📊 四、性能优化与监控

4.1 批量操作与并发控制

当 Bot 需要向大量用户发送消息时（如通知推送），必须控制并发速率。Telegram API 的限制是每秒最多 30 条消息（同一聊天）和每秒最多 20 条消息（不同聊天）：

// 批量发送消息（带并发控制）
async function broadcastMessage(
  userIds: number[],
  message: string,
  options?: { batchSize?: number; delayMs?: number }
) {
  const { batchSize = 20, delayMs = 1000 } = options ?? {};
  const results = { sent: 0, failed: 0, errors: [] as string[] };

  // 分批处理
  for (let i = 0; i < userIds.length; i += batchSize) {
    const batch = userIds.slice(i, i + batchSize);

    // 并发发送当前批次
    const promises = batch.map(async (userId) => {
      try {
        await bot.api.sendMessage(userId, message, { parse_mode: 'Markdown' });
        results.sent++;
      } catch (err: any) {
        results.failed++;
        // 处理常见的 Telegram API 错误
        if (err.description?.includes('blocked')) {
          results.errors.push(`用户 ${userId} 已屏蔽 Bot`);
        } else if (err.description?.includes('chat not found')) {
          results.errors.push(`用户 ${userId} 聊天不存在`);
        } else {
          results.errors.push(`用户 ${userId}: ${err.description}`);
        }
      }
    });

    await Promise.all(promises);

    // 批次间延迟（避免触发 Telegram API 限流）
    if (i + batchSize < userIds.length) {
      await new Promise((resolve) => setTimeout(resolve, delayMs));
    }
  }

  return results;
}

4.2 生产环境检查清单

部署 Telegram Bot 到生产环境前，确保以下所有项目都已就绪：

检查项 说明 推荐方案
✅ Webhook 部署 不使用 Long Polling Nginx 反向代理 + HTTPS
✅ Secret Token 防止 Webhook 伪造 环境变量，长度 ≥ 32 字符
✅ 限流中间件 防止滥用 每用户每分钟 20 次
✅ 错误监控 捕获未处理异常 Sentry / 自建日志
✅ 数据持久化 Session 和用户数据 SQLite WAL / PostgreSQL
✅ 健康检查 监控服务存活 /health 端点
✅ 优雅关闭 处理 SIGTERM 信号 取消 Webhook + 清理资源
✅ 日志记录 追踪请求和错误 Pino 结构化日志 
// 优雅关闭处理
process.on('SIGTERM', async () => {
  console.log('收到 SIGTERM 信号，正在优雅关闭...');

  // 1. 停止接受新的 Webhook 请求
  try {
    await bot.api.deleteWebhook({ drop_pending_updates: false });
    console.log('✅ Webhook 已取消');
  } catch (err) {
    console.error('取消 Webhook 失败:', err);
  }

  // 2. 关闭数据库连接
  // db.$client.close();

  // 3. 关闭 HTTP 服务器
  server.close(() => {
    console.log('✅ 服务器已关闭');
    process.exit(0);
  });

  // 5 秒后强制退出
  setTimeout(() => {
    console.error('⚠️ 强制退出');
    process.exit(1);
  }, 5000);
});

💡 五、最佳实践与避坑指南

5.1 常见坑点

  • ❌ 不要在回调查询中执行耗时操作 — answerCallbackQuery 有 3 秒超时。如果处理逻辑超过 3 秒，用户会看到「加载中…」的转圈动画。应先 answerCallbackQuery，再执行耗时操作。

  • ❌ 不要用 Markdown V1 — Telegram 有两种 Markdown 解析模式：Markdown（V1）和 MarkdownV2。V1 的转义规则混乱且不完整，永远使用 MarkdownV2 或 HTML。

  • ❌ 不要忽略 callback_query 的重复触发 — 用户可能快速多次点击同一按钮，导致回调被触发多次。使用 conversation 的 waitForCallbackQuery 或在处理前检查状态。

  • ❌ 不要硬编码 Chat ID — 测试时容易把消息发到错误的聊天。使用环境变量管理目标 Chat ID。

5.2 消息格式化最佳实践

// ❌ 错误写法：直接拼接字符串（特殊字符会导致解析失败）
const msg = `价格：$${price}（含税）`;
// 如果 price = "10.50"，消息中的 $ 会被当作 Markdown 标记

// ✅ 正确写法：使用 HTML 格式（更安全）
const msg = `价格：<b>¥${price}</b>（含税）`;
await ctx.reply(msg, { parse_mode: 'HTML' });

// ✅ 或者使用 MarkdownV2 并转义特殊字符
function escapeMarkdown(text: string): string {
  return text.replace(/([_*\[\]()~`>#+\-=|{}.!\\])/g, '\\$1');
}
const msg2 = `价格：*${escapeMarkdown(`¥${price}`)}*（含税）`;
await ctx.reply(msg2, { parse_mode: 'MarkdownV2' });

⚡ 关键结论： 如果你不确定 Markdown 转义规则，直接用 HTML 格式。HTML 的标签语法更直观，不需要处理复杂的字符转义，出错概率远低于 MarkdownV2。

📝 总结

构建生产级 Telegram Bot 的核心不是 API 调用，而是工程化——中间件架构、错误处理、限流防护、数据持久化、Webhook 部署和优雅关闭。grammY 提供了完善的类型安全和插件生态，让这些工程化实践变得简单。

推荐的技术栈组合：

  • 🔧 框架：grammY（TypeScript 原生，插件丰富）
  • 🗄️ 数据库：SQLite + Drizzle ORM（轻量、零配置、WAL 模式）
  • 📡 部署：Webhook + Nginx 反向代理 + HTTPS
  • 📊 监控：Pino 日志 + Sentry 错误追踪
  • 🔐 安全：Webhook Secret Token + 限流中间件

相关资源：

📚 相关文章

MCP 协议完全指南：从零构建 AI 工具服务器的实战教程

2026-07-01·15 分钟

AbortController 进阶模式：从请求取消到全局资源管理的信号传播艺术

2026-06-12·15 分钟

AI 生成前端代码质量工程：从粗糙原型到生产级 UI 的系统化方法

2026-06-12·15 分钟