50、Redis发布订阅：PUBLISH、SUBSCRIBE、PSUBSCRIBE、UNSUBSCRIBE、模式匹配

发布订阅模式，说白了就是「广播」机制。一个消息发出去，所有订阅了该频道的客户端都能收到。我在做实时消息推送系统时，这个功能帮了大忙。

Redis的发布订阅（Pub/Sub）是一种消息通信模式。发送者（Publisher）发送消息，订阅者（Subscriber）接收消息。它们之间不需要直接建立连接，通过频道（Channel）作为中介。

核心概念：发布者只管发，订阅者只管收，双方解耦。你想想看，这在微服务架构里多好用？

5.1 基本命令：PUBLISH 和 SUBSCRIBE

先看最基础的两个命令。PUBLISH用于发布消息，SUBSCRIBE用于订阅频道。

// 订阅者终端1
127.0.0.1:6379> SUBSCRIBE news:sports
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "news:sports"
3) (integer) 1

// 发布者终端2
127.0.0.1:6379> PUBLISH news:sports "湖人队夺冠了！"
(integer) 1

// 订阅者终端1 收到消息
1) "message"
2) "news:sports"
3) "湖人队夺冠了！"

嗯，这里要注意：SUBSCRIBE一旦执行，当前连接就进入「订阅模式」。什么意思？就是你不能在这个连接上再执行其他Redis命令了，只能等消息。我刚开始用的时候，老想着订阅完再查个数据，结果一直报错。

5.2 模式匹配：PSUBSCRIBE

PSUBSCRIBE支持通配符订阅。比如你想订阅所有新闻频道，不用一个个列出来。

// 订阅所有news开头的频道
127.0.0.1:6379> PSUBSCRIBE news:*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "news:*"
3) (integer) 1

// 发布消息到 news:sports
127.0.0.1:6379> PUBLISH news:sports "NBA总决赛开始了"
(integer) 1

// 订阅者收到
1) "pmessage"    // 注意这里是pmessage，不是message
2) "news:*"      // 匹配的模式
3) "news:sports" // 实际频道
4) "NBA总决赛开始了"

模式匹配支持的通配符：

  • *：匹配任意多个字符
  • ?：匹配一个字符
  • [ab]：匹配a或b

个人经验：我在项目中用PSUBSCRIBE订阅了 order:* 模式，所有订单状态变更都通过不同频道发布。这样新增订单类型时，不需要修改订阅代码。

5.3 取消订阅：UNSUBSCRIBE 和 PUNSUBSCRIBE

有订阅就得有退订。UNSUBSCRIBE退订具体频道，PUNSUBSCRIBE退订模式。

// 退订单个频道
UNSUBSCRIBE news:sports

// 退订所有频道（不指定参数）
UNSUBSCRIBE

// 退订模式
PUNSUBSCRIBE news:*

退订操作在C语言客户端里特别重要。为什么？因为如果你不主动退订，连接一直保持，资源就浪费了。我曾经有个服务，连接数一直涨，查了半天才发现是订阅后忘了退订。

5.4 C语言实战：完整示例

下面我用hiredis库写一个完整的发布订阅示例。hiredis是Redis官方推荐的C语言客户端。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <hiredis/hiredis.h>
#include <hiredis/async.h>
#include <hiredis/adapters/libevent.h>

// 订阅回调函数
void subscribe_callback(redisAsyncContext *c, void *reply, void *privdata) {
    redisReply *r = reply;
    if (r == NULL) return;

    // 判断消息类型
    if (r->type == REDIS_REPLY_ARRAY) {
        // 订阅确认消息
        if (strcmp(r->element[0]->str, "subscribe") == 0) {
            printf("成功订阅频道: %s\n", r->element[1]->str);
        }
        // 普通消息
        else if (strcmp(r->element[0]->str, "message") == 0) {
            printf("[频道: %s] 消息: %s\n", 
                   r->element[1]->str, r->element[2]->str);
        }
        // 模式匹配消息
        else if (strcmp(r->element[0]->str, "pmessage") == 0) {
            printf("[模式: %s] [频道: %s] 消息: %s\n", 
                   r->element[1]->str, r->element[2]->str, r->element[3]->str);
        }
    }
}

