多平台发布为什么要隔离状态

一个真实的故障案例

2026年5月20日，我遇到了一次奇怪的故障。

那天要发布一篇文章到微信和小红书两个平台。微信推送成功了，但小红书API返回了错误："图片格式不支持"。按理说，微信已经成功了，小红书失败不应该影响微信，两个平台是独立的。

但实际情况是：因为小红书失败，整个发布流程抛出了异常，异常向上传播，把微信的"已发布"状态也给回滚了。 Registry里，这篇文章的状态变回了"草稿"。

下次调度器扫描到这篇文章，看状态是"草稿"，又重新推了一遍微信。微信后台出现了重复的草稿。

问题的根源是：状态没有隔离。微信和小红书共用一个事务，一个失败，全部回滚。正确的做法应该是：每个平台独立执行，独立记录状态，一个平台失败不影响其他平台。

这就是"隔离状态"要解决的问题。

什么是状态隔离

状态隔离，指的是多平台发布时，每个平台的发布过程、状态记录、错误处理，都是独立的，互不影响。

具体包含三个维度：

1. 执行隔离

微信的发布任务和小红书的发布任务，在不同的执行流中运行。可以是不同的线程、不同的进程、不同的机器，甚至可以是不同的调度周期。

执行隔离的好处是：一个平台慢，不会拖慢其他平台。比如微博API响应慢，要等5秒，如果和小红书在同一个执行流，小红书也要等5秒。执行隔离后，微博慢慢跑，小红书不用等。

2. 状态隔离

每个平台维护自己的状态字段，不共用一个"总状态"。

错误做法：status = "published"（一个状态字段管所有平台）

正确做法：

{
  "wechat_status": "published",
  "xiaohongshu_status": "failed",
  "blog_status": "pending",
  "weibo_status": "published"
}

这样即使小红书失败，微信和微博的状态不受影响。查询"哪些文章微信已发布"，只看 wechat_status 字段，不用管其他平台。

3. 错误隔离

一个平台的错误，不应该导致整个发布流程终止。

错误做法：

try:
    publish_to_wechat(article)
    publish_to_xiaohongshu(article)  # 如果这行失败，抛出异常
    publish_to_blog(article)
except Exception as e:
    rollback_all()  # 微信也跟着回滚

正确做法：

results = {
    "wechat": publish_to_wechat(article),  # 成功
    "xiaohongshu": publish_to_xiaohongshu(article),  # 失败，但捕获异常
    "blog": publish_to_blog(article)
}

# 汇总结果，不是一失败就全部回滚
for platform, result in results.items():
    if result["status"] == "failed":
        log_error(platform, result["error"])
        schedule_retry(platform)

为什么需要状态隔离

原因1：各平台的API特性差异巨大

四个平台的API，稳定性、响应速度、限流策略完全不同：

微信公众平台：API相对稳定，响应时间1-2秒，每天可推送次数有限制（订阅号1次/天，服务号4次/月）。

小红书：没有官方API，只能用爬虫或第三方接口，稳定性差，经常返回"操作频繁请稍后再试"，需要重试策略。

微博：API响应快（500ms以内），但偶尔会丢包，推送请求发出去了，但微博后台没收到，需要主动查询确认。

博客（sikaoa.com）：基于Sanity CMS，API稳定，但没有"推送"概念，是"创建/更新文章"，操作是幂等的（重复调用不会创建重复文章）。

如果你用一个统一的状态机管所有平台，状态机会变得极其复杂，要处理各种边界情况。状态隔离后，每个平台有自己的状态机，简单很多。

原因2：故障域不同

假设你的网络出现故障，微信API连不上，但小红书能连上。如果状态不隔离，整个发布流程会卡在微信这一步，小红书也没法推。

状态隔离后，微信推送到一半失败了，标记 wechat_status = "failed"，但小红书和微博继续推，不受微信故障影响。

