技术栈C / C17 · Redis 7.x · hiredis 1.1.0 · redis-plus-plussw/redis1.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-plusC/C 生态的 Redis 客户端主要有三条路线库语言定位特点hiredisC官方推荐的底层客户端极简、零依赖、同步异步双模型、RESP 解析器独立可用是 redis-plus-plus、很多框架如 Drogon 的 redis 插件的底层redis-plus-plussw/redisC17hiredis 之上的现代 C 封装Redis/RedisCluster 类、命令族方法、Optional 返回值、Pipeline/Transaction、连接池、订阅回调、Lua 脚本cpp-redis / acreC第三方封装生态小维护不活跃hiredis 适合对性能与控制力要求极高的场景手写事件循环、嵌入式采集器redis-plus-plus 适合工程化程度高的 C 项目自动连接池、类型安全的命令 API、异常体系。两者可以混用redis-plus-plus 本身依赖 hiredis你可以在一个工程里同时使用。2. Redis 核心概念速览客户端视角必须知道的部分2.1 RESP 协议REdis Serialization Protocolhiredis 本质就是一个 RESP 协议解析器。客户端与 Redis 之间传输的每一条命令/回复都是 RESP 编码的文本帧类型首字节示例说明Simple StringOK\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 7StringSET / GET / INCR / SETNXint、embstr≤44 字节、rawHashHSET / HGET / HGETALL / HINCRBYlistpack小、hashtableListLPUSH / RPOP / BRPOP / LLENquicklistlistpack 节点链表SetSADD / SMEMBERS / SISMEMBERintset整数小集合、hashtableZSetZADD / ZSCORE / ZREVRANGE / ZRANGEBYSCORElistpack小、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 核心 API3.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; // typeINTEGER 时有效 size_t len; // str 长度二进制安全 char *str; // typeSTRING/ERROR/STATUS 时有效 size_t elements; // typeARRAY 时元素个数 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 核心 API3.2.1 连接与配置3.2.2 String / Hash / List / Set / ZSet 命令族节选// String auto val redis.get(foo); // OptionalStringstd::optionalstd::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_mapstd::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(); // vectorOptionalString 等 // TransactionMULTI/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 RedisClusterRedisCluster 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) # 方式二FetchContentredis-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 轮询。readerread.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_IOerrREDIS_ERR_EOF 语义下次再调用继续解析剩余字节——这就是粘包/半包天然被 hiredis 处理掉的原因。5.3 异步模型事件循环驱动redisAsyncContext 包装 redisContext把命令封装成 redisCallback 链表挂在连接上。redisAsyncCommand 只是把命令追加到输出缓冲真正发送发生在事件循环调用 redisAsyncHandleWritefd 可写时。回复到达后 redisAsyncHandleRead 解析按命令顺序触发回调hiredis 用 FIFO 保证命令-回调配对。没有事件循环 异步永远不会推进这是初学 hiredis 最大的困惑来源。5.4 redis-plus-plus 的封装层次Redis 持有 ConnectionPoolhiredis 连接数组 互斥借用归还每个命令方法从池中借连接执行再归还。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-err2redisCommand 返回 NULL 还继续访问崩溃 / 未定义行为判空后先看 c-errstr3redisReply 用完不 freeReplyObject内存泄漏长连接下尤其严重每条回复必释放4不判断 r-type 直接用 r-str类型错读INTEGER 无 str按 type 分派处理5管道 Append 次数与 GetReply 次数不匹配回复错位、逻辑错误每 Append 必须有对应 GetReply6异步回调里 freeReplyObject(reply)二次释放崩溃异步回调 reply 由框架管理7异步连接不接入任何事件循环命令永远发不出去/收不到注册 fd 可读可写事件并调用 HandleRead/Write8多线程共享同一个同步 context数据竞争、协议错乱每线程独立连接或用异步互斥9同步订阅后在同一连接上再发普通命令阻塞、命令排队无人消费订阅用专用连接/线程10未设超时redisSetTimeout网络抖动时无限阻塞显式设置读写超时11只查 connect 成功就以为一切 OK密码错误/AUTH 失败在首次命令才暴露连接后主动 PING12Cluster 直连单节点MOVED/ASK 错误用 Cluster 客户端或处理重定向13get 返回值不判空直接解引用未命中时崩溃用 if (val) / val.has_value()14zadd(key, member, score) 参数顺序记反数据写错顺序是 zadd(key, score, member)15命令抛异常不捕获进程中断捕获 sw::redis::Error 系列16池耗尽 wait_timeout 设 0快速失败或无限等待评估峰值并发合理配置17brpop 入参顺序错误弹出错误方向/顺序错乱按签名传先 key 后 timeout 语义看 IDE18订阅回调里执行阻塞/耗时操作卡住整个订阅线程、丢消息回调只做投递用队列交给工作线程19多个命令想原子执行却用多个调用中间状态可见用 Lua / Transaction20误把 exec() 返回值当单个结果类型不匹配编译/运行错误Pipeline 返回 vectorTransaction 同理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不保证原子TransactionMULTI/EXEC保证原子性。Q6: 为什么用 Lua 做分布式锁A: SETNX EXPIRE 两条命令非原子进程崩溃会死锁Lua 单次执行原子配合 token 校验释放避免误删。Q7: Redis 会丢数据吗A: 默认内存为主。纯缓存可接受要求可靠性用 AOF everysec 或直接走 Kafka 历史流。Q8: 大流量采集网关怎么组织 KeyA: 每设备每点位独立 Key TTL批量上报用 Pipeline/MSET命令下发用 List BRPOP。9. 总结C 侧访问 Redis 的主流路径是hiredis官方 C 客户端RESP 解析器 同步/异步双模型与其上层的redis-plus-plusC17 现代封装。对工业数采网关而言Redis 实时层承担点位最新值、命令下发、分布式锁、实时告警四类职责与 Kafka/TDengine 的历史流形成实时 历史双通道与 Go 侧 go-redis 构成跨语言对照。掌握本文的 API 分层、RESP 解析原理、异步事件驱动模型与 20 条高频坑点即可在生产采集链路上稳定落地。