// 发布消息函数
void publish_message(redisContext *c, const char *channel, const char *msg) {
    redisReply *reply = redisCommand(c, "PUBLISH %s %s", channel, msg);
    if (reply == NULL) {
        printf("发布失败: %s\n", c->errstr);
        return;
    }
    printf("消息已发布，%d个订阅者收到\n", reply->integer);
    freeReplyObject(reply);
}

int main() {
    // 创建同步连接用于发布
    redisContext *pub_ctx = redisConnect("127.0.0.1", 6379);
    if (pub_ctx == NULL || pub_ctx->err) {
        printf("连接Redis失败\n");
        return 1;
    }

    // 创建事件库和异步连接用于订阅
    struct event_base *base = event_base_new();
    redisAsyncContext *sub_ctx = redisAsyncConnect("127.0.0.1", 6379);
    if (sub_ctx->err) {
        printf("异步连接失败: %s\n", sub_ctx->errstr);
        return 1;
    }

    // 关联事件库
    redisLibeventAttach(sub_ctx, base);

    // 设置回调
    redisAsyncSetConnectCallback(sub_ctx, NULL);
    redisAsyncSetDisconnectCallback(sub_ctx, NULL);

    // 订阅频道和模式
    redisAsyncCommand(sub_ctx, subscribe_callback, NULL, 
                      "SUBSCRIBE news:sports news:tech");
    redisAsyncCommand(sub_ctx, subscribe_callback, NULL, 
                      "PSUBSCRIBE alert:*");

    printf("订阅成功，等待消息...\n");

    // 模拟发布消息
    sleep(1);
    publish_message(pub_ctx, "news:sports", "NBA总决赛湖人胜");
    publish_message(pub_ctx, "news:tech", "Redis 7.0发布");
    publish_message(pub_ctx, "alert:system", "CPU使用率超过90%");

    // 运行事件循环
    event_base_dispatch(base);

    // 清理
    redisFree(pub_ctx);
    redisAsyncDisconnect(sub_ctx);
    return 0;
}

避坑指南：我曾经在生产环境遇到一个问题——订阅者断线重连后，丢失了断线期间的消息。Redis的Pub/Sub不持久化消息，这是设计如此。如果你需要消息持久化，考虑用Redis Stream或者消息队列。

5.5 发布订阅模式流程图

Redis发布订阅模式流程图 发布者 (Publisher) Redis 服务器 频道: news:sports 订阅者 A 订阅者 B 订阅者 C (模式) PUBLISH SUBSCRIBE SUBSCRIBE PSUBSCRIBE 发布者：发送消息 订阅者：接收消息 模式订阅：PSUBSCRIBE Redis服务器：消息分发中心

5.6 实际应用场景

我在几个项目中用过Redis Pub/Sub，分享几个典型场景：

场景 频道设计 说明
实时通知 notify:user:{id} 用户登录、订单状态变更等实时推送
日志收集 log:{level} 不同级别日志分发到不同消费者
配置更新 config:reload 配置变更时通知所有服务重新加载
监控告警 alert:{type} 系统指标异常时触发告警

我的建议：频道命名要有层次结构，用冒号分隔。比如 order:create、order:pay，这样用PSUBSCRIBE order:* 就能订阅所有订单相关消息。

5.7 注意事项与最佳实践

  • 消息不持久化：订阅者断开期间的消息会丢失。需要持久化？用Redis Stream。
  • 订阅者数量：每个订阅者独占一个连接。大量订阅者会消耗连接资源。
  • 消息积压：如果订阅者处理慢，消息会在Redis缓冲区积压，可能导致内存暴涨。
  • 网络分区：Redis Pub/Sub没有确认机制，网络问题可能导致消息丢失。

我曾经踩过的坑：有个项目用Pub/Sub做订单状态同步，结果网络抖动导致丢消息，订单状态不一致。后来改成Redis Stream + 消费者组，才解决了问题。Pub/Sub适合实时性要求高、允许少量丢消息的场景。

Redis发布订阅模式，说白了就是「发广播」。简单、高效，但也要清楚它的局限性。选对场景，它就是一把利器。

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

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