18. 消息路由与分发：基于内容的路由、主题路由、通配符匹配、路由表设计

消息路由，说白了就是决定一条消息该往哪儿送。你想想看，一个系统里可能有几十个模块，有的关心温度数据，有的只处理按键事件，有的专门负责日志。如果每条消息都广播给所有人，那CPU就光忙着做无用功了。

我早年做的一个物联网网关项目，就吃过这个亏。当时所有传感器数据都走同一个队列，每个模块收到消息后自己判断要不要处理。结果呢？日志模块天天被温度数据刷屏，关键的控制指令反而经常延迟。后来我痛下决心，把路由机制加进去，问题才彻底解决。

18.1 基于内容的路由

基于内容的路由，是最直观的一种方式。消息本身携带一些关键字段，路由器根据这些字段的值来决定投递目标。

举个例子，假设我们的消息结构长这样：

typedef struct {
    uint32_t msg_id;
    uint16_t type;      // 消息类型
    uint16_t priority;  // 优先级
    uint32_t src_id;    // 源模块ID
    uint32_t dst_id;    // 目标模块ID
    uint8_t  data[64];  // 消息体
} msg_t;

路由器可以这样写：

void route_by_content(msg_t *msg) {
    // 根据消息类型分发
    switch (msg->type) {
        case MSG_TYPE_TEMPERATURE:
            enqueue(TEMP_QUEUE, msg);
            break;
        case MSG_TYPE_HUMIDITY:
            enqueue(HUMID_QUEUE, msg);
            break;
        case MSG_TYPE_ALARM:
            // 报警消息要同时发给多个模块
            enqueue(ALARM_QUEUE, msg);
            enqueue(LOG_QUEUE, msg);
            enqueue(DISPLAY_QUEUE, msg);
            break;
        default:
            enqueue(DEFAULT_QUEUE, msg);
            break;
    }
}

我的经验：基于内容的路由适合消息类型比较固定的场景。如果类型经常变，switch-case会越写越长，维护起来很痛苦。这时候就该考虑主题路由了。

18.2 主题路由

主题路由，是给每条消息打上一个「主题标签」。订阅者只关心自己感兴趣的主题，路由器负责把消息推给所有订阅了该主题的模块。

主题通常用字符串表示，比如：

  • sensor/temperature — 温度传感器数据
  • sensor/humidity — 湿度传感器数据
  • system/alarm — 系统报警
  • user/input/key — 按键输入

我习惯用哈希表来管理主题与订阅者的映射关系：

#define MAX_SUBSCRIBERS 16
#define MAX_TOPIC_LEN   32

typedef struct subscriber_node {
    queue_id_t queue_id;                // 订阅者的队列ID
    struct subscriber_node *next;       // 链表指针
} sub_node_t;

typedef struct {
    char topic[MAX_TOPIC_LEN];          // 主题字符串
    sub_node_t *subscribers;            // 订阅者链表
    uint8_t count;                      // 订阅者数量
} topic_entry_t;

// 主题路由表（简化版）
topic_entry_t route_table[64];
int route_table_size = 0;

发布消息时，路由器查找主题，然后遍历订阅者链表：

void route_by_topic(const char *topic, msg_t *msg) {
    for (int i = 0; i < route_table_size; i++) {
        if (strcmp(route_table[i].topic, topic) == 0) {
            sub_node_t *node = route_table[i].subscribers;
            while (node) {
                enqueue(node->queue_id, msg);
                node = node->next;
            }
            return;
        }
    }
    // 没有找到订阅者，消息丢弃或走默认路径
}

注意：字符串比较在嵌入式系统里是有代价的。如果主题数量超过几十个，建议用哈希值代替字符串比较，或者用枚举值做主题ID。我曾经在一个项目里用纯字符串匹配，结果CPU占用率直接飙到30%，换成哈希后降到2%以下。

18.3 通配符匹配

通配符匹配，是主题路由的进阶玩法。订阅者可以用通配符来表达「一类主题」，而不是一个个去订阅。

常见的通配符有两种：

  • + — 匹配单层。比如 sensor/+/temperature 匹配 sensor/room1/temperature 和 sensor/room2/temperature
  • # — 匹配多层。比如 sensor/# 匹配所有以 sensor/ 开头的主题

实现通配符匹配，核心是一个递归函数：

