一、引子:一次点击到上屏,卡在哪了

现象很典型:

  • 界面响应"能感觉到慢",但 CPU 占用只有 20%
  • 掉帧是偶发的,压测时复现不了
  • 单看每个函数的耗时都正常,没有任何一个函数"很慢"

这类问题最容易把人带进死胡同:你去优化那个"最慢的函数",改了半天,现象不变。

换个问法就有出口:不是"谁慢",而是"谁在等谁"。

一旦开始查"等",锁就会浮出水面。而锁的问题里,最常见的不是"忘了加锁",而是锁的范围画错了——把不该锁的东西锁进去了。


二、先认清锁的三笔开销

讨论粒度之前,得先知道"锁到底花的是什么钱"。

开销量级(单核 MCU,仅供数量级参考)说明
加解锁本身几十~几百个 CPU 周期无争抢时最快,只涉及几条原子指令和调度器状态修改
等待(争抢)微秒到毫秒,甚至无限拿不到锁必须让出 CPU,等多久取决于持锁者的临界区有多长
上下文切换几微秒睡眠/唤醒的代价,争抢的必然伴生品

三者的比例关系决定了一切:

  • 无争抢时,锁的代价 ≈ 第一笔,几乎可以忽略
  • 有争抢时,代价由第二笔主导,而第二笔完全由持锁时长决定

所以有一句值得记住的话:

锁的性能不取决于锁有多快,而取决于锁被持有的时间有多长、被抢的概率有多大。

粒度优化的本质,就是把这两者同时压下去。


三、粒度五级阶梯

从"一把大锁"到"无锁",中间有几级明确的台阶:

级别做法并发度漏锁风险实现复杂度典型场景
全局大锁一个模块/一个子系统一把锁最低最低(全串行)最低资源极少的 MCU、逻辑简单的模块
模块锁每个子模块一把低低低模块边界清晰的系统
对象锁每个实例/每个对象一把中中(要管生命周期)中多实例场景(多个数据源、多个设备)
分片锁按 key 哈希到 N 把锁高中高中高大表、缓存、键值存储
无锁原子操作 / 单生产者-单消费者队列最高取决于正确性高标志位、计数器、环形缓冲

三条经验:

  1. 粒度提升是用复杂度换并发度,不是免费的午餐
  2. 每一级都需要上一级的纪律作为支撑——细粒度锁更容易出锁顺序问题
  3. 大多数系统不需要走到最右列,分片锁往往就是性价比拐点

四、对照一:全局大锁——整帧渲染

这是最典型、也最容易造成"偶发卡顿"的大锁。

/* 一把锁保护整帧渲染 */
void render_frame(struct view_tree *root)
{
    lock_take(&g_render_lock, WAIT_FOREVER);

    layout(root);          /* 约 5 ms:计算每个视图的位置尺寸 */
    draw_tree(root);       /* 约 12 ms:遍历视图树,逐节点绘制 */
    flush_to_screen();     /* 约 3 ms:把脏区推到显示硬件 */

    lock_release(&g_render_lock);
}

看起来没问题:渲染过程确实要保证一致性,不能画到一半被改。

问题在于这把锁同时也挡住了谁:

被挡住的操作后果
界面变更请求(改文字、改颜色)排队等 20ms,用户连续操作时请求堆积
输入事件处理(触摸)触摸响应延迟,主观感受"点不动"
其他需要读共享状态的线程全部串行化

于是现象变成:CPU 不高,但延迟大。因为大家在等,不是在算。

改进思路:把锁从"绘制时长"缩到"取快照时长"

/* 改进:只保护共享状态的读取,绘制阶段用私有快照 */
void render_frame(struct view_tree *root)
{
    struct draw_list *list;

    /* 第一段:短锁,只做"读共享状态 → 生成本帧私有快照" */
    if (lock_take(&g_state_lock, WAIT_MS(5)) != OK) {
        return;                       /* 拿不到就跳过本帧,不阻塞 */
    }
    list = build_draw_list(root);     /* 约 1 ms */
    lock_release(&g_state_lock);

    /* 第二段:无锁,操作的是本帧私有数据 */
    draw_draw_list(list);             /* 约 12 ms */
    flush_to_screen();                /* 约 3 ms */

    free_draw_list(list);
}

关键变化:

  • 持锁时间从 20ms 降到 1ms(20 倍)
  • 绘制阶段不再阻塞任何界面变更——变更只是"下一帧生效"
  • 引入了快照边界这个概念:临界区到快照为止,之后数据是私有的

这个模式在渲染、日志、统计、协议打包里反复出现,值得抽象成固定套路:

"读共享状态 → 生成私有副本 → 锁外慢处理"

代价是什么?内存(多一份快照)和延迟(变更下一帧才可见)。这是可以接受的工程权衡。

