libvalkey Standalone API 完全指南同步/异步连接、命令执行、Pipelining 与 TLS 实战【免费下载链接】placeholderkvA flexible distributed key-value database that is optimized for caching and other realtime workloads.项目地址: https://gitcode.com/GitHub_Trending/pl/placeholderkv导读libvalkey是 Valkey 官方维护的 C 语言客户端库用于在 C/C 程序中以 RESP 协议与 Valkey 服务器通信。本文基于 deps/libvalkey/docs/standalone.md 展开完整覆盖 standalone非集群模式下的同步 API 与异步 API连接建立与选项配置、命令构造printf风格格式化与 argv 数组、回复类型与错误处理、Pipeline 批处理、Reader 调优、RESP3 Push 消息、分配器注入以及 TLS 支持的启用方式。读完本文你将能够写出健壮、可上线运行的 Valkey C 客户端程序并能依据源码理解其底层行为。说明本文是对官方文档的完整扩写示例与结论均可在仓库内对应源码与示例程序中找到依据。API 完整参考请以 include/valkey/valkey.h 与 src/valkey.c 为准。同步 API 概览同步 API 的“表面积”非常小核心只需要掌握少量函数。这些函数整体上非常类似printf的工作方式——你提供格式字符串和变参libvalkey 负责把参数构造成合法的 RESP 命令并发送给服务器。整个同步 API 围绕valkeyContext展开。从源码看valkeyContext封装了一次连接的全部状态错误码err与错误描述errstr、文件描述符fd、写缓冲区obuf、协议解析器reader、连接类型connection_type、连接/命令超时以及用户私有数据privdata等参见 include/valkey/valkey.h。建立连接libvalkey 提供多个便捷的连接函数覆盖 TCP、Unix Socket、非阻塞、绑定源地址、复用已打开 fd 等场景完整清单见 include/valkey/valkey.hvalkeyContext *valkeyConnect(const char *host, int port); valkeyContext *valkeyConnectUnix(const char *path); // 还有一组便捷结构体用来指定各种连接选项。 valkeyContext *valkeyConnectWithOptions(valkeyOptions *opt);其他常用变体还包括valkeyConnectWithTimeout、valkeyConnectNonBlock、valkeyConnectBindNonBlock、valkeyConnectUnixWithTimeout、valkeyConnectFd(valkeyFD fd)等均可直接复用。连接时需要区分两种失败模式无法分配valkeyContext结构体时返回NULL典型如内存耗尽能建立上下文但连接本身有问题时设置上下文的err成员。因此标准的错误处理写法是先判空、再查errvalkeyContext *ctx valkeyConnect(localhost, 6379); if (ctx NULL || ctx-err) { fprintf(stderr, Error connecting: %s\n, ctx ? ctx-errstr : OOM); }多地址解析行为当一个主机名解析出多个 IP 地址时libvalkey 会按顺序逐个尝试直到某个地址连接成功或全部失败为止。注意connect_timeout是按地址生效的因此当多个地址都不可达时总的连接等待时间最高可能达到 N × timeoutN 为解析出的地址数量。在设计高可用客户端的超时参数时需要把这个叠加效应考虑进去。连接选项valkeyOptionsvalkeyOptions是一个辅助结构体集中描述连接目标与各种行为开关。除了连接信息还包含connect_timeout、command_timeout、privdata、RESP3 PUSH 回调等字段完整定义见 include/valkey/valkey.h。基本用法如下valkeyOptions opt {0}; // 设置主要连接信息 if (tcp) { VALKEY_OPTIONS_SET_TCP(opt, localhost, 6379); } else { VALKEY_OPTIONS_SET_UNIX(opt, /tmp/valkey.sock); } // 可以把任意数据挂到 context 上 VALKEY_OPTIONS_SET_PRIVDATA(opt, my_data);源码中这三个宏的实现include/valkey/valkey.h分别设置type VALKEY_CONN_TCP/VALKEY_CONN_UNIX、填充endpoint联合体TCP 的ipport或 Unix 的unix_socket路径以及设置privdata和对应的析构函数free_privdata。值得注意VALKEY_OPTIONS_SET_PRIVDATA除数据指针外还接受一个析构函数context 释放时会自动调用用于释放用户资源除 TCP 与 Unix 外valkeyOptions的endpoint联合体还支持VALKEY_CONN_USERFD操作一个已打开的 fd以及实验性的VALKEY_CONN_RDMA见 include/valkey/valkey.h官方示例 examples/blocking-push.c 演示了完整的valkeyOptions初始化流程设置 TCP、挂载 privdata 并指定析构、再设置 PUSH 回调后调用valkeyConnectWithOptions。选项标志位valkeyOptions.options是一个位域支持以下标志定义与注释见 include/valkey/valkey.hFlagDescriptionVALKEY_OPT_NONBLOCK建立非阻塞连接。VALKEY_OPT_REUSEADDR设置SO_REUSEADDRsocket 选项。VALKEY_OPT_PREFER_IPV4VALKEY_OPT_PREFER_IPV6VALKEY_OPT_PREFER_IP_UNSPEC控制调用getaddrinfo时的地址族偏好。VALKEY_OPT_PREFER_IP_UNSPEC会以AF_UNSPEC调用同时搜索 IPv4 与 IPv6 地址libvalkey 默认偏好 IPv4。VALKEY_OPT_NO_PUSH_AUTOFREE不安装默认的 RESP3 PUSH 处理器默认处理器会拦截并释放这些消息。适用于需要带内in-band处理这些消息的场景。VALKEY_OPT_NOAUTOFREEREPLIES异步执行完回复回调后不自动调用freeReplyObject。VALKEY_OPT_NOAUTOFREE异步连接/通信失败时不自动释放valkeyAsyncContext仅当用户显式调用valkeyAsyncDisconnect或valkeyAsyncFree时才释放。VALKEY_OPT_MPTCP使用多路径 TCPMPTCP。注意只有服务器与客户端同时支持 MPTCP 时才建立 MPTCP 连接否则退化为普通 TCP 连接。源码还提供了便捷宏VALKEY_OPTIONS_SET_MPTCP(opts, ip, port)include/valkey/valkey.h一步完成 TCP 参数与 MPTCP 标志的设置。执行命令核心命令接口是一个printf风格的变参函数传入格式字符串与可变参数libvalkey 负责构造 RESP 命令并发送valkeyReply *reply valkeyCommand(ctx, INCRBY %s %d, counter, 42); if (reply NULL) { fprintf(stderr, Communication error: %s\n, c-err ? c-errstr : Unknown error); } else if (reply-type VALKEY_REPLY_ERROR) { fprintf(stderr, Error response from server: %s\n, reply-str); } else if (reply-type ! VALKEY_REPLY_INTEGER) { // 非常罕见但值得检查。 fprintf(stderr, Error: Non-integer reply to INCRBY?\n); } printf(New value of counter is %lld\n, reply-integer); freeReplyObject(reply);二进制安全%b格式符需要向服务器发送二进制安全的数据时使用%b格式符并额外传入长度参数struct binary { int x; int y; } {0xdeadbeef, 0xcafebabe}; valkeyReply *reply valkeyCommand(ctx, SET %s %b, some-key, binary, sizeof(binary));数组形式valkeyCommandArgv命令也可以由“参数数组 可选的长度数组”构造。不提供长度数组时libvalkey 会对每个参数执行strlenconst char *argv[] {SET, captain, James Kirk}; const size_t argvlens[] {3, 7, 10}; valkeyReply *reply valkeyCommandArgv(ctx, 3, argv, argvlens); // 错误处理方式与 valkeyCommand 相同底层实现上valkeyCommand等价于valkeyAppendCommandvalkeyGetReply的组合见 include/valkey/valkey.h 的注释阻塞上下文里它会追加命令并立即读取回复非阻塞上下文里它只做追加并始终返回NULL。使用回复valkeyReplyvalkeyCommand与valkeyCommandArgv成功时返回valkeyReply指针发生严重错误如与服务器通信失败、内存不足时返回NULL。回复为NULL时通过valkeyContext-err查询错误码、valkeyContext-errstr查询人类可读的错误描述。拿到非空valkeyReply后必须检查valkeyReply-type字段判断回复类型。例如命令本身出错时回复类型为VALKEY_REPLY_ERROR具体错误字符串在reply-str中。回复类型一览valkeyReply的完整定义位于 include/valkey/valkey.h类型常量定义于 include/valkey/read.hVALKEY_REPLY_ERROR— 错误回复错误字符串在reply-str。VALKEY_REPLY_STATUS— 状态回复如OK在reply-str。VALKEY_REPLY_INTEGER— 整数回复在reply-integer。VALKEY_REPLY_DOUBLE— 浮点回复在reply-dval以及reply-str。VALKEY_REPLY_NIL— nil 回复。VALKEY_REPLY_BOOL— 布尔回复在reply-integer。VALKEY_REPLY_BIGNUM— 目前尚未使用若出现字符串会在reply-str。VALKEY_REPLY_STRING— 字符串回复在reply-str。VALKEY_REPLY_VERB— verbatim 字符串回复内容在reply-str其类型标注在reply-vtype。VALKEY_REPLY_ARRAY— 数组回复元素在reply-element元素个数在reply-elements。VALKEY_REPLY_MAP— Map 回复结构与VALKEY_REPLY_ARRAY相同仅语义上表示键值对同样通过reply-element与reply-elements访问。VALKEY_REPLY_SET— 表示集合的类数组回复例如SMEMBERS的结果通过reply-element与reply-elements访问。VALKEY_REPLY_ATTR— 属性回复目前 valkey-server 尚未使用。VALKEY_REPLY_PUSH— 带外out-of-bandPush 回复同样具有数组形态。断开连接与清理libvalkey 返回的所有非空valkeyReply结构体都应由调用方用freeReplyObject释放断开连接并释放上下文则调用valkeyFreevalkeyReply *reply valkeyCommand(ctx, set %s %s, foo, bar); // 错误处理 ... freeReplyObject(reply); // 断开连接并释放上下文 valkeyFree(ctx);除valkeyFree外include/valkey/valkey.h 还提供了valkeyFreeKeepFd它释放 context 但保留底层 fd 供调用方继续使用此外还有valkeyReconnect沿用初始连接选项原地重连与valkeySetTimeout运行时调整超时等辅助函数适合写断线重连逻辑。Pipelining管道批处理valkeyCommand与valkeyCommandArgv每条命令都会产生一次服务器往返。如果需要批量发送命令可以用valkeyAppendCommand与valkeyAppendCommandArgv实现管道valkeyAppendCommand只是把命令追加到valkeyContext的输出缓冲区并不会真正发送直到第一次调用valkeyGetReply读取回复时整个输出缓冲区才会一次性交付给服务器。// 追加命令期间不会有任何数据发往服务器。 for (size_t i 0; i 100000; i) { if (valkeyAppendCommand(c, INCRBY key:%zu %zu, i, i) ! VALKEY_OK) { fprintf(stderr, Error appending command: %s\n, c-errstr); exit(1); } } // 第一次调用 valkeyGetReply 时整个输出缓冲区一次性发出。 for (size_t i 0; i 100000; i) { if (valkeyGetReply(c, (void**)reply) ! VALKEY_OK) { fprintf(stderr, Error reading reply %zu: %s\n, i, c-errstr); exit(1); } else if (reply-type ! VALKEY_REPLY_INTEGER) { fprintf(stderr, Error: Non-integer reply to INCRBY?\n); exit(1); } printf(INCRBY key:%zu %lld\n, i, reply-integer); freeReplyObject(reply); }从 include/valkey/valkey.h 的注释可以确认valkeyGetReply的完整语义在阻塞上下文中它先检查是否有未消费的回复有则直接返回否则刷新输出缓冲区到 socket并持续读取直到拿到一条回复。因此它天然支持“先攒一批命令、再统一收回复”的流水线模式。valkeyGetReply也可用于非管道场景例如订阅场景中持续阻塞读取消息valkeyReply *reply valkeyCommand(c, SUBSCRIBE channel); assert(reply ! NULL !c-err); while (valkeyGetReply(c, (void**)reply) VALKEY_OK) { // 处理消息... freeReplyObject(reply); }错误处理如前述发生通信错误时 libvalkey 返回NULL并在上下文中设置err与errstr。具体错误码定义于 include/valkey/read.hVALKEY_ERR_IO— 连接读写出现问题应结合errno进一步定位。VALKEY_ERR_EOF— 服务器关闭了连接。VALKEY_ERR_PROTOCOL— 解析回复时出现协议错误。VALKEY_ERR_TIMEOUT— 连接、读取或写入超时。VALKEY_ERR_OOM— 内存不足。VALKEY_ERR_OTHER— 其他错误查看c-errstr获取详情。线程安全valkeyContext结构体不是线程安全的。除非你非常清楚自己在做什么否则不应在多个线程之间共享同一个 context。多线程程序的常规做法是每个线程持有独立的连接。Reader 配置协议解析器调优libvalkey 的上下文还提供了若干可定制机制作用于内部的协议解析器valkeyReader结构体定义见 include/valkey/read.h。输入缓冲区上限maxbuflibvalkey 使用一块缓冲区暂存接收到的字节缓冲区清空后会收缩回可配置的最大值默认16KB宏VALKEY_READER_MAX_BUF见 include/valkey/read.h。为避免频繁重复分配可以将该值调大设为0表示“无上限”context-reader-maxbuf 0;数组元素上限maxelements默认情况下libvalkey 拒绝解析元素数超过 2^32−1即 4,294,967,295的类数组回复默认宏VALKEY_READER_MAX_ARRAY_ELEMENTS见 include/valkey/read.h。该值可以设为任意 64 位值或设0表示“无限制”context-reader-maxelements 0;RESP3 Push 回复RESP 协议第三版引入了带外out-of-band的 “push” 回复它们可能在数据流的任意时刻到达。默认情况下libvalkey 会处理这些消息后直接丢弃。如果应用需要对 PUSH 消息执行特定动作可以安装自定义处理器在消息到达时被回调也可以把 push handler 设为NULL让消息以“带内”in-band方式交付——这在阻塞式订阅循环中很有用。注意也可以在valkeyOptions结构体中指定 push handler在初始化时即生效。同步上下文void my_push_handler(void *privdata, void *reply) { // 在同步上下文中处理完回复后需要自行释放它。 } // 初始化等... valkeySetPushCallback(c, my_push_handler);异步上下文void my_async_push_handler(valkeyAsyncContext *ac, void *reply) { // 与其他异步回复一样libvalkey 会自动释放它 // 除非你用 VALKEY_OPT_NOAUTOFREE 配置了上下文。 } // 初始化等... valkeyAsyncSetPushCallback(ac, my_async_push_handler);实战示例客户端缓存失效监听仓库中的 examples/blocking-push.c 是一个完整的 RESP3 PUSH 实战用例它通过HELLO 3切换到 RESP3 协议并开启CLIENT TRACKING ON随后读取 key 再改写 key从而触发客户端缓存失效的 invalidation push 消息。其pushReplyHandlerexamples/blocking-push.c会解析VALKEY_REPLY_PUSH结构并打印失效的 key 名——这正是构建“客户端缓存 服务端失效通知”缓存层的标准模板。分配器注入Allocator injectionlibvalkey 内部通过一层间接层调用标准分配函数维护一个全局结构体其中保存指向实际分配器的函数指针。默认它们就是malloc、calloc、realloc等。可以按如下方式覆盖结构体valkeyAllocFuncs定义见 include/valkey/alloc.hvalkeyAllocFuncs my_allocators { .mallocFn my_malloc, .callocFn my_calloc, .reallocFn my_realloc, .strdupFn my_strdup, .freeFn my_free, }; // libvalkey 会返回之前设置的分配器便于恢复。 valkeyAllocFuncs old valkeySetAllocators(my_allocators);也可以重置回 glibc 或 musl 的默认实现valkeyResetAllocators();注意vk_calloc会处理nmemb * size溢出size_t的情况溢出时返回NULL。从 include/valkey/alloc.h 可以看到这个溢出检查发生在调用用户自定义的callocFn之前因此即使注入任意分配器也不会出现乘法溢出导致的越界分配。异步 APIlibvalkey 提供完整的异步 API并支持众多事件库。每种事件库的具体接入方式请参考 examples 目录下的对应示例文件如async-libev.c、async-libevent.c、async-libuv.c、async-ae.c、async-poll.c、async-macosx.c、async-qt.cpp等。建立异步连接libvalkey 通过valkeyAsyncContext管理异步连接其使用方式与同步上下文类似。声明与接口见 include/valkey/async.hvalkeyAsyncContext *ac valkeyAsyncConnect(localhost, 6379); if (ac NULL) { fprintf(stderr, Error: Out of memory trying to allocate valkeyAsyncContext\n); exit(1); } else if (ac-err) { fprintf(stderr, Error: %s (%d)\n, ac-errstr, ac-err); exit(1); } // 如果使用 libev valkeyLibevAttach(EV_DEFAULT_ ac); valkeySetConnectCallback(ac, my_connect_callback); valkeySetDisconnectCallback(ac, my_disconnect_callback); ev_run(EV_DEFAULT_ 0);异步上下文应当持有连接回调connect callback连接尝试完成无论成功或出错时都会被调用它可以持有断连回调disconnect callback连接断开因错误或用户主动请求时被调用。断连回调触发后上下文对象总是会被释放。头文件中还提供了valkeyAsyncConnectWithOptions复用valkeyOptions可同时携带连接选项与 PUSH 回调、valkeyAsyncConnectBind、valkeyAsyncConnectUnix等变体。执行异步命令异步上下文中的命令执行与同步类似区别在于可以传入一个回调在收到回复时被调用struct my_app_data { size_t incrby_replies; size_t get_replies; }; void my_incrby_callback(valkeyAsyncContext *ac, void *r, void *privdata) { struct my_app_data *data privdata; valkeyReply *reply r; assert(reply ! NULL reply-type VALKEY_REPLY_INTEGER); printf(Incremented value: %lld\n, reply-integer); ># 构建并安装默认库 sudo make install # 启用全部可选功能 sudo USE_TLS1 USE_RDMA1 make install # 如果 openssl 安装在非默认位置 sudo USE_TLS1 OPENSSL_PREFIX/path/to/openssl make install使用 CMake 构建# 构建并安装 mkdir build cd build cmake -DCMAKE_BUILD_TYPERelWithDebInfo .. sudo make install # 启用 TLS 与 RDMA mkdir build cd build cmake -DCMAKE_BUILD_TYPERelWithDebInfo -DENABLE_TLS1 -DENABLE_RDMA1 .. sudo make install进一步探索完整 API 参考include/valkey/valkey.h、include/valkey/async.h、include/valkey/read.h、include/valkey/alloc.h、include/valkey/net.h、include/valkey/tls.h核心实现src/valkey.c上下文与命令、src/async.c异步核心、src/read.cRESP 协议解析器、src/net.csocket 连接、src/tls.c可运行示例examples 目录下blocking.c、blocking-push.c、blocking-tls.c与各事件库的async-*.c构建与安装说明deps/libvalkey/README.md【免费下载链接】placeholderkvA flexible distributed key-value database that is optimized for caching and other realtime workloads.项目地址: https://gitcode.com/GitHub_Trending/pl/placeholderkv创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考