brpc Memcached 客户端完全指南基于二进制协议的工业级访问方案【免费下载链接】brpcbrpc is an Industrial-grade RPC framework using C Language, which is often used in high performance system such as Search, Storage, Machine learning, Advertisement, Recommendation etc. brpc means better RPC.项目地址: https://gitcode.com/GitHub_Trending/brpc/brpcbrpc 在 RPC 框架之上直接内置了对 memcached 二进制协议的支持让开发者无需引入额外客户端库即可用统一的方式访问 memcached 单机或集群。本文以 docs/en/memcache_client.md 为骨架结合仓库源码与测试用例系统讲解 Channel 初始化、MemcacheRequest/MemcacheResponse 的使用、pipeline 批量操作、集群访问策略以及底层协议实现原理读完即可上手写出高性能的 memcached 访问代码。为什么要用 brpc 访问 memcachedmemcached 是常用的缓存服务。为了使用户更快捷地访问 memcached 并充分利用 bthread 的并发能力brpc 直接在框架层面支持了 memcache 协议无需依赖第三方客户端。官方示例位于 example/memcache_c。注意brpc 只支持 memcache 的二进制协议binary protocol。memcached 在 1.3 之前只有文本协议如今支持文本协议的意义甚微如果你的 memcached 早于 1.3请升级到新版本。相比使用官方客户端 libmemcachedbrpc 的实现有以下优势文档原文要点线程安全Channel本身是线程安全的不需要为每个线程建立独立的 client多个线程可共享同一个 Channel 并发访问。访问模式丰富支持同步、异步、半同步等访问方式并能通过 ParallelChannel 等组合 Channel 以声明式方式定义复杂的访问模式。复用 brpc 基础设施支持多种连接方式single / pooled / short 等以及超时、backup request、取消、tracing、内置服务等一系列 brpc 提供的成熟能力。有明确的 request 和 response 概念libmemcached 没有这一概念收到的消息不能直接与发出的消息对应用户必须自行做额外的簿记bookkeeping工作且很容易出错而 brpc 中一次CallMethod对应明确的请求与响应。当前实现充分利用了 RPC 的并发机制并尽量避免了拷贝。根据官方文档的记载一个 client 可以轻松把一个同机 memcached 实例版本 1.4.15压到极限单连接约 9 万 QPS多连接约 33 万 QPS。在大部分情况下brpc client 都能充分发挥 memcached 的性能。访问单台 memcached第一步创建 Channel创建一个访问 memcached 的Channel只需把协议设置为PROTOCOL_MEMCACHE#include brpc/memcache.h #include brpc/channel.h brpc::ChannelOptions options; options.protocol brpc::PROTOCOL_MEMCACHE; if (channel.Init(0.0.0.0:11211, options) ! 0) { // 11211 是 memcached 的默认端口 LOG(FATAL) Fail to init channel to memcached; return -1; } ...Channel::Init的第一个参数既可以是ip:port形式的地址也可以是命名服务地址配合集群访问使用。更多连接选项如connection_type、timeout_ms、max_retry可以像 example/memcache_c/client.cpp 中那样在ChannelOptions中设置。第二步发送一个 SET 请求往 memcached 中设置一份数据// 写入 keyhello valueworld flags0xdeadbeef10 秒后失效cas_value 传 0 表示忽略 CAS。 brpc::MemcacheRequest request; brpc::MemcacheResponse response; brpc::Controller cntl; if (!request.Set(hello, world, 0xdeadbeef/*flags*/, 10/*expiring seconds*/, 0/*ignore cas*/)) { LOG(FATAL) Fail to SET request; return -1; } channel.CallMethod(nullptr, cntl, request, response, nullptr/*done*/); if (cntl.Failed()) { LOG(FATAL) Fail to access memcached, cntl.ErrorText(); return -1; } if (!response.PopSet(nullptr)) { LOG(FATAL) Fail to SET memcached, response.LastError(); return -1; } ...上述代码的关键要点请求与响应类型必须固定请求类型必须是MemcacheRequest响应类型必须是MemcacheResponse否则CallMethod会失败。不需要 stub直接调用channel.CallMethodmethod参数填nullptr即可。这背后的校验逻辑在 memcache_binary_protocol.cpp 的SerializeMemcacheRequest中序列化前会检查request-GetDescriptor() ! MemcacheRequest::descriptor()不匹配直接报Must be MemcacheRequest。pipeline 模式调用request.XXX()增加操作本例 XXXSet。一个 request 内多次调用不同的操作这些操作会被同时送到 memcached这就是常说的 pipeline 模式。底层实现是把每个操作编码成独立的二进制协议包依次追加到同一个IOBuf见_buf并由_pipelined_count记录操作个数见 memcache.cpp 中的_pipelined_count。依次 Pop 结果调用response.PopXXX()弹出每个操作的结果本例 XXXSet。成功返回true失败返回false此时调用response.LastError()可获得错误信息。XXX必须和 request 中操作的顺序一一对应否则 Pop 会失败例如把PopSet换成PopGet会得到错误信息not a GET response见 memcache.cpp 中PopGet对 command 的校验。Pop 结果独立于 RPC 结果即使某个值没能成功写入 memcached例如业务层失败RPC 本身可能仍然是成功的。RPC 失败指的是连接断开、超时这类传输层错误。如果业务上要求操作真正成功才算成功那么既要判断 RPC 成功也要判断 PopXXX 成功。完整支持的操作清单目前支持以下请求操作memcache.h 中均有声明对应实现见 memcache.cppbool Get(const butil::StringPiece key); bool Set(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Add(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Replace(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Append(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Prepend(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Delete(const Slice key); bool Flush(uint32_t timeout); bool Increment(const Slice key, uint64_t delta, uint64_t initial_value, uint32_t exptime); bool Decrement(const Slice key, uint64_t delta, uint64_t initial_value, uint32_t exptime); bool Touch(const Slice key, uint32_t exptime); bool Version();各操作语义与参数说明操作参数含义说明Get(key)key读取一个 key 的值。内部走MC_BINARY_GETmemcache.cppSet/Add/Replace/Append/Prependkey, value, flags, exptime, cas_value存储类操作底层共用Store()编码memcache.cpp。Append/Prepend要求 value 非空否则直接返回 falseDelete(key)key删除内部走MC_BINARY_DELETEFlush(timeout)timeout 秒清空缓存timeout 为 0 时不带 extrasmemcache.cppIncrement/Decrementkey, delta, initial_value, exptime计数操作。当 key 不存在时以initial_value为初值创建exptime为过期时间底层共用Counter()编码memcache.cppTouch(key, exptime)key, 新的过期秒数更新过期时间底层走MC_BINARY_TOUCHVersion()无获取服务器版本常用于探活关于 CASData Version Checkcas_value非 0 时操作只有在 item 存在且其 CAS 值与传入值一致时才会成功见 memcache.h 的注释可用于实现乐观锁传 0 表示忽略 CAS 检查。对应的响应操作Pop 系列// 任一操作失败时可调用 response.LastError() 获取错误文本。 bool PopGet(IOBuf* value, uint32_t* flags, uint64_t* cas_value); bool PopGet(std::string* value, uint32_t* flags, uint64_t* cas_value); bool PopSet(uint64_t* cas_value); bool PopAdd(uint64_t* cas_value); bool PopReplace(uint64_t* cas_value); bool PopAppend(uint64_t* cas_value); bool PopPrepend(uint64_t* cas_value); bool PopDelete(); bool PopFlush(); bool PopIncrement(uint64_t* new_value, uint64_t* cas_value); bool PopDecrement(uint64_t* new_value, uint64_t* cas_value); bool PopTouch(); bool PopVersion(std::string* version);其中PopGet提供了IOBuf与std::string两种取值形式memcache.cppIOBuf版本可避免大 value 的拷贝PopGet成功时会同时返回 flags 和 CAS 值。PopIncrement/PopDecrement成功时通过new_value返回计数操作后的新值。响应状态码StatusMemcacheResponse::Status枚举定义了二进制协议中的响应状态码memcache.hstatus_str()给出了对应的人类可读文本memcache.cppSTATUS_SUCCESS 0x00成功STATUS_KEY_ENOENT 0x01key 不存在The key does not existSTATUS_KEY_EEXISTS 0x02key 已存在STATUS_E2BIG 0x03参数过长STATUS_EINVAL 0x04非法参数STATUS_NOT_STORED 0x05未存储STATUS_DELTA_BADVAL 0x06非法 deltaSTATUS_AUTH_ERROR 0x20/STATUS_AUTH_CONTINUE 0x21认证错误/认证继续STATUS_UNKNOWN_COMMAND 0x81未知命令STATUS_ENOMEM 0x82内存不足完整的客户端示例与运行方式example/memcache_c/client.cpp 是一个多线程压测/访问示例展示了真实场景下的完整用法启动时先用--batch * --thread_num个 pipeline SET 请求灌入数据然后启动多个发送线程持续 GET。支持--use_bthread选择用 pthread 还是 bthread 发送请求直观对比两种并发模型的 QPS 与延迟。支持--use_couchbase配合--bucket_name、--bucket_password使用CouchbaseAuthenticator做 SASL 认证见 couchbase_authenticator.cpp。常用命令行参数参数默认值含义--server0.0.0.0:11211memcached 服务器地址--thread_num10发送请求的线程数--use_bthreadfalse是否用 bthread 发送请求--connection_type连接方式single、pooled、short--timeout_ms100RPC 超时毫秒--max_retry3最大重试次数不含首次 RPC--batch1单个 request 内 pipeline 的操作数--key/--valuehello / world读写用的 key 与 value--exptime0过期秒数0 表示永不过期--load_balancer负载均衡算法集群访问时使用--use_couchbasefalse是否使用 Couchbase 认证运行方式假设已完成 brpc 的编译安装cd example/memcache_c make # 依赖仓库根目录的 config.mk 与 output/ 下的 libbrpc ./memcache_client --server127.0.0.1:11211 --thread_num10 --batch4底层协议实现剖析二进制协议封装memcached 二进制协议每个包固定 24 字节头由MemcacheRequestHeader/MemcacheResponseHeader定义memcache_binary_header.h字段包括 magic、command、key_length、extras_length、data_type、vbucket_id/status、total_body_length、opaque、cas_value。源码中通过BAIDU_CASSERT(sizeof(...) 24)在编译期强制校验头部大小memcache_binary_protocol.cpp。magic 值只有两个请求MC_MAGIC_REQUEST 0x80响应MC_MAGIC_RESPONSE 0x81。命令码opcode从 0x00 到 0x24 定义了 GET/SET/ADD/DELETE/INCR/DECR/FLUSH/TOUCH/VERSION 等二十余种命令memcache_binary_header.h。请求序列化时MemcacheRequest直接按上述头格式把操作逐条拼进IOBufStore、GetOrDelete、Counter、Touch、Version等编码函数见 memcache.cpp实现了尽量零拷贝。服务端协议解析与 pipeline 对齐memcache_binary_protocol.cpp 实现了协议处理三件套SerializeMemcacheRequestL197-L210把MemcacheRequest内部已编码好的raw_buffer()原样交给网络层并把pipelined_count记录到 Controller供响应对齐使用。ParseMemcacheMessageL77-L155逐个解析响应的 24 字节头校验 magic 必须是MC_MAGIC_RESPONSEtotal_body_length不得超过全局 flag--max_body_size默认值可在gflags中调整测试中曾将其设为 1024 来验证超限拒绝逻辑见 brpc_memcache_unittest.cpp并根据 socket 上的PipelinedInfo对齐 pipeline 中第几个响应。ProcessMemcacheResponseL157-L195校验响应类型为MemcacheResponse把解析出的原始字节直接移交给响应的raw_buffer()并核对响应的pipelined_count与请求一致不一致则报错。此外PackMemcacheRequestL212-L227支持在请求前附加认证凭据例如 Couchbase bucket 的 SASL 认证GetMemcacheMethodName将方法名统一映射为memcached便于 rpcz 等内置服务展示。测试用例印证test/brpc_memcache_unittest.cpp 中有完整的集成测试会自动拉起本机 memcached监听 11211 端口sanity 用例L160-L229完整演示 Flush、GET 不存在的 keyLastError() Not found、SET 后 GET 回读校验 value 与 flags、以及 CAS 命中/未命中两种场景——CAS 值匹配时 SET 成功不匹配时PopSet返回 false。incr_and_decr 用例L231-L264在同一个 request 里 pipeline 了Increment(2)、Decrement(1)、Increment(3)三个操作验证初始值为 10 时新值依次为 10、9、12说明多操作 pipeline 与逐条 Pop 的完整链路可用。version 用例L266-L284通过PopVersion读取 memcached 版本号。访问 memcached 集群建立一个使用c_md5负载均衡算法的 Channel就可以访问挂载在对应命名服务下的 memcached 集群brpc::ChannelOptions options; options.protocol brpc::PROTOCOL_MEMCACHE; brpc::Channel channel; if (channel.Init(list:///path/to/server_list, c_md5, options) ! 0) { LOG(FATAL) Fail to init channel to memcached cluster; return -1; }这里channel.Init的第一个参数是命名服务地址第二个参数是负载均衡算法名c_md5memcached 集群中每个 key 会按 MD5 哈希一致性映射到具体节点。使用集群时必须注意以下约束每个MemcacheRequest应只包含一个操作或确保其中所有操作使用同一个 key。在当前实现下一个 request 内的多个操作总会发往同一个 server请求体作为一个整体被发送。如果 pipeline 中的 key 分布在集群的不同节点上结果必然错误。这种情况下你必须把一个 request 拆分成多个每个只包含一个操作。这一点在官方文档中有明确说明也与 memcache_binary_protocol.cpp 中请求整体序列化的实现一致。另一种选择是沿用常见的twemproxy方案额外部署代理层客户端像访问单机一样访问集群。该方案需要部署 proxy 并增加一次网络往返的延迟但可以省去客户端侧的拆分逻辑。总结brpc 的 memcache 客户端以协议即一等公民的设计把 memcached 二进制协议无缝纳入 RPC 框架开发者用ChannelMemcacheRequest/MemcacheResponse即可完成单机与集群访问天然获得线程安全、pipeline、超时重试、负载均衡、认证等能力。核心代码集中在 src/brpc/memcache.h 与 src/brpc/memcache.cpp请求/响应编码协议层位于 src/brpc/policy/memcache_binary_protocol.cpp示例与测试则分别见 example/memcache_c/client.cpp 与 test/brpc_memcache_unittest.cpp可作为二次开发与性能调优的直接参考。【免费下载链接】brpcbrpc is an Industrial-grade RPC framework using C Language, which is often used in high performance system such as Search, Storage, Machine learning, Advertisement, Recommendation etc. brpc means better RPC.项目地址: https://gitcode.com/GitHub_Trending/brpc/brpc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 brpc Memcached 客户端完全指南：基于二进制协议的工业级访问方案
brpc Memcached 客户端完全指南基于二进制协议的工业级访问方案【免费下载链接】brpcbrpc is an Industrial-grade RPC framework using C Language, which is often used in high performance system such as Search, Storage, Machine learning, Advertisement, Recommendation etc. brpc means better RPC.项目地址: https://gitcode.com/GitHub_Trending/brpc/brpcbrpc 在 RPC 框架之上直接内置了对 memcached 二进制协议的支持让开发者无需引入额外客户端库即可用统一的方式访问 memcached 单机或集群。本文以 docs/en/memcache_client.md 为骨架结合仓库源码与测试用例系统讲解 Channel 初始化、MemcacheRequest/MemcacheResponse 的使用、pipeline 批量操作、集群访问策略以及底层协议实现原理读完即可上手写出高性能的 memcached 访问代码。为什么要用 brpc 访问 memcachedmemcached 是常用的缓存服务。为了使用户更快捷地访问 memcached 并充分利用 bthread 的并发能力brpc 直接在框架层面支持了 memcache 协议无需依赖第三方客户端。官方示例位于 example/memcache_c。注意brpc 只支持 memcache 的二进制协议binary protocol。memcached 在 1.3 之前只有文本协议如今支持文本协议的意义甚微如果你的 memcached 早于 1.3请升级到新版本。相比使用官方客户端 libmemcachedbrpc 的实现有以下优势文档原文要点线程安全Channel本身是线程安全的不需要为每个线程建立独立的 client多个线程可共享同一个 Channel 并发访问。访问模式丰富支持同步、异步、半同步等访问方式并能通过 ParallelChannel 等组合 Channel 以声明式方式定义复杂的访问模式。复用 brpc 基础设施支持多种连接方式single / pooled / short 等以及超时、backup request、取消、tracing、内置服务等一系列 brpc 提供的成熟能力。有明确的 request 和 response 概念libmemcached 没有这一概念收到的消息不能直接与发出的消息对应用户必须自行做额外的簿记bookkeeping工作且很容易出错而 brpc 中一次CallMethod对应明确的请求与响应。当前实现充分利用了 RPC 的并发机制并尽量避免了拷贝。根据官方文档的记载一个 client 可以轻松把一个同机 memcached 实例版本 1.4.15压到极限单连接约 9 万 QPS多连接约 33 万 QPS。在大部分情况下brpc client 都能充分发挥 memcached 的性能。访问单台 memcached第一步创建 Channel创建一个访问 memcached 的Channel只需把协议设置为PROTOCOL_MEMCACHE#include brpc/memcache.h #include brpc/channel.h brpc::ChannelOptions options; options.protocol brpc::PROTOCOL_MEMCACHE; if (channel.Init(0.0.0.0:11211, options) ! 0) { // 11211 是 memcached 的默认端口 LOG(FATAL) Fail to init channel to memcached; return -1; } ...Channel::Init的第一个参数既可以是ip:port形式的地址也可以是命名服务地址配合集群访问使用。更多连接选项如connection_type、timeout_ms、max_retry可以像 example/memcache_c/client.cpp 中那样在ChannelOptions中设置。第二步发送一个 SET 请求往 memcached 中设置一份数据// 写入 keyhello valueworld flags0xdeadbeef10 秒后失效cas_value 传 0 表示忽略 CAS。 brpc::MemcacheRequest request; brpc::MemcacheResponse response; brpc::Controller cntl; if (!request.Set(hello, world, 0xdeadbeef/*flags*/, 10/*expiring seconds*/, 0/*ignore cas*/)) { LOG(FATAL) Fail to SET request; return -1; } channel.CallMethod(nullptr, cntl, request, response, nullptr/*done*/); if (cntl.Failed()) { LOG(FATAL) Fail to access memcached, cntl.ErrorText(); return -1; } if (!response.PopSet(nullptr)) { LOG(FATAL) Fail to SET memcached, response.LastError(); return -1; } ...上述代码的关键要点请求与响应类型必须固定请求类型必须是MemcacheRequest响应类型必须是MemcacheResponse否则CallMethod会失败。不需要 stub直接调用channel.CallMethodmethod参数填nullptr即可。这背后的校验逻辑在 memcache_binary_protocol.cpp 的SerializeMemcacheRequest中序列化前会检查request-GetDescriptor() ! MemcacheRequest::descriptor()不匹配直接报Must be MemcacheRequest。pipeline 模式调用request.XXX()增加操作本例 XXXSet。一个 request 内多次调用不同的操作这些操作会被同时送到 memcached这就是常说的 pipeline 模式。底层实现是把每个操作编码成独立的二进制协议包依次追加到同一个IOBuf见_buf并由_pipelined_count记录操作个数见 memcache.cpp 中的_pipelined_count。依次 Pop 结果调用response.PopXXX()弹出每个操作的结果本例 XXXSet。成功返回true失败返回false此时调用response.LastError()可获得错误信息。XXX必须和 request 中操作的顺序一一对应否则 Pop 会失败例如把PopSet换成PopGet会得到错误信息not a GET response见 memcache.cpp 中PopGet对 command 的校验。Pop 结果独立于 RPC 结果即使某个值没能成功写入 memcached例如业务层失败RPC 本身可能仍然是成功的。RPC 失败指的是连接断开、超时这类传输层错误。如果业务上要求操作真正成功才算成功那么既要判断 RPC 成功也要判断 PopXXX 成功。完整支持的操作清单目前支持以下请求操作memcache.h 中均有声明对应实现见 memcache.cppbool Get(const butil::StringPiece key); bool Set(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Add(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Replace(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Append(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Prepend(const Slice key, const Slice value, uint32_t flags, uint32_t exptime, uint64_t cas_value); bool Delete(const Slice key); bool Flush(uint32_t timeout); bool Increment(const Slice key, uint64_t delta, uint64_t initial_value, uint32_t exptime); bool Decrement(const Slice key, uint64_t delta, uint64_t initial_value, uint32_t exptime); bool Touch(const Slice key, uint32_t exptime); bool Version();各操作语义与参数说明操作参数含义说明Get(key)key读取一个 key 的值。内部走MC_BINARY_GETmemcache.cppSet/Add/Replace/Append/Prependkey, value, flags, exptime, cas_value存储类操作底层共用Store()编码memcache.cpp。Append/Prepend要求 value 非空否则直接返回 falseDelete(key)key删除内部走MC_BINARY_DELETEFlush(timeout)timeout 秒清空缓存timeout 为 0 时不带 extrasmemcache.cppIncrement/Decrementkey, delta, initial_value, exptime计数操作。当 key 不存在时以initial_value为初值创建exptime为过期时间底层共用Counter()编码memcache.cppTouch(key, exptime)key, 新的过期秒数更新过期时间底层走MC_BINARY_TOUCHVersion()无获取服务器版本常用于探活关于 CASData Version Checkcas_value非 0 时操作只有在 item 存在且其 CAS 值与传入值一致时才会成功见 memcache.h 的注释可用于实现乐观锁传 0 表示忽略 CAS 检查。对应的响应操作Pop 系列// 任一操作失败时可调用 response.LastError() 获取错误文本。 bool PopGet(IOBuf* value, uint32_t* flags, uint64_t* cas_value); bool PopGet(std::string* value, uint32_t* flags, uint64_t* cas_value); bool PopSet(uint64_t* cas_value); bool PopAdd(uint64_t* cas_value); bool PopReplace(uint64_t* cas_value); bool PopAppend(uint64_t* cas_value); bool PopPrepend(uint64_t* cas_value); bool PopDelete(); bool PopFlush(); bool PopIncrement(uint64_t* new_value, uint64_t* cas_value); bool PopDecrement(uint64_t* new_value, uint64_t* cas_value); bool PopTouch(); bool PopVersion(std::string* version);其中PopGet提供了IOBuf与std::string两种取值形式memcache.cppIOBuf版本可避免大 value 的拷贝PopGet成功时会同时返回 flags 和 CAS 值。PopIncrement/PopDecrement成功时通过new_value返回计数操作后的新值。响应状态码StatusMemcacheResponse::Status枚举定义了二进制协议中的响应状态码memcache.hstatus_str()给出了对应的人类可读文本memcache.cppSTATUS_SUCCESS 0x00成功STATUS_KEY_ENOENT 0x01key 不存在The key does not existSTATUS_KEY_EEXISTS 0x02key 已存在STATUS_E2BIG 0x03参数过长STATUS_EINVAL 0x04非法参数STATUS_NOT_STORED 0x05未存储STATUS_DELTA_BADVAL 0x06非法 deltaSTATUS_AUTH_ERROR 0x20/STATUS_AUTH_CONTINUE 0x21认证错误/认证继续STATUS_UNKNOWN_COMMAND 0x81未知命令STATUS_ENOMEM 0x82内存不足完整的客户端示例与运行方式example/memcache_c/client.cpp 是一个多线程压测/访问示例展示了真实场景下的完整用法启动时先用--batch * --thread_num个 pipeline SET 请求灌入数据然后启动多个发送线程持续 GET。支持--use_bthread选择用 pthread 还是 bthread 发送请求直观对比两种并发模型的 QPS 与延迟。支持--use_couchbase配合--bucket_name、--bucket_password使用CouchbaseAuthenticator做 SASL 认证见 couchbase_authenticator.cpp。常用命令行参数参数默认值含义--server0.0.0.0:11211memcached 服务器地址--thread_num10发送请求的线程数--use_bthreadfalse是否用 bthread 发送请求--connection_type连接方式single、pooled、short--timeout_ms100RPC 超时毫秒--max_retry3最大重试次数不含首次 RPC--batch1单个 request 内 pipeline 的操作数--key/--valuehello / world读写用的 key 与 value--exptime0过期秒数0 表示永不过期--load_balancer负载均衡算法集群访问时使用--use_couchbasefalse是否使用 Couchbase 认证运行方式假设已完成 brpc 的编译安装cd example/memcache_c make # 依赖仓库根目录的 config.mk 与 output/ 下的 libbrpc ./memcache_client --server127.0.0.1:11211 --thread_num10 --batch4底层协议实现剖析二进制协议封装memcached 二进制协议每个包固定 24 字节头由MemcacheRequestHeader/MemcacheResponseHeader定义memcache_binary_header.h字段包括 magic、command、key_length、extras_length、data_type、vbucket_id/status、total_body_length、opaque、cas_value。源码中通过BAIDU_CASSERT(sizeof(...) 24)在编译期强制校验头部大小memcache_binary_protocol.cpp。magic 值只有两个请求MC_MAGIC_REQUEST 0x80响应MC_MAGIC_RESPONSE 0x81。命令码opcode从 0x00 到 0x24 定义了 GET/SET/ADD/DELETE/INCR/DECR/FLUSH/TOUCH/VERSION 等二十余种命令memcache_binary_header.h。请求序列化时MemcacheRequest直接按上述头格式把操作逐条拼进IOBufStore、GetOrDelete、Counter、Touch、Version等编码函数见 memcache.cpp实现了尽量零拷贝。服务端协议解析与 pipeline 对齐memcache_binary_protocol.cpp 实现了协议处理三件套SerializeMemcacheRequestL197-L210把MemcacheRequest内部已编码好的raw_buffer()原样交给网络层并把pipelined_count记录到 Controller供响应对齐使用。ParseMemcacheMessageL77-L155逐个解析响应的 24 字节头校验 magic 必须是MC_MAGIC_RESPONSEtotal_body_length不得超过全局 flag--max_body_size默认值可在gflags中调整测试中曾将其设为 1024 来验证超限拒绝逻辑见 brpc_memcache_unittest.cpp并根据 socket 上的PipelinedInfo对齐 pipeline 中第几个响应。ProcessMemcacheResponseL157-L195校验响应类型为MemcacheResponse把解析出的原始字节直接移交给响应的raw_buffer()并核对响应的pipelined_count与请求一致不一致则报错。此外PackMemcacheRequestL212-L227支持在请求前附加认证凭据例如 Couchbase bucket 的 SASL 认证GetMemcacheMethodName将方法名统一映射为memcached便于 rpcz 等内置服务展示。测试用例印证test/brpc_memcache_unittest.cpp 中有完整的集成测试会自动拉起本机 memcached监听 11211 端口sanity 用例L160-L229完整演示 Flush、GET 不存在的 keyLastError() Not found、SET 后 GET 回读校验 value 与 flags、以及 CAS 命中/未命中两种场景——CAS 值匹配时 SET 成功不匹配时PopSet返回 false。incr_and_decr 用例L231-L264在同一个 request 里 pipeline 了Increment(2)、Decrement(1)、Increment(3)三个操作验证初始值为 10 时新值依次为 10、9、12说明多操作 pipeline 与逐条 Pop 的完整链路可用。version 用例L266-L284通过PopVersion读取 memcached 版本号。访问 memcached 集群建立一个使用c_md5负载均衡算法的 Channel就可以访问挂载在对应命名服务下的 memcached 集群brpc::ChannelOptions options; options.protocol brpc::PROTOCOL_MEMCACHE; brpc::Channel channel; if (channel.Init(list:///path/to/server_list, c_md5, options) ! 0) { LOG(FATAL) Fail to init channel to memcached cluster; return -1; }这里channel.Init的第一个参数是命名服务地址第二个参数是负载均衡算法名c_md5memcached 集群中每个 key 会按 MD5 哈希一致性映射到具体节点。使用集群时必须注意以下约束每个MemcacheRequest应只包含一个操作或确保其中所有操作使用同一个 key。在当前实现下一个 request 内的多个操作总会发往同一个 server请求体作为一个整体被发送。如果 pipeline 中的 key 分布在集群的不同节点上结果必然错误。这种情况下你必须把一个 request 拆分成多个每个只包含一个操作。这一点在官方文档中有明确说明也与 memcache_binary_protocol.cpp 中请求整体序列化的实现一致。另一种选择是沿用常见的twemproxy方案额外部署代理层客户端像访问单机一样访问集群。该方案需要部署 proxy 并增加一次网络往返的延迟但可以省去客户端侧的拆分逻辑。总结brpc 的 memcache 客户端以协议即一等公民的设计把 memcached 二进制协议无缝纳入 RPC 框架开发者用ChannelMemcacheRequest/MemcacheResponse即可完成单机与集群访问天然获得线程安全、pipeline、超时重试、负载均衡、认证等能力。核心代码集中在 src/brpc/memcache.h 与 src/brpc/memcache.cpp请求/响应编码协议层位于 src/brpc/policy/memcache_binary_protocol.cpp示例与测试则分别见 example/memcache_c/client.cpp 与 test/brpc_memcache_unittest.cpp可作为二次开发与性能调优的直接参考。【免费下载链接】brpcbrpc is an Industrial-grade RPC framework using C Language, which is often used in high performance system such as Search, Storage, Machine learning, Advertisement, Recommendation etc. brpc means better RPC.项目地址: https://gitcode.com/GitHub_Trending/brpc/brpc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考