变体:命令队列

如果快照成本也高,可以进一步改成"生产者投递命令、消费者批量执行":

void ui_set_text(widget_t *w, const char *s)
{
    struct cmd c = { .type = CMD_SET_TEXT, .target = w, .arg = strdup(s) };
    ring_push(&g_cmd_queue, &c);      /* 无锁队列,单生产者单消费者 */
}

/* 渲染线程每帧开头批量消费 */
void render_frame(void)
{
    struct cmd c;
    while (ring_pop(&g_cmd_queue, &c) == OK) {
        apply_cmd(&c);                /* 只有渲染线程改视图,天然无竞争 */
    }
    draw_tree();
    flush_to_screen();
}

此时根本不需要锁——所有修改都收敛到渲染线程,跨线程只传递命令。


五、对照二:对象锁——每个数据源一把

场景:系统同时从多个数据源采集数据(总线设备、传感器、配置),每个数据源各自有一份数据结构。

class DataParser {
public:
    void update(const uint8_t *frame, size_t len)
    {
        std::lock_guard<std::mutex> guard(m_dataMutex);   /* 每个实例一把 */
        parse_into(m_data, frame, len);
    }

    Value get(FieldId id) const
    {
        std::lock_guard<std::mutex> guard(m_dataMutex);
        return m_data[id];
    }

private:
    mutable std::mutex m_dataMutex;
    DataMap m_data;
};

收益:不同数据源的解析互不阻塞。A 源解析耗时 2ms,不再拖住 B 源的读取。

代价与坑:

坑说明
锁数量膨胀N 个实例 = N 把锁,每个都要正确初始化
生命周期对象析构时锁还在被其他线程等待 → 必须先"停止服务"再销毁
初始化竞态见下方
内存开销每把锁都有控制块,几十字节到上百字节,实例成千上万时不可忽略

对象锁最隐蔽的坑:初始化竞态

DataParser *get_parser(int id)
{
    if (g_parsers[id] == nullptr) {                 /* ① 无锁检查 */
        std::lock_guard<std::mutex> guard(g_initMutex);
        if (g_parsers[id] == nullptr) {             /* ② 双检 */
            g_parsers[id] = new DataParser();       /* ③ 发布 */
        }
    }
    return g_parsers[id];                           /* ④ 无锁返回 */
}

这就是著名的双重检查锁定(DCLP)。它在单核 MCU 上有两个真实风险:

  1. 编译器/CPU 重排:③ 的"分配内存 → 构造对象 → 写指针"可能被重排成"写指针 → 构造对象",此时另一个线程在 ④ 拿到的是未构造完成的对象
  2. 中断打断:即使没有重排,③ 执行到一半被中断,ISR 若调用同一函数就会拿到半成品

正确做法,按优先级从高到低:

/* 方案 A:启动阶段单线程一次性初始化(最简单也最可靠) */
void app_init(void)
{
    for (int i = 0; i < PARSER_NUM; i++)
        g_parsers[i] = parser_create(i);     /* 此时只有一个线程 */
}

/* 方案 B:静态对象,零运行时初始化 */
static DataParser g_parsers[PARSER_NUM];     /* 全局静态,构造在 main 之前完成 */

/* 方案 C:必须延迟初始化时,用原子发布 + 获取屏障 */
DataParser *p = atomic_load_acquire(&g_parsers[id]);
if (p == NULL) {
    p = parser_create(id);
    atomic_store_release(&g_parsers[id], p);   /* 保证构造完成后才可见 */
}

结论:对象锁本身不复杂,"何时创建锁、何时发布对象"才是事故高发区。


六、对照三:分片锁——大表并发

场景:一张被多线程频繁读写的键值表、缓存、路由表。

#define SHARD_N 16

typedef struct {
    lock_t  lock;
    map_t   map;
} shard_t;

typedef struct {
    shard_t sh[SHARD_N];
} sharded_map_t;

/* 哈希到分片:乘法散列取高位,避免低位规律性 */
static inline shard_t *shard_of(sharded_map_t *m, uint32_t key)
{
    uint32_t h = key * 2654435761u;      /* Knuth 乘法散列 */
    return &m->sh[h >> 28];              /* 取高 4 位 → 0..15 */
}

int smap_put(sharded_map_t *m, uint32_t key, const void *val)
{
    shard_t *s = shard_of(m, key);
    lock_take(&s->lock, WAIT_FOREVER);
    int rc = map_put(&s->map, key, val);
    lock_release(&s->lock);
    return rc;
}

收益:不同分片的操作完全并行,理论并发度 = 分片数。

