Redis 发布/订阅模式与消息队列

news2024/12/16 12:37:28

- - 1. Redis 发布/订阅机制
  - - 1.1 基本概念与实现
    - 1.2 C++ 代码实现
    - - 1.2.1 安装 hiredis
      - 1.2.2 发布者（Publisher）
      - 1.2.3 订阅者（Subscriber）
    - 1.3 发布/订阅应用场景
    - 1.4 高级特性与优化
  - 2. 使用 Redis 实现简单的消息队列
  - - 2.1 基本概念
    - 2.2 C++ 代码实现
    - - 2.2.1 生产者（Producer）
      - 2.2.2 消费者（Consumer）
    - 2.3 消息队列应用场景
    - 2.4 可靠性与持久性保障
  - 总结

Redis 提供了强大的消息传递机制，其中**发布/订阅（Pub/Sub）和消息队列（Message Queue）**是实现实时消息传递和任务调度的常用方式。这两种机制广泛应用于实时通知系统、事件驱动架构、日志收集和任务队列等场景。本文将深入探讨 Redis 的这两种机制，并结合 C++ 代码进行详细实现。

1. Redis 发布/订阅机制

1.1 基本概念与实现

概念：
Redis 的发布/订阅模式通过频道（Channel）来传递消息。生产者（发布者）将消息发布到一个或多个频道，消费者（订阅者）则订阅相应的频道以接收消息。发布者与订阅者之间解耦，发布者无需关心订阅者，订阅者也无需关心发布者。

1.2 C++ 代码实现

要在 C++ 中实现 Redis 发布/订阅机制，可以使用 hiredis 库，它提供了对 Redis 的 C 语言 API 支持。

1.2.1 安装 hiredis

首先，确保已安装 hiredis 库。可以通过以下命令在 Linux 上安装：

sudo apt-get install libhiredis-dev

1.2.2 发布者（Publisher）

发布者使用 PUBLISH 命令将消息发送到指定频道。以下是 C++ 示例代码：

#include <iostream>
#include <hiredis/hiredis.h>

int main() {
    // 连接到 Redis 服务器
    redisContext *c = redisConnect("127.0.0.1", 6379);
    if (c == NULL || c->err) {
        std::cerr << "Connection error: " << (c ? c->errstr : "can't allocate redis context") << std::endl;
        return 1;
    }

    // 发布消息
    const char *channel = "news";
    const char *message = "Redis 发布/订阅模式学习";

    redisReply *reply = (redisReply *)redisCommand(c, "PUBLISH %s %s", channel, message);
    std::cout << "发布消息到频道 '" << channel << "': " << message << std::endl;

    // 释放资源
    freeReplyObject(reply);
    redisFree(c);

    return 0;
}

说明：
这段代码连接到 Redis 服务器，并通过 PUBLISH 命令将消息 "Redis 发布/订阅模式学习" 发布到 news 频道。

1.2.3 订阅者（Subscriber）

订阅者使用 SUBSCRIBE 命令订阅一个或多个频道，并进入“订阅模式”，等待接收来自频道的消息。以下是订阅者的 C++ 实现代码：

#include <iostream>
#include <hiredis/hiredis.h>

void subscribeCallback(redisContext *c, void *reply, void *privdata) {
    redisReply *r = (redisReply *)reply;
    if (r->type == REDIS_REPLY_ARRAY && r->elements == 3) {
        std::cout << "收到消息: " << r->element[2]->str << std::endl;
    }
}

int main() {
    // 连接到 Redis 服务器
    redisContext *c = redisConnect("127.0.0.1", 6379);
    if (c == NULL || c->err) {
        std::cerr << "Connection error: " << (c ? c->errstr : "can't allocate redis context") << std::endl;
        return 1;
    }

    // 订阅频道
    const char *channel = "news";
    redisReply *reply = (redisReply *)redisCommand(c, "SUBSCRIBE %s", channel);
    std::cout << "已订阅频道: " << channel << std::endl;

    // 进入订阅模式，接收消息
    while (true) {
        redisGetReply(c, (void **)&reply);
        subscribeCallback(c, reply, nullptr);
        freeReplyObject(reply);
    }

    redisFree(c);

    return 0;
}

说明：
这段代码通过 SUBSCRIBE 命令订阅 news 频道，进入“订阅模式”，并通过 subscribeCallback 函数实时处理接收到的消息。

1.3 发布/订阅应用场景

实时通知系统：如消息推送、公告发布、实时聊天。
事件驱动架构：微服务或模块通过订阅感兴趣的事件，实时响应其他服务的变化。
日志收集：通过订阅日志频道，将日志数据实时发送至日志分析系统。

1.4 高级特性与优化

模式匹配订阅（PSUBSCRIBE）：支持使用通配符订阅多个频道，例如，订阅所有以 news: 开头的频道。
性能优化：
- 连接池管理：使用连接池管理 Redis 客户端连接，避免每次订阅时都创建新的连接。
- 分布式发布/订阅：通过 Redis 分布式部署，支持高并发发布/订阅。

2. 使用 Redis 实现简单的消息队列

2.1 基本概念

Redis 的 List 数据结构非常适合用于实现消息队列。常用的命令包括：

LPUSH：将消息推入队列。
RPUSH：将消息推入队列的另一端。
BRPOP：从队列的另一端阻塞地弹出消息。

2.2 C++ 代码实现

2.2.1 生产者（Producer）

生产者将消息推入队列。以下是 C++ 示例代码：

#include <iostream>
#include <hiredis/hiredis.h>

int main() {
    // 连接到 Redis 服务器
    redisContext *c = redisConnect("127.0.0.1", 6379);
    if (c == NULL || c->err) {
        std::cerr << "Connection error: " << (c ? c->errstr : "can't allocate redis context") << std::endl;
        return 1;
    }

    // 将消息推送到队列
    const char *queue = "task_queue";
    redisReply *reply = (redisReply *)redisCommand(c, "LPUSH %s %s", queue, "Task 1");
    std::cout << "任务 'Task 1' 已加入队列" << std::endl;

    // 释放资源
    freeReplyObject(reply);
    redisFree(c);

    return 0;
}

说明：
这段代码将任务 "Task 1" 推入队列 task_queue，作为生产者的一部分。

2.2.2 消费者（Consumer）

消费者使用 BRPOP 命令从队列中阻塞地获取消息。以下是消费者的 C++ 示例代码：

#include <iostream>
#include <hiredis/hiredis.h>

int main() {
    // 连接到 Redis 服务器
    redisContext *c = redisConnect("127.0.0.1", 6379);
    if (c == NULL || c->err) {
        std::cerr << "Connection error: " << (c ? c->errstr : "can't allocate redis context") << std::endl;
        return 1;
    }

    // 阻塞地从队列获取任务
    const char *queue = "task_queue";
    redisReply *reply = (redisReply *)redisCommand(c, "BRPOP %s 0", queue);
    if (reply->type == REDIS_REPLY_ARRAY && reply->elements == 2) {
        std::cout << "处理任务: " << reply->element[1]->str << std::endl;
    }

    // 释放资源
    freeReplyObject(reply);
    redisFree(c);

    return 0;
}