当前位置： 首页 > news >正文

Redis实现数据传输简介

news 2025/7/31 8:57:13

Redis 可以通过其内置的数据结构和消息机制实现高效的数据传输，尤其适合跨进程、跨服务的实时通信。以下是详细的使用方法和示例，涵盖 队列、发布/订阅、流（Stream） 三种主要模式，并附上 C++ 实现代码。

1. 使用 Redis List 实现队列传输

原理

  • 生产者：通过 LPUSH 或 RPUSH 将数据插入队列。
  • 消费者：通过 RPOP 或 LPOP 从队列取出数据。
  • 阻塞操作：BLPOP/BRPOP 可避免轮询，节省 CPU 资源。

适用场景

  • 任务队列（如异步任务处理）。
  • 日志收集。
  • 跨服务数据同步。

示例代码（C++）

(1) 生产者（发送数据）
#include <hiredis/hiredis.h>
#include <iostream>
void producer() {
redisContext *c = redisConnect("127.0.0.1", 6379);
if (!c || c->err) {
std::cerr << "Connection error: " << (c ? c->errstr : "unknown") << std::endl;
return;
}
// 插入数据到队列头部（LPUSH）
redisReply *reply = (redisReply *)redisCommand(c, "LPUSH my_queue data1");
freeReplyObject(reply);
reply = (redisReply *)redisCommand(c, "LPUSH my_queue data2");
freeReplyObject(reply);
std::cout << "Producer: Sent data to queue." << std::endl;
redisFree(c);
}
(2) 消费者（接收数据）
void consumer() {
redisContext *c = redisConnect("127.0.0.1", 6379);
if (!c || c->err) {
std::cerr << "Connection error: " << (c ? c->errstr : "unknown") << std::endl;
return;
}
// 阻塞式获取数据（BRPOP，超时时间 0 表示无限等待）
redisReply *reply = (redisReply *)redisCommand(c, "BRPOP my_queue 0");
if (reply && reply->type == REDIS_REPLY_ARRAY && reply->elements == 2) {
std::cout << "Consumer: Received -> " << reply->element[1]->str << std::endl;
} else {
std::cout << "Consumer: Queue is empty or error occurred." << std::endl;
}
freeReplyObject(reply);
redisFree(c);
}
(3) 完整流程
int main() {
// 启动生产者（终端1）
// producer();
// 启动消费者（终端2）
consumer();
return 0;
}

2. 使用 Redis Pub/Sub 实现发布/订阅传输

原理

  • 发布者：通过 PUBLISH 向频道（channel）发送消息。
  • 订阅者：通过 SUBSCRIBE 监听频道并接收消息。
  • 特点：
    • 实时性强，但消息不持久化（离线订阅者会丢失消息）。
    • 支持多对多通信（一个频道可被多个订阅者监听）。

适用场景

  • 实时通知（如用户上线提醒）。
  • 事件广播（如系统状态变更）。

示例代码（C++）