三个必须知道的边界:

  1. 分片数不是越多越好:每把锁有内存开销,且分片过多会让"跨分片操作"变复杂(遍历、统计、原子批量更新)
  2. 热点 key 依然串行:所有 key 都落在同一分片时,等于退回单锁。分片解决的是"不同 key 的并发",不解决"同一 key 的竞争"
  3. 跨分片原子性消失:需要同时改两个分片时,你必须按固定顺序加锁,否则引入死锁

同一 key 竞争怎么解?

分片无能为力,得换思路:

思路做法适用
减少临界区把"计算"移出锁,锁内只做指针赋值通用
批量化累积多次写,一次提交写多读少
写合并只保留最后一个值(latest-write-wins)状态类数据
无锁结构原子指针交换 + 副本读多写少、数据小

读多写少的经典做法是"副本替换":

/* 读多写少:读者完全无锁 */
config_t *g_cfg;      /* 原子指针 */

config_t *config_get(void)
{
    return atomic_load(&g_cfg);          /* 单条原子读 */
}

void config_update(const config_t *newv)
{
    config_t *copy = config_clone(newv);
    config_t *old  = atomic_exchange(&g_cfg, copy);
    config_retire(old);                  /* 等所有读者退出后再释放 */
}

注意 config_retire 这一步不能省:读者可能在 config_get() 返回后被抢占,此时旧对象还在被引用。单核上可以用引用计数或"延迟回收"(等到所有任务都经历过一次调度点后释放)。


七、什么时不该优化粒度

这一节和前面同样重要。细粒度锁有真实代价,盲目优化会引入更难查的 bug。

不该优化的信号原因
争抢概率很低一年撞不上几次,细化只增加代码量和漏锁风险
临界区内操作极短(小于锁开销的 10 倍)换锁的开销可能大于等待的开销
锁顺序难以固定细粒度天然需要多锁,多锁就要有全局顺序,否则死锁
团队规模小、代码变动少维护成本高于收益
没有测量数据凭感觉优化粒度,是事故的主要来源

一句口诀:

先测量争抢,再谈粒度。没有数据的优化都是猜测。


八、如何测量锁争抢(可落地)

给锁套一层薄封装,统计四个指标:

typedef struct {
    uint32_t take_cnt;          /* 加锁总次数 */
    uint32_t contended_cnt;     /* 需要等待的次数(争抢次数) */
    uint64_t wait_sum_us;       /* 累计等待时长 */
    uint64_t wait_max_us;       /* 单次最长等待 */
    uint64_t hold_sum_us;       /* 累计持有时长 */
    uint64_t hold_max_us;       /* 单次最长持有 */
} lock_stat_t;

static inline void lock_take_prof(lock_stat_t *st, lock_t *l)
{
    uint64_t t0 = now_us();

    if (lock_try(l) != OK) {                 /* 先试一次:失败即说明有争抢 */
        st->contended_cnt++;
        lock_take(l, WAIT_FOREVER);
    }

    uint64_t t1 = now_us();
    st->take_cnt++;
    st->wait_sum_us += (t1 - t0);
    if ((t1 - t0) > st->wait_max_us) {
        st->wait_max_us = t1 - t0;
    }
    st->hold_start = t1;                     /* 简化示意 */
}

static inline void lock_release_prof(lock_stat_t *st, lock_t *l)
{
    uint64_t hold = now_us() - st->hold_start;
    st->hold_sum_us += hold;
    if (hold > st->hold_max_us) {
        st->hold_max_us = hold;
    }
    lock_release(l);
}

指标怎么读

指标计算判读
争抢率contended_cnt / take_cnt> 5% 说明粒度或临界区长度有问题
平均等待wait_sum_us / contended_cnt与业务帧预算对比
最大等待wait_max_us比平均值重要得多:平均值漂亮但最大值超预算,就是偶发卡顿的元凶
最大持有hold_max_us直接对标"最坏情况响应时间"
持有/等待比hold_sum / wait_sum远大于 1 时,说明瓶颈在持锁者而非锁本身

最重要的一条:偶发问题看最大值和长尾,不要看平均值。平均值会把一次 200ms 的阻塞稀释到看不见,而用户恰恰就记住了那 200ms。


九、粒度决策清单

在动代码之前,逐条自问:

  1. 是否测过争抢率? 没有数据就不优化
  2. 临界区里有没有可以移出去的计算? 通常能砍掉 80% 的持有时间
  3. 锁保护的是"数据"还是"流程"? 保护流程往往过宽
  4. 能否用快照/命令队列消灭锁?单消费者模型下常常可以
  5. 细化后锁顺序是否全局唯一? 不能固定就别细化
  6. 锁对象的生命周期是否覆盖所有使用者? 尤其析构/重启场景
  7. 是否有"热点 key"?分片解决不了它
  8. 临界区里是否调用了可能阻塞的函数?日志、内存分配、IO 都会毁掉一切

Logo

社区规范:仅讨论OpenHarmony相关问题。

更多推荐