再比如，小红书账号被封了，所有小红书发布都失败。这时候你希望其他平台继续发，只停用小红书。如果状态不隔离，可能整个发布流程都停了。

原因3：重试策略不同

不同平台的最佳重试策略不同：

微信：失败后立即重试意义不大，因为失败通常是内容问题（比如有敏感词），重试还是会失败。正确做法是：失败→人工审核→修改内容→重新推送。所以微信失败不需要自动重试，标记为"失败待人工处理"即可。

小红书：失败大多是临时性的（"操作频繁"），等10分钟再试可能就成功了。所以需要自动重试，比如每隔10分钟重试一次，最多重试5次。

微博：失败可能是网络抖动，立即重试通常能成功。所以重试策略是：失败→等待3秒→重试→如果还失败，标记为"失败待人工处理"。

博客：失败通常是Sanity token过期，需要重新登录。这种失败不适合重试，应该立即停止所有博客发布任务，发告警通知人工处理。

如果用一个统一的状态机，很难为不同平台配置不同的重试策略。状态隔离后，每个平台模块自己管理重试，灵活很多。

状态隔离的实现方案

jianfei-wechat v4中，状态隔离的实现分为三层：

第一层：任务队列隔离

位于 v4/scheduler/queue.py，为每个平台维护独立的任务队列。

PLATFORM_TABLES = {
    "wechat": "wechat_tasks",
    "xiaohongshu": "xiaohongshu_tasks",
    "weibo": "weibo_tasks",
    "blog": "blog_tasks"
}

调度器从各自队列取任务，互不影响。微信队列堵了（比如大量文章等待推送），不影响小红书队列的消费。

第二层：状态字段隔离

位于 v4/articles.py 的 ArticleRecord 类，每个平台有独立的状态字段。

@dataclass
class ArticleRecord:
    # 通用字段
    id: str
    title: str
    status: str  # 顶层聚合状态
    
    # 微信专用
    wechat_draft_status: str = ""
    wechat_draft_media_id: str = ""
    
    # 小红书专用
    xiaohongshu_status: str = ""
    xiaohongshu_post_id: str = ""
    
    # 微博专用
    weibo_status: str = ""
    weibo_post_id: str = ""
    
    # 博客专用
    blog_status: str = ""
    blog_post_id: str = ""
    blog_url: str = ""

虽然这些字段在同一个数据类里，但更新时是独立的。更新 wechat_draft_status 不会连带修改 xiaohongshu_status。

第三层：错误隔离

位于 v4/platforms/ 目录下，每个平台的发布函数都自己捕获异常，不向上传播。

以微信发布为例（v4/platforms/wechat.py）：

def publish_to_wechat(article: ArticleRecord) -> dict:
    try:
        media_id = wechat_api.upload_draft(article.md_path)
        return {
            "status": "success",
            "media_id": media_id
        }
    except WechatAPIError as e:
        log_error(f"Wechat publish failed: {e}")
        return {
            "status": "failed",
            "error": str(e),
            "should_retry": False  # 微信失败通常不重试
        }

调用方（调度器）拿到返回值，根据 status 字段更新对应平台的状态，不抛异常。

状态隔离的代价

状态隔离不是免费的，有三个代价：

代价1：顶层状态聚合变复杂

状态隔离后，每个平台有自己的子状态。那"这篇文章到底有没有发布完成"，要看所有平台的子状态，这叫"状态聚合"。

聚合逻辑不复杂，但容易写错。比如规则是"所有平台都成功才算发布完成"，但如果某篇文章只配置了微信和博客两个平台，"完成"的定义是 wechat_status == "published" AND blog_status == "published"。如果错误地写成 wechat_status == "published" AND xiaohongshu_status == "published"，会因为小红书状态是空字符串而误判为"未完成"。

解决方法：在 ArticleRecord 里加一个 platforms 字段，记录这篇文章要发布到哪些平台。聚合状态时，只检查 platforms 里列出的平台。

代价2：用户体验变复杂