int topic_match(const char *pattern, const char *topic) {
    // 递归终止条件
    if (*pattern == '\0' && *topic == '\0') return 1;
    if (*pattern == '#' && *(pattern+1) == '\0') return 1;  // #匹配剩余所有
    
    if (*pattern == '+') {
        // +匹配单层，跳过当前层级
        const char *next_slash = strchr(topic, '/');
        if (next_slash == NULL) {
            // 没有下一层了，检查pattern是否也结束
            return (*(pattern+1) == '\0') ? 1 : 0;
        }
        return topic_match(pattern+2, next_slash+1);  // 跳过 '/' 和当前层
    }
    
    if (*pattern == *topic) {
        return topic_match(pattern+1, topic+1);
    }
    
    return 0;  // 不匹配
}

有了这个匹配函数，路由器的查找逻辑就变成：

void route_with_wildcard(const char *topic, msg_t *msg) {
    for (int i = 0; i < route_table_size; i++) {
        if (topic_match(route_table[i].topic, topic)) {
            sub_node_t *node = route_table[i].subscribers;
            while (node) {
                enqueue(node->queue_id, msg);
                node = node->next;
            }
        }
    }
}

避坑指南：我曾经在通配符匹配上踩过一个坑。当时用递归实现，但主题层级很深（超过10层），导致栈溢出。后来改成迭代版本才稳定。如果你的系统栈空间有限，建议用循环+状态机来实现匹配，别用递归。

18.4 路由表设计

路由表是整个消息路由系统的核心数据结构。设计得好，查找快、更新方便；设计得烂，系统性能直接崩。

我总结了几种常见的路由表设计方案：

方案 优点 缺点 适用场景
线性表 实现简单，内存连续 查找O(n)，订阅多时慢 主题少（<20个）
哈希表 查找O(1)，速度快 需要处理哈希冲突，内存碎片 主题中等（20-200个）
字典树(Trie) 支持通配符，前缀匹配快 实现复杂，内存开销大 主题多且有层级关系
位图索引 查找极快，内存固定 主题数量受限，灵活性差 主题固定且数量少

我个人最常用的是哈希表+链地址法。哈希函数选简单的，比如：

uint32_t hash_topic(const char *topic) {
    uint32_t hash = 5381;
    int c;
    while ((c = *topic++)) {
        hash = ((hash << 5) + hash) + c;  // djb2算法
    }
    return hash % TABLE_SIZE;
}

路由表的更新操作也很关键。订阅和取消订阅都要保证原子性：

int subscribe(const char *topic, queue_id_t queue_id) {
    uint32_t idx = hash_topic(topic);
    // 查找是否已有该主题的条目
    topic_entry_t *entry = find_entry(idx, topic);
    if (entry == NULL) {
        // 新建条目
        entry = create_entry(topic);
        if (entry == NULL) return -1;  // 内存不足
        // 插入到哈希表
        entry->next = table[idx];
        table[idx] = entry;
    }
    // 检查是否已经订阅
    sub_node_t *node = entry->subscribers;
    while (node) {
        if (node->queue_id == queue_id) return 0;  // 已订阅
        node = node->next;
    }
    // 添加新订阅者
    sub_node_t *new_node = malloc(sizeof(sub_node_t));
    if (new_node == NULL) return -1;
    new_node->queue_id = queue_id;
    new_node->next = entry->subscribers;
    entry->subscribers = new_node;
    entry->count++;
    return 0;
}

我的建议：路由表设计时，一定要考虑「订阅者数量」和「主题数量」的比例。如果每个主题只有一两个订阅者，用线性表就够了。如果每个主题有几十个订阅者，那查找订阅者链表本身也会成为瓶颈，这时候可以考虑用数组代替链表，或者用跳表加速。

18.5 整体架构图

下面这张图展示了消息路由与分发的整体流程：

消息路由与分发架构 消息发布者 消息队列 消息路由器 基于内容 / 主题 / 通配符 路由表查找 分发决策 路由表 主题 → 订阅者列表 哈希表 / 字典树 通配符规则 订阅者 A 订阅者 B 订阅者 C 订阅者 D 消息流 订阅/取消订阅 路由器核心 路由表

从这张图可以看出，消息从发布者进入队列，路由器从队列中取出消息，根据路由表中的规则（基于内容、主题或通配符）决定投递目标，最终将消息分发给对应的订阅者。订阅者也可以随时发起订阅或取消订阅的请求，动态更新路由表。

嗯，这里要注意一点：路由器的处理速度必须跟上消息的到达速度。如果消息来得太快，路由器处理不过来，队列就会积压。我一般会在路由器前面加一个限流机制，或者用多级队列来缓冲。

好了，消息路由与分发的内容就讲到这里。核心就是三件事：定义好消息的「身份」（内容或主题），设计好路由表的数据结构，实现高效的匹配算法。把这三点做好，你的消息系统就能灵活应对各种业务场景了。

公众号：蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →