本章目录
第 7 章
群里人一多，消息开始丢
同一个群每分钟只能发 20 条，超了 Telegram 直接拒绝。加上昵称里一个下划线就能让消息发不出去——这一章处理所有「发不出去」的情况。
消息发不出去的三种原因
bot 在小群里跑得好好的，换到几百人的群就开始「偶尔没反应」。别急着怀疑自己的业务逻辑——发送这一步本身就会失败，而且有三种完全不同的原因：
| 原因 | 报错里的关键词 | 什么时候撞上 |
|---|---|---|
| 发太快 | TelegramRetryAfterToo Many Requests | 群人一多、或者定时任务批量推送时 |
| 格式不对 | can't parse entities | 用了 Markdown / HTML 格式，而内容里有特殊字符——最常见的来源是用户昵称 |
| 太长 | message is too long | 一条消息超过 4096 字符 |
发太快：Telegram 的限制到底是多少
官方没给一份完整的数字表，但实践中公认的大致边界是：
| 全局 | 每秒约 30 条 |
| 同一个群 | 每分钟约 20 条 —— 这条最容易撞 |
| 同一个人的私聊 | 每秒约 1 条 |
「每分钟 20 条」意味着平均三秒才能发一条。一个 200 人的群同时签到，你要回 200 条消息——按这个速度得十分钟。
正确的应对：听它的，别硬来
错误做法是立刻重试——那只会让情况更糟，你的请求会堆成雪球。正确做法是按它说的秒数等：
import asyncioimport loggingfrom aiogram import Botfrom aiogram.exceptions import ( TelegramRetryAfter, TelegramNetworkError, TelegramForbiddenError, TelegramBadRequest,)log = logging.getLogger(__name__)async def safe_send(bot: Bot, chat_id: int, text: str, retries: int = 3): """发消息，自动处理限流和网络抖动。发不出去返回 None。""" for attempt in range(retries): try: return await bot.send_message(chat_id, text) except TelegramRetryAfter as e: # Telegram 明确告诉你等多久，照做 log.warning("限流，等 %s 秒", e.retry_after) await asyncio.sleep(e.retry_after) except TelegramNetworkError: # 网络抖动，指数退避：1、2、4 秒 wait = 2 ** attempt log.warning("网络问题，%s 秒后重试", wait) await asyncio.sleep(wait) except TelegramForbiddenError: # 用户把 bot 拉黑了，或者 bot 被踢出群。重试没意义 log.info("发不了 %s：对方不让发", chat_id) return None except TelegramBadRequest as e: # 格式或长度不对。重试还是同样的错，直接放弃并记下来 log.error("消息本身有问题：%s", e) return None log.error("重试 %s 次仍然失败：%s", retries, chat_id) return None
很多人写重试只 except Exception 然后无脑重来。但被拉黑和格式错误重试一万次也是同样的结果，白白浪费配额还拖慢别人的消息。
分清「等一等能好的」和「等多久都不会好的」，比重试逻辑本身更重要。
主动限速：别等被拒了再退
退避是被动挨打。如果你知道自己要往同一个群发很多条（比如定时推送），主动放慢更好：
import timefrom collections import defaultdict_last_sent: dict[int, float] = defaultdict(float)MIN_GAP = 3.2 # 同一个群，两条之间至少隔这么久（20 条/分钟）async def paced_send(bot: Bot, chat_id: int, text: str): """发之前先看看离上一条够不够久。""" gap = time.monotonic() - _last_sent[chat_id] if gap < MIN_GAP: await asyncio.sleep(MIN_GAP - gap) _last_sent[chat_id] = time.monotonic() return await safe_send(bot, chat_id, text)
用 time.monotonic() 而不是 time.time()——后者会被系统对时影响，可能突然跳一大截或者倒退。
200 人签到就回 200 条，本身就是设计问题。改成每人签到只回一个短提示，或者干脆只在私聊回，群里的问题就消失了。
限流代码是兜底，不是让你放心刷屏的许可证。
格式不对：昵称里的一个下划线就能炸
第 3 章我特意让所有回复都用纯文本，就是为了绕开这件事。现在说清楚为什么。
如果你用 parse_mode="MarkdownV2"，那么这 18 个字符都必须转义：
_ * [ ] ( ) ~ ` > # + - = | { } . !
注意里面有 小数点 和 减号。也就是说「价格 9.9 元」这句话直接发会失败。实测几个真实昵称：
| 用户昵称 | 转义后 | 不转义会怎样 |
|---|---|---|
小明_测试 | 小明\_测试 | 下划线被当成斜体标记，整条消息解析失败 |
张三(备注) | 张三\(备注\) | 括号被当成链接语法 |
价格 9.9 元 | 价格 9\.9 元 | 连小数点都得转 |
正常中文昵称 | 原样 | 没事 |
😀 emoji 用户 | 原样 | emoji 不用转 |
两个解法，先说更省事的那个
解法一：用 HTML 格式，不用 Markdown。HTML 只有三个字符要转义（< > &），而且 aiogram 自带函数：
from aiogram import htmlfrom aiogram.enums import ParseModename = message.from_user.full_name # 可能含任何字符await message.answer( f"欢迎 <b>{html.quote(name)}</b>", parse_mode=ParseMode.HTML,)
html.quote() 处理掉那三个字符就安全了。凡是来自用户的内容，进模板之前都过一遍它。
解法二：干脆不用格式。像第 3 章那样纯文本发，一个特殊字符都不用管。签到 bot 这种场景，加粗不加粗根本不影响体验。
格式错误不会让程序崩，只会让某些用户收不到消息——恰好是昵称带特殊字符的那些人。你看日志一片正常，用户那边一脸问号。
所以宁可不用格式，也别用了不转义。
太长：4096 字符
/rank 只显示前十，撞不到。但只要有「导出全部记录」这类功能，迟早会撞。分片函数很短：
LIMIT = 4096def split_text(text: str, limit: int = LIMIT) -> list[str]: """按 4096 切开，优先在换行处切。""" parts = [] while len(text) > limit: cut = text.rfind("\n", 0, limit) if cut <= 0: cut = limit # 一整段没换行，只能硬切 parts.append(text[:cut]) text = text[cut:].lstrip("\n") if text: parts.append(text) return parts
刚好 4096（1 段）、4097（2 段）、900 行带换行的长文（在换行处切开）、10000 个连续字符没有换行（硬切 3 段）—— 每段都不超 4096。
写这类函数一定要测「刚好等于上限」和「上限 +1」，差一错误就藏在这儿。
但更该问的是：为什么要发这么长？超过 4096 的内容，用文件发出去体验好得多：
from aiogram.types import BufferedInputFiledata = report_text.encode("utf-8")await message.answer_document( BufferedInputFile(data, filename="记录.txt"), caption="你的全部签到记录",)
全局错误处理：别让一条消息搞挂整个 bot
到目前为止，任何一个 handler 抛异常，用户那边是完全没有反馈的——日志里一堆红字，他只看到 bot 装死。
加一个全局兜底：
from aiogram.types import ErrorEvent@dp.errors()async def on_error(event: ErrorEvent) -> bool: """任何 handler 抛出的异常都会到这里。""" logging.exception("处理更新时出错: %s", event.exception) # 尽量给用户一句反馈，别让他对着空气 msg = event.update.message or getattr(event.update.callback_query, "message", None) if msg: try: await msg.answer("出了点问题，我已经记下来了。稍后再试试") except Exception: pass # 连报错都发不出去，那就算了 return True # True = 已处理，别再往上抛
logging.exception | 会连完整堆栈一起记下来，比 logging.error 有用得多 |
| 回复也包在 try 里 | 错误处理本身也会失败（比如刚好撞上限流）。它挂了就真没救了 |
return True | 告诉 aiogram「这个异常我接了」。不返回的话它会继续往上抛 |
注意 @dp.errors() 要挂在 Dispatcher 上，不是 Router。
顺手把日志配好
出了事要能查。第 2 章那句 logging.basicConfig(level=logging.INFO) 太简陋了——它不带时间戳，出事你连什么时候发生的都不知道：
logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", datefmt="%m-%d %H:%M:%S",)# aiogram 自己的日志有点吵，调高一档logging.getLogger("aiogram.event").setLevel(logging.WARNING)
第 10 章讲部署时会把日志接到 systemd，那时候这个格式就派上用场了。
验一遍
| 怎么试 | 应该看到 |
|---|---|
在某个 handler 里故意写 1/0，然后触发它 | 用户收到「出了点问题」，bot 没退出，日志里有完整堆栈 |
把 Telegram 昵称改成 test_user，用 MarkdownV2 发一条带昵称的消息 | 不转义 → 发送失败；用 html.quote + HTML 模式 → 正常 |
| 循环给同一个群发 30 条 | 日志里出现「限流，等 N 秒」，但最终都发出去了 |
| 发一条 5000 字符的消息 | 自动切成两条 |
| 把 bot 从群里踢掉再让它发消息 | 日志记一句「对方不让发」，不重试 |
这一章的要点
- 发送失败有三种：发太快、格式不对、太长，各有各的解法
- 限流要按
retry_after等，不能立刻重试；网络问题用指数退避 - 分清能重试和不能重试的异常：被拉黑、格式错误，重试一万次也白搭
- 用 HTML 模式 +
html.quote()，别用 MarkdownV2（18 个特殊字符，小数点都算） - 加
@dp.errors()全局兜底，错误处理本身也要包 try - 最好的限流方案是少发消息，代码只是兜底
把签到的回复改成：群里只回一个极短的确认，详细积分信息私聊发给他。这样群里的消息量降下来了，用户拿到的信息反而更全。
提示：私聊要用 bot.send_message(user_id, ...)，而且要用上 safe_send —— 如果用户没跟 bot 私聊过，会抛 TelegramForbiddenError，这时候要在群里提示他先私聊一下。
限流、转义、超长——这三类问题你撞到过哪种？报错原文贴进来最好。留不留联系方式都行。