如果用户在Dashboard上看到一篇文章的状态是"部分完成"，他不知道是什么意思，得点进去看每个平台的子状态。

解决方法：Dashboard上同时显示顶层状态和各平台图标。顶层状态用文字（"已完成"、"部分完成"、"失败"），平台子状态用图标（✅已发布、⏳待推送、❌失败、⚠️审核中）。这样一眼就能看出哪篇有问题，不用逐个点进去看。

代价3：重试逻辑分散

状态不隔离时，重试逻辑集中在一个地方（比如调度器里统一重试）。状态隔离后，每个平台模块自己管重试，代码分散。

解决方法：把重试逻辑抽象成装饰器或工具函数，各平台模块共用。比如：

# v4/utils/retry.py
def with_retry(max_retries=3, delay=10):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if i == max_retries - 1:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator

# v4/platforms/xiaohongshu.py
@with_retry(max_retries=5, delay=600)  # 小红书：重试5次，间隔10分钟
def publish_to_xiaohongshu(article):
    ...

实际案例：一次部分失败的处理

再看一个实际案例。2026年6月8日，文章《Agent协作的边界》发布到四个平台，结果：

微信：成功
小红书：失败（图片审核不通过）
微博：成功
博客：失败（Sanity token过期）

如果状态不隔离，这次发布会被标记为"失败"，然后全部回滚。但实际上微信和微博已经成功了，回滚会导致重复推送。

因为状态隔离，实际处理是：

微信状态更新为 wechat_draft_status = "published"
小红书状态更新为 xiaohongshu_status = "failed"，并记录失败原因
微博状态更新为 weibo_status = "published"
博客状态更新为 blog_status = "failed"，并发送告警通知

用户在Dashboard上看到这篇文章的状态是"部分完成"，点进去看到：

微信：✅ 已发布
小红书：❌ 失败（图片审核不通过）
微博：✅ 已发布
博客：❌ 失败（token过期）

用户可以：

忽略小红书和博客的失败（如果觉得不重要）
替换小红书的图片，重新推送
重新登录Sanity，重试博客发布

所有操作都是针对单个平台的，不影响其他平台。

总结

状态隔离的核心思想：把多平台发布看成多个独立的子任务，而不是一个原子任务。

这样做的好处：

容错性强：一个平台失败，其他平台不受影响
灵活度高：每个平台可以配置不同的重试策略、超时时间、错误处理
故障域小：某个平台的API挂了，只影响这个平台，不拖垮整个系统

代价是复杂度上移：顶层需要额外的逻辑来聚合各平台的子状态，给用户一个统一视图。

但这个代价是值得的。当你从单平台发布进化到多平台发布，状态隔离不是可选项，是必选项。没有状态隔离，多平台发布系统会在第一次遇到部分失败时崩溃。

隔离状态，才能稳健扩展。

文章信息

写作时间：2026-06-13
字数：约3100字
状态：草稿
下一步：配图生成、质量检查、发布到博客

一个真实的故障案例

2026年5月20日，我遇到了一次奇怪的故障。

下次调度器扫描到这篇文章，看状态是"草稿"，又重新推了一遍微信。微信后台出现了重复的草稿。

这就是"隔离状态"要解决的问题。

什么是状态隔离

状态隔离，指的是多平台发布时，每个平台的发布过程、状态记录、错误处理，都是独立的，互不影响。

具体包含三个维度：

1. 执行隔离

微信的发布任务和小红书的发布任务，在不同的执行流中运行。可以是不同的线程、不同的进程、不同的机器，甚至可以是不同的调度周期。

2. 状态隔离

每个平台维护自己的状态字段，不共用一个"总状态"。

错误做法：status = "published"（一个状态字段管所有平台）

正确做法：

{
  "wechat_status": "published",
  "xiaohongshu_status": "failed",
  "blog_status": "pending",
  "weibo_status": "published"
}

