技术栈：C / C++17 · Redis 7.x · hiredis 1.1.0 · redis-plus-plus（sw/redis++）1.3.x
配套环境：Windows 11 / Linux（本文示例以 CMake + vcpkg/FetchContent 集成为主）
1. 背景
1.1 工业数采的双路径存储模型
工业数据采集（CNC / PLC / 传感器）的落地链路通常分为两条路径：
- 历史流：设备点位数据以消息形式进入 Kafka，最终落入 TDengine 等时序数据库，用于离线分析、告警回溯、OEE 统计。特点是数据量大、只追加、不需要毫秒级响应。
- 实时层：设备最新值、命令下发队列、多网关互斥锁、实时告警等场景需要毫秒级读写，这类数据用内存缓存（Redis）承载，再定期/按需下沉到历史存储。
在这条链路中，Go 侧通常使用 go-redis（已有专篇），而C++ 采集网关（尤其是直接对接 FANUC FOCAS、三菱 MC 协议、西门子 snap7 的采集端）也需要一个可靠、高性能的 Redis 客户端来维护"点位最新值缓存"、"命令队列"、"网关分布式锁"。
1.2 为什么选择 hiredis 与 redis-plus-plus
C/C++ 生态的 Redis 客户端主要有三条路线：
| 库 | 语言 | 定位 | 特点 |
|---|---|---|---|
| hiredis | C | 官方推荐的底层客户端 | 极简、零依赖、同步+异步双模型、RESP 解析器独立可用；是 redis-plus-plus、很多框架（如 Drogon 的 redis 插件）的底层 |
| redis-plus-plus（sw/redis++） | C++17 | hiredis 之上的现代 C++ 封装 | Redis/RedisCluster 类、命令族方法、Optional 返回值、Pipeline/Transaction、连接池、订阅回调、Lua 脚本 |
| cpp-redis / acre | C++ | 第三方封装 | 生态小，维护不活跃 |
hiredis 适合对性能与控制力要求极高的场景（手写事件循环、嵌入式采集器）；redis-plus-plus 适合工程化程度高的 C++ 项目（自动连接池、类型安全的命令 API、异常体系）。两者可以混用：redis-plus-plus 本身依赖 hiredis，你可以在一个工程里同时使用。
2. Redis 核心概念速览（客户端视角必须知道的部分）
2.1 RESP 协议（REdis Serialization Protocol）
hiredis 本质就是一个 RESP 协议解析器。客户端与 Redis 之间传输的每一条命令/回复都是 RESP 编码的文本帧：
| 类型 | 首字节 | 示例 | 说明 |
|---|---|---|---|
| Simple String | + | +OK\r\n | 状态回复 |
| Error | - | -ERR unknown command\r\n | 错误回复 |
| Integer | : | :1000\r\n | 整数回复（INCR 等） |
| Bulk String | $ | $5\r\nhello\r\n | 二进制安全的字符串 |
| Array | * | *2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n | 多条回复 |
客户端发送命令时把参数构造成 Array 帧：*3\r\n$3\r\nSET\r\n$3\r\nfoo\r\n$3\r\nbar\r\n。
2.2 五种基本数据结构与底层编码
| 结构 | 常用命令 | 底层编码（Redis 7） |
|---|---|---|
| String | SET / GET / INCR / SETNX | int、embstr（≤44 字节）、raw |
| Hash | HSET / HGET / HGETALL / HINCRBY | listpack（小）、hashtable |
| List | LPUSH / RPOP / BRPOP / LLEN | quicklist（listpack 节点链表） |
| Set | SADD / SMEMBERS / SISMEMBER | intset（整数小集合）、hashtable |
| ZSet | ZADD / ZSCORE / ZREVRANGE / ZRANGEBYSCORE | listpack（小）、skiplist + dict |
此外还有 Bitmap、HyperLogLog、Geo、Stream 等扩展类型。
2.3 关键语义：TTL、过期与淘汰
- SET key val EX 30 / SETEX / EXPIRE：给实时点位最新值设置 TTL，自动过期清理"失联设备"的残留数据。
- Redis 的过期删除是惰性删除 + 定期删除，客户端不该依赖"过期后立刻不可见"（有窗口期）。
- 内存淘汰策略（maxmemory-policy）决定内存满时的行为，生产建议 allkeys-lru 或 noeviction（宁可报错不丢数据）。
3. API 说明
3.1 hiredis 核心 API
3.1.1 连接管理
#include <hiredis/hiredis.h> redisContext *redisConnect(const char *ip, int port); // 同步连接 redisContext *redisConnectWithTimeout(const char *ip, int port, const struct timeval tv); // 带连接超时 void redisFree(redisContext *c); // 释放连接 int redisSetTimeout(redisContext *c, const struct timeval tv); // 设置读写超时redisContext 关键字段：
typedef struct redisContext { int err; // 0=正常；REDIS_ERR_IO / REDIS_ERR_EOF / REDIS_ERR_PROTOCOL / REDIS_ERR_OOM / REDIS_ERR_OTHER char errstr[128]; // 错误描述 int fd; // 套接字，-1 表示未连接 int flags; // REDIS_BLOCK / REDIS_CONNECTED 等 } redisContext;3.1.2 同步命令
void *redisCommand(redisContext *c, const char *format, ...); // printf 风格格式化 void *redisCommandArgv(redisContext *c, int argc, const char **argv, const size_t *argvlen); // 数组形式，二进制安全 // 返回值是 redisReply*，必须用 freeReplyObject() 释放 typedef struct redisReply { int type; // REDIS_REPLY_STRING / ERROR / INTEGER / NIL / STATUS / ARRAY / DOUBLE ... long long integer; // type==INTEGER 时有效 size_t len; // str 长度（二进制安全） char *str; // type==STRING/ERROR/STATUS 时有效 size_t elements; // type==ARRAY 时元素个数 struct redisReply **element; // 元素数组 } redisReply; void freeReplyObject(void *reply);3.1.3 管道（Pipelining）
// 写入不等待回复 redisAppendCommand(c, "SET foo bar"); redisAppendCommand(c, "GET foo"); // 逐个读取（每个 Append 必须对应一次 GetReply） redisReply *reply = NULL; redisGetReply(c, (void **)&reply); // SET 的结果 freeReplyObject(reply); redisGetReply(c, (void **)&reply); // GET 的结果3.1.4 异步模型
redisAsyncContext *redisAsyncConnect(const char *ip, int port); void redisAsyncFree(redisAsyncContext *ac); int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback fn); // fn(ac, status) int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback fn); // fn(ac, status) int redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, ...); // 事件循环驱动（核心！必须接入 epoll/select/libevent/uv 等） void redisAsyncHandleRead(redisAsyncContext *ac); void redisAsyncHandleWrite(redisAsyncContext *ac);异步回调原型：void callback(redisAsyncContext *c, void *reply, void *privdata);（reply 用完调用释放函数；回调内不要阻塞）。
3.1.5 发布订阅（同步方式）
redisReply *reply = redisCommand(c, "SUBSCRIBE chan:alarm"); freeReplyObject(reply); while (redisGetReply(c, (void **)&reply) == REDIS_OK) { // reply->type == REDIS_REPLY_ARRAY // element[0]=subscribe/message, element[1]=channel, element[2]=payload freeReplyObject(reply); }3.2 redis-plus-plus 核心 API
3.2.1 连接与配置
3.2.2 String / Hash / List / Set / ZSet 命令族（节选）
// String auto val = redis.get("foo"); // OptionalString，std::optional<std::string> redis.set("foo", "bar"); redis.set("foo", "bar", std::chrono::seconds(30)); // 带 TTL auto n = redis.incr("counter"); // 返回 long long // Hash redis.hset("dev:1", "axis_x", "123.45"); auto x = redis.hget("dev:1", "axis_x"); std::unordered_map<std::string, std::string> kv; redis.hgetall("dev:1", std::inserter(kv, kv.begin())); // List redis.lpush("cmd:queue", "MOVE_AXIS 10"); auto item = redis.rpop("cmd:queue"); // OptionalString redis.brpop("cmd:queue", std::chrono::seconds(1)); // 阻塞弹出 // Set / ZSet redis.sadd("online_gateways", "gw-1"); bool online = redis.sismember("online_gateways", "gw-1"); redis.zadd("alarm_rank", "high", 10); // 注意：redis-plus-plus 的 zadd 参数顺序是 score, member auto rank = redis.zrevrange("alarm_rank", 0, 9);注意：redis-plus-plus 命令方法的参数顺序与原生 Redis 命令不一定一致（如 zadd(key, score, member)），以头文件签名 / IDE 提示为准，这是最常见的踩坑点之一。
3.2.3 Pipeline 与 Transaction
// Pipeline：批量发送、一次 exec() 取结果 auto pipe = redis.pipeline(); pipe.set("a", "1"); pipe.set("b", "2"); auto results = pipe.exec(); // vector<OptionalString> 等 // Transaction：MULTI/EXEC，可配合 WATCH auto tx = redis.multi(); tx.set("a", "1"); tx.incr("counter"); auto tx_results = tx.exec();3.2.4 订阅与 Lua
// 订阅（阻塞运行，回调在订阅线程执行，回调内勿阻塞） redis.subscribe({"chan:alarm"}, [](const std::string &chan, const std::string &msg) { std::cout << chan << ": " << msg << std::endl; }); // Lua 脚本（原子执行） auto script = R"( local v = redis.call('GET', KEYS[1]) if v == false then return 0 end redis.call('DEL', KEYS[1]) return 1 )"; long long n = redis.eval(script, {"key"}, {}, 0); // 注意签名：eval(script, keys, args, num_keys)3.2.5 RedisCluster
RedisCluster cluster("tcp://127.0.0.1:7000,tcp://127.0.0.1:7001,tcp://127.0.0.1:7002"); cluster.set("foo", "bar"); // 自动按 CRC16 路由到正确分片4. 详细使用说明
4.0 环境搭建
# 方式一：vcpkg # vcpkg install hiredis redis-plus-plus find_package(hiredis CONFIG REQUIRED) find_package(redis++ CONFIG REQUIRED) target_link_libraries(app PRIVATE hiredis::hiredis redis++::redis++) # 方式二：FetchContent（redis-plus-plus 需要 hiredis 作为依赖） FetchContent_Declare(hiredis URL https://github.com/redis/hiredis/archive/refs/tags/v1.1.0.tar.gz) FetchContent_Declare(redis-plus-plus URL https://github.com/sewenew/redis-plus-plus/archive/refs/tags/1.3.11.tar.gz)4.1 hiredis 最小同步示例（点位读写）
#include <hiredis/hiredis.h> #include <cstdio> #include <cstdlib> int main() { redisContext *c = redisConnect("127.0.0.1", 6379); if (c == nullptr || c->err) { std::fprintf(stderr, "connect failed: %s\n", c ? c->errstr : "no mem"); if (c) redisFree(c); return 1; } // SET redisReply *r = (redisReply *)redisCommand(c, "SET dev:1:axis_x 123.45"); if (r == nullptr) { std::fprintf(stderr, "cmd error: %s\n", c->errstr); redisFree(c); return 1; } freeReplyObject(r); // 必须释放！ // GET + 判断类型 r = (redisReply *)redisCommand(c, "GET dev:1:axis_x"); if (r->type == REDIS_REPLY_STRING) std::printf("axis_x = %.*s\n", (int)r->len, r->str); else std::printf("not found / nil\n"); freeReplyObject(r); // INCR 返回 INTEGER r = (redisReply *)redisCommand(c, "INCR dev:1:cmd_seq"); if (r->type == REDIS_REPLY_INTEGER) std::printf("seq = %lld\n", r->integer); freeReplyObject(r); redisFree(c); return 0; }4.2 hiredis 管道批量写入（采集网关周期性批量上报最新值）
redisAppendCommand(c, "MSET dev:1:axis_x 100.1 dev:1:axis_y 200.2"); redisAppendCommand(c, "MSET dev:2:axis_x 300.3 dev:2:axis_y 400.4"); // 读取两个回复 redisReply *r = nullptr; redisGetReply(c, (void **)&r); freeReplyObject(r); redisGetReply(c, (void **)&r); freeReplyObject(r);也可以直接用 redisCommandArgv 构造 MSET 参数数组（参数值无需转义、天然二进制安全）。
4.3 hiredis 异步 + 事件循环（对接 libuv 风格集成）
void onReply(redisAsyncContext *ac, void *reply, void *privdata) { if (reply == nullptr) { /* 连接已断开 */ return; } redisReply *r = (redisReply *)reply; if (r->type == REDIS_REPLY_STRING) { std::printf("async got: %.*s\n", (int)r->len, r->str); } // hiredis 异步回调里 reply 无需手动释放（框架会处理），注意别 freeReplyObject！ } int main() { redisAsyncContext *ac = redisAsyncConnect("127.0.0.1", 6379); if (ac->err) { std::fprintf(stderr, "async connect: %s\n", ac->errstr); return 1; } redisAsyncSetConnectCallback(ac, [](redisAsyncContext *c, int status) { if (status != REDIS_OK) { std::fprintf(stderr, "connect fail\n"); return; } redisAsyncCommand(c, onReply, nullptr, "GET dev:1:axis_x"); }); redisAsyncSetDisconnectCallback(ac, [](redisAsyncContext *c, int status) { std::fprintf(stderr, "disconnected: %d\n", status); }); // 事件循环驱动（示意）：libevent / libuv / 自行 select // 注册 ac->fd 的可读/可写事件后分别调用 redisAsyncHandleRead / redisAsyncHandleWrite // ... redisAsyncFree(ac); return 0; }4.4 redis-plus-plus 工程化示例：连接池 + 管道 + 分布式锁（工业数采）
#include <sw/redis++/redis++.h> #include <iostream> using namespace sw::redis; int main() { ConnectionOptions opts; opts.host = "127.0.0.1"; opts.port = 6379; opts.socket_timeout = std::chrono::milliseconds(300); opts.connect_timeout = std::chrono::milliseconds(200); ConnectionPoolOptions pool_opts; pool_opts.pool_size = 16; pool_opts.wait_timeout = std::chrono::milliseconds(200); Redis redis(opts, pool_opts); // 1) 点位最新值 + TTL redis.set("dev:1:axis_x", "123.45", std::chrono::seconds(60)); auto v = redis.get("dev:1:axis_x"); if (v) std::cout << "axis_x=" << *v << std::endl; // 2) 管道批量写入 auto pipe = redis.pipeline(); pipe.set("dev:1:axis_y", "456.78"); pipe.set("dev:1:axis_z", "789.01"); pipe.expire("dev:1:axis_y", std::chrono::seconds(60)); auto res = pipe.exec(); // 3) 分布式锁（Lua 保证原子性：SETNX + EXPIRE） auto lock_script = R"( if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then redis.call('EXPIRE', KEYS[1], ARGV[2]) return 1 else return 0 end )"; long long locked = redis.eval(lock_script, {"gw:lock:axis"}, {"gw-1", "10"}, 1); if (locked == 1) std::cout << "lock acquired" << std::endl; // 释放：先校验 token 再 DEL（避免误删他人锁） auto unlock_script = R"( if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end )"; redis.eval(unlock_script, {"gw:lock:axis"}, {"gw-1"}, 1); // 4) 实时告警 PubSub redis.publish("chan:alarm", "axis_x overflow"); return 0; }4.5 工业数采链路建议（C++ 采集网关视角）
FANUC/MC/snap7 设备 │ FOCAS / MC 协议 / S7 协议 ▼ C++ 采集网关 │ hiredis / redis-plus-plus ▼ Redis 实时层（点位最新值 TTL / 命令下发队列 List / 网关分布式锁 / 告警 PubSub） │ 定时批量 ▼ Kafka（历史流）→ TDengine（时序落库）→ 分析 / 告警平台
- 点位最新值：String + TTL，键 dev:{id}:{point}。
- 命令下发：List cmd:{dev}，采集线程 BRPOP 阻塞取命令执行后回写状态。
- 多网关互斥：Lua 分布式锁，避免多网关同时下发冲突命令。
- 实时告警：PubSub chan:alarm，监控线程订阅并转 MQTT/日志。
5. 底层实现剖析
5.1 hiredis 结构：连接层与解析器分离
hiredis 的核心设计是同步/异步共用同一套 RESP 解析器：
- net.c：非阻塞 socket 建立、超时设置（SO_RCVTIMEO / SO_SNDTIMEO 或 poll 轮询）。
- reader（read.c）：纯内存状态机，输入字节流，输出一棵 redisReply 树。不依赖 socket，所以异步模型复用同一解析器。
- redisContext 的 reader 字段是解析器实例；obuf 是待发送缓冲。
5.2 RESP 解析状态机（reader 内部要点）
解析器按首字节分派：
- 读到 $（Bulk String）：先读 $<len>\r\n，若 len == -1 表示 nil，否则读 len 字节 + \r\n，构造 REDIS_REPLY_STRING。
- 读到 *（Array）：先读 *<count>\r\n，若 count == -1 表示 nil array，否则递归解析 count 个元素，构造 REDIS_REPLY_ARRAY。
- 读到 + / - / :：读一行，分别是 STATUS / ERROR / INTEGER。
每次 redisGetReply 会尝试从 reader 取一棵完整回复树；若字节流不完整（半包），返回 REDIS_ERR_IO（err=REDIS_ERR_EOF 语义），下次再调用继续解析剩余字节——这就是"粘包/半包"天然被 hiredis 处理掉的原因。
5.3 异步模型：事件循环驱动
- redisAsyncContext 包装 redisContext，把命令封装成 redisCallback 链表挂在连接上。
- redisAsyncCommand 只是把命令追加到输出缓冲；真正发送发生在事件循环调用 redisAsyncHandleWrite（fd 可写）时。
- 回复到达后 redisAsyncHandleRead 解析，按命令顺序触发回调（hiredis 用 FIFO 保证命令-回调配对）。
- 没有事件循环 = 异步永远不会推进，这是初学 hiredis 最大的困惑来源。
5.4 redis-plus-plus 的封装层次
- Redis 持有 ConnectionPool（hiredis 连接数组 + 互斥借用归还），每个命令方法从池中借连接执行再归还。
- RedisCluster 内部维护 slot → node 映射（16384 个 CRC16 slot），收到 MOVED/ASK 后自动重路由。
- Pipeline/Transaction 封装成 PipelineImpl/TransactionImpl，延迟到 exec() 统一发送+收集。
- 异常体系：Error、TimeoutError、OOMError、ClosedError、IoError 等，所有命令可能抛 Error，工程上必须捕获。
6. 常错点 / 坑（Top 20）
| 坑点 | 后果 | 正确姿势 | |
|---|---|---|---|
| 1 | 不检查 redisContext 的 err / NULL | 空指针解引用、静默失败 | 连接后立即检查 c && !c->err |
| 2 | redisCommand 返回 NULL 还继续访问 | 崩溃 / 未定义行为 | 判空后先看 c->errstr |
| 3 | redisReply 用完不 freeReplyObject | 内存泄漏（长连接下尤其严重） | 每条回复必释放 |
| 4 | 不判断 r->type 直接用 r->str | 类型错读（INTEGER 无 str） | 按 type 分派处理 |
| 5 | 管道 Append 次数与 GetReply 次数不匹配 | 回复错位、逻辑错误 | 每 Append 必须有对应 GetReply |
| 6 | 异步回调里 freeReplyObject(reply) | 二次释放崩溃 | 异步回调 reply 由框架管理 |
| 7 | 异步连接不接入任何事件循环 | 命令永远发不出去/收不到 | 注册 fd 可读可写事件并调用 HandleRead/Write |
| 8 | 多线程共享同一个同步 context | 数据竞争、协议错乱 | 每线程独立连接，或用异步+互斥 |
| 9 | 同步订阅后在同一连接上再发普通命令 | 阻塞、命令排队无人消费 | 订阅用专用连接/线程 |
| 10 | 未设超时（redisSetTimeout） | 网络抖动时无限阻塞 | 显式设置读写超时 |
| 11 | 只查 connect 成功就以为一切 OK | 密码错误/AUTH 失败在首次命令才暴露 | 连接后主动 PING |
| 12 | Cluster 直连单节点 | MOVED/ASK 错误 | 用 Cluster 客户端或处理重定向 |
| 13 | get 返回值不判空直接解引用 | 未命中时崩溃 | 用 if (val) / val.has_value() |
| 14 | zadd(key, member, score) 参数顺序记反 | 数据写错 | 顺序是 zadd(key, score, member) |
| 15 | 命令抛异常不捕获 | 进程中断 | 捕获 sw::redis::Error 系列 |
| 16 | 池耗尽 + wait_timeout 设 0 | 快速失败或无限等待 | 评估峰值并发合理配置 |
| 17 | brpop 入参顺序错误 | 弹出错误方向/顺序错乱 | 按签名传（先 key 后 timeout 语义看 IDE） |
| 18 | 订阅回调里执行阻塞/耗时操作 | 卡住整个订阅线程、丢消息 | 回调只做投递，用队列交给工作线程 |
| 19 | 多个命令想原子执行却用多个调用 | 中间状态可见 | 用 Lua / Transaction |
| 20 | 误把 exec() 返回值当单个结果 | 类型不匹配编译/运行错误 | Pipeline 返回 vector，Transaction 同理 |
7. 性能优化清单
- 管道化批量：采集网关 100ms 周期把 N 个点位 MSET/Pipeline 一次发送，吞吐可提升 10~50 倍。
- 连接池：redis-plus-plus 默认池化；hiredis 自行维护连接复用，避免每次命令建连。
- 避免大 Key：单 Key 超过几十 KB 会拖慢整个实例；点位数据按点拆 Key。
- 合理 TTL：实时点位设 TTL，防止僵尸 Key 堆积。
- 序列化最小化：数值点位用字符串化数字即可，不要套 JSON。
- Lua 原子化：分布式锁/计数/校验用 Lua，省 RTT 且原子。
- 订阅线程独立：PubSub 高频消息不要与业务命令共用连接。
- 关闭 AOF 或调整策略：纯缓存场景可 appendonly no；需要持久化用 everysec。
- 客户端侧超时收敛：socket_timeout 统一几百 ms，避免雪崩式阻塞。
- 监控：INFO stats / SLOWLOG 治理慢命令；客户端侧统计命令耗时。
8. FAQ 速查表
Q1: hiredis 和 redis-plus-plus 该选哪个？A: 需要极致控制力/零依赖/嵌入式 → hiredis；工程化开发（连接池、类型安全、Cluster、订阅封装）→ redis-plus-plus。两者可同工程混用。
Q2: hiredis 同步 API 是线程安全的吗？A: 不是。一个 context 只能被一个线程使用；多线程请各自建连接或加锁。
Q3: 为什么异步连接明明 connect 成功却收不到回调？A: 因为没有事件循环驱动 redisAsyncHandleRead/Write。必须把 ac->fd 注册到 epoll/select/libevent/libuv。
Q4: redis-plus-plus 的 Redis 对象是线程安全的吗？A: 是的，Redis 内部用连接池 + 互斥保证多线程安全，可跨线程共享。
Q5: Pipeline 和 Transaction 的区别？A: Pipeline 只做批量发送减少 RTT，不保证原子；Transaction（MULTI/EXEC）保证原子性。
Q6: 为什么用 Lua 做分布式锁？A: SETNX + EXPIRE 两条命令非原子，进程崩溃会死锁；Lua 单次执行原子，配合 token 校验释放避免误删。
Q7: Redis 会丢数据吗？A: 默认内存为主。纯缓存可接受；要求可靠性用 AOF everysec 或直接走 Kafka 历史流。
Q8: 大流量采集网关怎么组织 Key？A: 每设备每点位独立 Key + TTL；批量上报用 Pipeline/MSET；命令下发用 List + BRPOP。
9. 总结
C++ 侧访问 Redis 的主流路径是hiredis（官方 C 客户端，RESP 解析器 + 同步/异步双模型）与其上层的redis-plus-plus（C++17 现代封装）。对工业数采网关而言，Redis 实时层承担"点位最新值、命令下发、分布式锁、实时告警"四类职责，与 Kafka/TDengine 的历史流形成"实时 + 历史"双通道，与 Go 侧 go-redis 构成跨语言对照。掌握本文的 API 分层、RESP 解析原理、异步事件驱动模型与 20 条高频坑点，即可在生产采集链路上稳定落地。