(1) 发布者
#include <hiredis/hiredis.h>
#include <iostream>
void publisher() {
redisContext *c = redisConnect("127.0.0.1", 6379);
if (!c || c->err) {
std::cerr << "Connection error: " << (c ? c->errstr : "unknown") << std::endl;
return;
}
// 发布消息到频道
redisReply *reply = (redisReply *)redisCommand(c, "PUBLISH my_channel Hello, Redis!");
if (reply && reply->type == REDIS_REPLY_INTEGER) {
std::cout << "Publisher: Sent message to " << reply->integer << " subscribers." << std::endl;
}
freeReplyObject(reply);
redisFree(c);
}
(2) 订阅者
#include <hiredis/hiredis.h>
#include <iostream>
#include <thread>
#include <atomic>
std::atomic<bool> running(true);
void subscriber() {
redisContext *c = redisConnect("127.0.0.1", 6379);
if (!c || c->err) {
std::cerr << "Connection error: " << (c ? c->errstr : "unknown") << std::endl;
return;
}
// 订阅频道（hiredis 需要轮询获取消息）
while (running) {
redisReply *reply = (redisReply *)redisCommand(c, "SUBSCRIBE my_channel");
if (reply && reply->type == REDIS_REPLY_ARRAY && reply->elements == 3) {
std::string type(reply->element[0]->str);
if (type == "message") {
std::cout << "Subscriber: Received -> " << reply->element[2]->str << std::endl;
}
}
freeReplyObject(reply);
// 实际项目中建议用异步库（如 cpp_redis）或事件驱动模型
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
redisFree(c);
}
(3) 完整流程（需多线程）
#include <thread>
int main() {
// 启动订阅者（终端1）
std::thread sub_thread(subscriber);
// 启动发布者（终端2）
publisher();
// 停止订阅者（实际项目中可通过信号或条件变量控制）
std::this_thread::sleep_for(std::chrono::seconds(1));
running = false;
sub_thread.join();
return 0;
}

注意：hiredis 的 SUBSCRIBE 需要轮询，实际项目中建议使用异步库（如 cpp_redis）或事件驱动模型。

3. 使用 Redis Stream 实现消息队列（高级）

原理

  • 生产者：通过 XADD 向流（stream）添加消息。
  • 消费者：通过 XREAD 或消费者组（XGROUP CREATE + XREADGROUP）读取消息。
  • 特点：
    • 消息持久化（可回溯）。
    • 支持消费者组（避免重复消费）。
    • 类似 Kafka 的轻量级实现。

适用场景

  • 需要消息确认和重试的场景（如订单处理）。
  • 高可靠性要求的日志流。

示例代码（C++）

(1) 生产者
#include <hiredis/hiredis.h>
#include <iostream>
void stream_producer() {
redisContext *c = redisConnect("127.0.0.1", 6379);
if (!c || c->err) {
std::cerr << "Connection error: " << (c ? c->errstr : "unknown") << std::endl;
return;
}
// 添加消息到流（* 表示自动生成 ID）
redisReply *reply = (redisReply *)redisCommand(c, "XADD mystream * field1 value1 field2 value2");
if (reply && reply->type == REDIS_REPLY_STRING) {
std::cout << "Stream Producer: Added message with ID -> " << reply->str << std::endl;
}
freeReplyObject(reply);
redisFree(c);
}
(2) 消费者（简单读取）
void stream_consumer_simple() {
redisContext *c = redisConnect("127.0.0.1", 6379);
if (!c || c->err) {
std::cerr << "Connection error: " << (c ? c->errstr : "unknown") << std::endl;
return;
}
// 读取流中的最新消息（COUNT 1 表示最多 1 条）
redisReply *reply = (redisReply *)redisCommand(c, "XREAD COUNT 1 STREAMS mystream 0");
if (reply && reply->type == REDIS_REPLY_ARRAY && reply->elements == 1) {
redisReply *stream_array = reply->element[0];
if (stream_array->type == REDIS_REPLY_ARRAY && stream_array->elements == 2) {
std::string stream_name(stream_array->element[0]->str);
redisReply *messages = stream_array->element[1];
if (messages->type == REDIS_REPLY_ARRAY && messages->elements > 0) {
std::cout << "Stream Consumer: Received message ID -> "
<< messages->element[0]->element[0]->str << std::endl;
}
}
}
freeReplyObject(reply);
redisFree(c);
}
(3) 消费者组（高级用法）
// 创建消费者组（只需执行一次）
void create_consumer_group() {
redisContext *c = redisConnect("127.0.0.1", 6379);
redisCommand(c, "XGROUP CREATE mystream mygroup 0 MKSTREAM");
redisFree(c);
}
// 消费者组读取消息
void stream_consumer_group() {
redisContext *c = redisConnect("127.0.1", 6379);
if (!c || c->err) return;
// 从消费者组读取消息（BLOCK 0 表示无限阻塞）
redisReply *reply = (redisReply *)redisCommand(
c, "XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS mystream >"
);
// 处理消息...
freeReplyObject(reply);
redisFree(c);
}

4. 三种传输方式对比

方式持久化消费者组延迟适用场景
List否否低简单队列、任务调度
Pub/Sub否否极低实时通知、事件广播
Stream是是低高可靠性消息队列

5. 最佳实践建议

  1. 简单队列：优先用 List + BRPOP。
  2. 实时通知：用 Pub/Sub，但需容忍消息丢失。
  3. 高可靠性队列：用 Stream + 消费者组。
  4. 性能优化：
    • 批量操作（MGET/MSET）。
    • 使用 Pipeline 减少网络往返。
  5. 错误处理：
    • 检查 Redis 命令返回值（如 NULL 或错误类型）。
    • 重试机制（如指数退避）。

总结

  • Redis List：适合简单队列，实现容易。
  • Redis Pub/Sub：适合实时通知，但消息不持久化。
  • Redis Stream：适合高可靠性消息队列，功能最强大。
  • C++ 客户端：推荐 cpp_redis 或 redis-plus-plus 简化异步操作。

根据业务需求选择合适的模式，并结合错误处理和性能优化，即可高效使用 Redis 进行数据传输！

http://www.lryc.cn/news/604803.html

相关文章：

  • Kubernetes高级调度02
  • Elasticsearch 索引管理 API 实战：涵盖创建、查看、修改、删除及别名
  • Redis 面试全解析：从数据结构到集群架构（含实战解决方案）
  • 设计模式之单例模式及其在多线程下的使用
  • 【C#】DevExpress.XtraEditors.MemoEdit memoEditLog控件讲解
  • Rabbitmq中常见7种模式介绍
  • pytorch小记（三十三）：PyTorch 使用 TensorBoard 可视化训练过程（含完整示例）
  • 用 Go Typed Client 快速上手 Elasticsearch —— 从建索引到聚合的完整实战
  • 8.Linux ： 日志的管理与时钟同步的配置
  • Rabbit MQ的消息模式-Java原生代码
  • YOLO-01目标检测基础
  • 02 基于sklearn的机械学习-特征降维（特征选择、PCA）、KNN算法、模型选择与调优（交叉验证、朴素贝叶斯算法、拉普拉斯平滑）
  • Android调用python库和方法的实现
  • YOLOv5u：无锚点检测的革命性进步
  • android-PMS-创建新用户流程
  • 舆情监测专员需要哪些常用软件工具?
  • 基于 Hadoop 生态圈的数据仓库实践 —— OLAP 与数据可视化（一）
  • 论文Review 3DGSSLAM S3PO-GS | ICCV 2025 港科广出品！| 高效快速的3DGSSLAM！
  • sqli-labs：Less-1关卡详细解析
  • CMS框架漏洞
  • 3D Web轻量化引擎HOOPS Communicator数据处理与流式加载能力概述
  • 【音视频】WebRTC-Web 音视频采集与播放
  • 【预判一手面试问题：排序】
  • 依托客户满意度分析协助企业精准把握市场趋势​（满意度调查）
  • 智能AI医疗物资/耗材管理系统升级改造方案分析
  • InfluxDB 与 Java 框架集成：Spring Boot 实战（二）
  • VSCode插件开发完整教程：从零开始创建文件导出插件
  • Python 程序设计讲义（37）：字符串的处理方法——设置字符串居中显示：center() 方法
  • 图像平滑处理
  • 9.项目起步（3）