这样即使小红书失败，微信和微博的状态不受影响。查询"哪些文章微信已发布"，只看 wechat_status 字段，不用管其他平台。

3. 错误隔离

一个平台的错误，不应该导致整个发布流程终止。

错误做法：

try:
    publish_to_wechat(article)
    publish_to_xiaohongshu(article)  # 如果这行失败，抛出异常
    publish_to_blog(article)
except Exception as e:
    rollback_all()  # 微信也跟着回滚

正确做法：

results = {
    "wechat": publish_to_wechat(article),  # 成功
    "xiaohongshu": publish_to_xiaohongshu(article),  # 失败，但捕获异常
    "blog": publish_to_blog(article)
}

# 汇总结果，不是一失败就全部回滚
for platform, result in results.items():
    if result["status"] == "failed":
        log_error(platform, result["error"])
        schedule_retry(platform)

为什么需要状态隔离

原因1：各平台的API特性差异巨大

四个平台的API，稳定性、响应速度、限流策略完全不同：

微信公众平台：API相对稳定，响应时间1-2秒，每天可推送次数有限制（订阅号1次/天，服务号4次/月）。

小红书：没有官方API，只能用爬虫或第三方接口，稳定性差，经常返回"操作频繁请稍后再试"，需要重试策略。

微博：API响应快（500ms以内），但偶尔会丢包，推送请求发出去了，但微博后台没收到，需要主动查询确认。

博客（sikaoa.com）：基于Sanity CMS，API稳定，但没有"推送"概念，是"创建/更新文章"，操作是幂等的（重复调用不会创建重复文章）。

如果你用一个统一的状态机管所有平台，状态机会变得极其复杂，要处理各种边界情况。状态隔离后，每个平台有自己的状态机，简单很多。

原因2：故障域不同

假设你的网络出现故障，微信API连不上，但小红书能连上。如果状态不隔离，整个发布流程会卡在微信这一步，小红书也没法推。

状态隔离后，微信推送到一半失败了，标记 wechat_status = "failed"，但小红书和微博继续推，不受微信故障影响。

再比如，小红书账号被封了，所有小红书发布都失败。这时候你希望其他平台继续发，只停用小红书。如果状态不隔离，可能整个发布流程都停了。

原因3：重试策略不同

不同平台的最佳重试策略不同：

小红书：失败大多是临时性的（"操作频繁"），等10分钟再试可能就成功了。所以需要自动重试，比如每隔10分钟重试一次，最多重试5次。

微博：失败可能是网络抖动，立即重试通常能成功。所以重试策略是：失败→等待3秒→重试→如果还失败，标记为"失败待人工处理"。

博客：失败通常是Sanity token过期，需要重新登录。这种失败不适合重试，应该立即停止所有博客发布任务，发告警通知人工处理。

如果用一个统一的状态机，很难为不同平台配置不同的重试策略。状态隔离后，每个平台模块自己管理重试，灵活很多。

状态隔离的实现方案

jianfei-wechat v4中，状态隔离的实现分为三层：

第一层：任务队列隔离

位于 v4/scheduler/queue.py，为每个平台维护独立的任务队列。

PLATFORM_TABLES = {
    "wechat": "wechat_tasks",
    "xiaohongshu": "xiaohongshu_tasks",
    "weibo": "weibo_tasks",
    "blog": "blog_tasks"
}

调度器从各自队列取任务，互不影响。微信队列堵了（比如大量文章等待推送），不影响小红书队列的消费。

第二层：状态字段隔离

位于 v4/articles.py 的 ArticleRecord 类，每个平台有独立的状态字段。

@dataclass
class ArticleRecord:
    # 通用字段
    id: str
    title: str
    status: str  # 顶层聚合状态
    
    # 微信专用
    wechat_draft_status: str = ""
    wechat_draft_media_id: str = ""
    
    # 小红书专用
    xiaohongshu_status: str = ""
    xiaohongshu_post_id: str = ""
    
    # 微博专用
    weibo_status: str = ""
    weibo_post_id: str = ""
    
    # 博客专用
    blog_status: str = ""
    blog_post_id: str = ""
    blog_url: str = ""

虽然这些字段在同一个数据类里，但更新时是独立的。更新 wechat_draft_status 不会连带修改 xiaohongshu_status。

第三层：错误隔离

位于 v4/platforms/ 目录下，每个平台的发布函数都自己捕获异常，不向上传播。

以微信发布为例（v4/platforms/wechat.py）：

def publish_to_wechat(article: ArticleRecord) -> dict:
    try:
        media_id = wechat_api.upload_draft(article.md_path)
        return {
            "status": "success",
            "media_id": media_id
        }
    except WechatAPIError as e:
        log_error(f"Wechat publish failed: {e}")
        return {
            "status": "failed",
            "error": str(e),
            "should_retry": False  # 微信失败通常不重试
        }

调用方（调度器）拿到返回值，根据 status 字段更新对应平台的状态，不抛异常。

状态隔离的代价

状态隔离不是免费的，有三个代价：

代价1：顶层状态聚合变复杂

状态隔离后，每个平台有自己的子状态。那"这篇文章到底有没有发布完成"，要看所有平台的子状态，这叫"状态聚合"。

解决方法：在 ArticleRecord 里加一个 platforms 字段，记录这篇文章要发布到哪些平台。聚合状态时，只检查 platforms 里列出的平台。

代价2：用户体验变复杂

如果用户在Dashboard上看到一篇文章的状态是"部分完成"，他不知道是什么意思，得点进去看每个平台的子状态。

代价3：重试逻辑分散

状态不隔离时，重试逻辑集中在一个地方（比如调度器里统一重试）。状态隔离后，每个平台模块自己管重试，代码分散。

解决方法：把重试逻辑抽象成装饰器或工具函数，各平台模块共用。比如：

# v4/utils/retry.py
def with_retry(max_retries=3, delay=10):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if i == max_retries - 1:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator

# v4/platforms/xiaohongshu.py
@with_retry(max_retries=5, delay=600)  # 小红书：重试5次，间隔10分钟
def publish_to_xiaohongshu(article):
    ...

实际案例：一次部分失败的处理

再看一个实际案例。2026年6月8日，文章《Agent协作的边界》发布到四个平台，结果：

微信：成功
小红书：失败（图片审核不通过）
微博：成功
博客：失败（Sanity token过期）

如果状态不隔离，这次发布会被标记为"失败"，然后全部回滚。但实际上微信和微博已经成功了，回滚会导致重复推送。

因为状态隔离，实际处理是：

微信状态更新为 wechat_draft_status = "published"
小红书状态更新为 xiaohongshu_status = "failed"，并记录失败原因
微博状态更新为 weibo_status = "published"
博客状态更新为 blog_status = "failed"，并发送告警通知

用户在Dashboard上看到这篇文章的状态是"部分完成"，点进去看到：

微信：✅ 已发布
小红书：❌ 失败（图片审核不通过）
微博：✅ 已发布
博客：❌ 失败（token过期）

用户可以：

忽略小红书和博客的失败（如果觉得不重要）
替换小红书的图片，重新推送
重新登录Sanity，重试博客发布

所有操作都是针对单个平台的，不影响其他平台。

总结

状态隔离的核心思想：把多平台发布看成多个独立的子任务，而不是一个原子任务。

这样做的好处：

容错性强：一个平台失败，其他平台不受影响
灵活度高：每个平台可以配置不同的重试策略、超时时间、错误处理
故障域小：某个平台的API挂了，只影响这个平台，不拖垮整个系统

代价是复杂度上移：顶层需要额外的逻辑来聚合各平台的子状态，给用户一个统一视图。

隔离状态，才能稳健扩展。

文章信息

写作时间：2026-06-13
字数：约3100字
状态：草稿
下一步：配图生成、质量检查、发布到博客