7天用Go从零实现分布式缓存 GeeCache —— 第五天：注册节点与 HTTP 客户端，打通多节点远程通信

• 示例工程

本文是「7天用 Go 从零实现分布式缓存 GeeCache」系列的第五篇，主题是为 GeeCache 增加分布式节点能力：借助第四天实现的一致性哈希算法注册节点、选择节点，并编写约 90 行的 HTTP 客户端与远程节点服务端通信，从而让缓存查询可以从「单机命中」扩展到「多节点协作」。读完本文，你将掌握PeerPicker/PeerGetter两个核心接口的设计思路、HTTPPool如何同时充当服务端与客户端，以及如何用 3 个本地端口 + 1 个 API 端口完整跑通一套多节点缓存集群。

1 流程回顾：今天要补上第 ⑵ 步

在 GeeCache 的整体设计中，一次key查询遵循如下流程：

是 接收 key --> 检查是否被缓存 -----> 返回缓存值 ⑴ | 否 是 |-----> 是否应当从远程节点获取 -----> 与远程节点交互 --> 返回缓存值 ⑵ | 否 |-----> 调用`回调函数`，获取值并添加到缓存 --> 返回缓存值 ⑶

前几天的实现已经完成了：

• 流程 ⑴：从本地缓存命中并返回（Group.Get中先查mainCache，见 geecache.go）；
• 流程 ⑶：未命中时调用回调函数从数据源加载，并回填缓存（getLocally+populateCache）。

今天要实现的正是流程 ⑵：从远程节点获取缓存值。将它进一步细化：

使用一致性哈希选择节点 是 是 |-----> 是否是远程节点 -----> HTTP 客户端访问远程节点 --> 成功？-----> 服务端返回返回值 | 否 ↓ 否 |----------------------------> 回退到本地节点处理。

可以看到，这一步需要解决三个问题：如何抽象"选择节点"和"访问节点"的行为（接口设计）、如何用 HTTP 与远程节点通信（客户端实现）、如何把这两件事接入主流程（load改造）。

2 抽象 PeerPicker 与 PeerGetter 两个接口

首先要做的是面向接口编程：把"根据 key 选择节点"和"从节点获取数据"两个行为抽象出来，而不是直接依赖具体的 HTTP 实现。为此，新增 peers.go：

package geecache // PeerPicker is the interface that must be implemented to locate // the peer that owns a specific key. type PeerPicker interface { PickPeer(key string) (peer PeerGetter, ok bool) } // PeerGetter is the interface that must be implemented by a peer. type PeerGetter interface { Get(group string, key string) ([]byte, error) }

• PeerPicker：负责"节点选择"。PickPeer(key)根据传入的 key 返回应当由哪个节点（PeerGetter）处理该 key，以及是否命中了一个有效的远程节点（ok）。
• PeerGetter：负责"节点数据访问"，也就是流程中的 HTTP 客户端。Get(group, key)从指定 group 中查找 key 对应的缓存值，返回[]byte。

这里的ok布尔返回值很关键：当 key 经一致性哈希计算后落在本机节点上时，PickPeer应当返回ok=false，由调用方回退到本地处理，避免节点间无限转发。

3 节点选择与 HTTP 客户端

第三天实现中，HTTPPool已经通过ServeHTTP提供了服务端能力（解析/<basepath>/<groupname>/<key>路径、调用group.Get、返回字节流，见 http.go）。但通信是双向的，服务端还需要配套的客户端。本节为HTTPPool补齐客户端与节点选择能力。

3.1 第一步：httpGetter 实现 PeerGetter

创建具体的 HTTP 客户端类httpGetter，实现PeerGetter接口（源码见 http.go）：

type httpGetter struct { baseURL string } func (h *httpGetter) Get(group string, key string) ([]byte, error) { u := fmt.Sprintf( "%v%v/%v", h.baseURL, url.QueryEscape(group), url.QueryEscape(key), ) res, err := http.Get(u) if err != nil { return nil, err } defer res.Body.Close() if res.StatusCode != http.StatusOK { return nil, fmt.Errorf("server returned: %v", res.Status) } bytes, err := ioutil.ReadAll(res.Body) if err != nil { return nil, fmt.Errorf("reading response body: %v", err) } return bytes, nil } var _ PeerGetter = (*httpGetter)(nil)

• baseURL表示将要访问的远程节点的地址前缀，例如http://example.com/_geecache/。注意它以basePath（/_geecache/）结尾，与服务端的路由前缀严格对应；
• 拼接 URL 时使用url.QueryEscape对group与key做转义，保证 key 中即使含特殊字符也不会破坏路径结构；
• 使用http.Get()发起请求，校验状态码为200后，通过ioutil.ReadAll读取响应体并转换为[]byte；
• 末尾的var _ PeerGetter = (*httpGetter)(nil)是编译期断言，确保httpGetter确实实现了PeerGetter接口，一旦接口签名变化，编译立刻报错。

3.2 第二步：HTTPPool 增加节点注册与选择

接着为HTTPPool增加节点管理成员（完整定义见 http.go）：

const ( defaultBasePath = "/_geecache/" defaultReplicas = 50 ) // HTTPPool implements PeerPicker for a pool of HTTP peers. type HTTPPool struct { // this peer's base URL, e.g. "https://example.net:8000" self string basePath string mu sync.Mutex // guards peers and httpGetters peers *consistenthash.Map httpGetters map[string]*httpGetter // keyed by e.g. "http://10.0.0.2:8008" }

新增的两个成员变量：

• peers：类型为一致性哈希算法的Map（第四天实现），负责根据具体的 key 选择节点；
• httpGetters：映射「远程节点地址 → 对应的httpGetter」。每个远程节点都对应一个独立的httpGetter，因为httpGetter携带的baseURL与节点地址强相关；
• mu sync.Mutex：保护peers与httpGetters的并发读写，因为Set/PickPeer可能在多 goroutine 下被同时调用；
• defaultReplicas = 50表示每个真实节点在哈希环上生成 50 个虚拟节点，虚拟节点越多，节点增减时数据迁移的均衡性越好。

3.3 第三步：实现 PeerPicker 接口

为HTTPPool实现Set与PickPeer两个方法（源码见 http.go）：

// Set updates the pool's list of peers. func (p *HTTPPool) Set(peers ...string) { p.mu.Lock() defer p.mu.Unlock() p.peers = consistenthash.New(defaultReplicas, nil) p.peers.Add(peers...) p.httpGetters = make(map[string]*httpGetter, len(peers)) for _, peer := range peers { p.httpGetters[peer] = &httpGetter{baseURL: peer + p.basePath} } } // PickPeer picks a peer according to key func (p *HTTPPool) PickPeer(key string) (PeerGetter, bool) { p.mu.Lock() defer p.mu.Unlock() if peer := p.peers.Get(key); peer != "" && peer != p.self { p.Log("Pick peer %s", peer) return p.httpGetters[peer], true } return nil, false } var _ PeerPicker = (*HTTPPool)(nil)

• Set()：实例化一致性哈希（虚拟节点数 50，哈希函数默认取crc32.ChecksumIEEE），把传入的全部节点地址加入哈希环；同时为每个节点创建对应的httpGetter，其baseURL由节点地址 + basePath拼接而成；
• PickPeer()：包装一致性哈希的Get()方法，根据 key 选出节点；关键判断是peer != p.self——如果选出的节点就是本机，则返回(nil, false)，由主流程回退到本地处理，避免节点自我递归请求。

至此，HTTPPool一身二任：既具备提供 HTTP 服务的能力（ServeHTTP），也具备根据具体 key 创建 HTTP 客户端、从远程节点拉取缓存值的能力（PickPeer+httpGetter）。

关于一致性哈希的行为，可以参考其单元测试 consistenthash_test.go：用注入的假哈希函数验证"添加节点后 key 映射到最近虚拟节点、新增节点只影响少量 key"等特性，这也是分布式缓存中节点变化时最小化数据迁移的基石。

4 集成主流程：RegisterPeers 与 load 改造

最后，把上述能力接入Group的主流程（geecache.go）。Group新增peers PeerPicker字段，并增加两个方法、修改一个方法：

// A Group is a cache namespace and associated data loaded spread over type Group struct { name string getter Getter mainCache cache peers PeerPicker } // RegisterPeers registers a PeerPicker for choosing remote peer func (g *Group) RegisterPeers(peers PeerPicker) { if g.peers != nil { panic("RegisterPeerPicker called more than once") } g.peers = peers } func (g *Group) load(key string) (value ByteView, err error) { if g.peers != nil { if peer, ok := g.peers.PickPeer(key); ok { if value, err = g.getFromPeer(peer, key); err == nil { return value, nil } log.Println("[GeeCache] Failed to get from peer", err) } } return g.getLocally(key) } func (g *Group) getFromPeer(peer PeerGetter, key string) (ByteView, error) { bytes, err := peer.Get(g.name, key) if err != nil { return ByteView{}, err } return ByteView{b: bytes}, nil }

三个关键改动：

• RegisterPeers()：将实现了PeerPicker接口的HTTPPool注入到Group中；用panic防止重复注册，保证节点配置只生效一次；
• getFromPeer()：使用实现了PeerGetter的httpGetter访问远程节点，将返回的[]byte包装成不可变的ByteView（ByteView提供了Len / ByteSlice / String等只读访问方法，见 byteview.go）；
• load()改造：先判断是否配置了节点选择器，若是则调用PickPeer(key)选节点；选中的是远程节点则调用getFromPeer()；若请求失败，则打印日志并回退到getLocally()本地处理——这是分布式缓存的重要容错设计：单个远程节点故障不应拖垮整个查询链路。

5 main 函数测试：3 节点集群 + API 服务

5.1 main 函数结构

main.go 的代码比较多，但逻辑非常简单：启动两种角色——用户不感知的缓存服务器（8001/8002/8003）和用户感知的 API 服务（9999）。

var db = map[string]string{ "Tom": "630", "Jack": "589", "Sam": "567", } func createGroup() *geecache.Group { return geecache.NewGroup("scores", 2<<10, geecache.GetterFunc( func(key string) ([]byte, error) { log.Println("[SlowDB] search key", key) if v, ok := db[key]; ok { return []byte(v), nil } return nil, fmt.Errorf("%s not exist", key) })) } func startCacheServer(addr string, addrs []string, gee *geecache.Group) { peers := geecache.NewHTTPPool(addr) peers.Set(addrs...) gee.RegisterPeers(peers) log.Println("geecache is running at", addr) log.Fatal(http.ListenAndServe(addr[7:], peers)) } func startAPIServer(apiAddr string, gee *geecache.Group) { http.Handle("/api", http.HandlerFunc( func(w http.ResponseWriter, r *http.Request) { key := r.URL.Query().Get("key") view, err := gee.Get(key) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/octet-stream") w.Write(view.ByteSlice()) })) log.Println("fontend server is running at", apiAddr) log.Fatal(http.ListenAndServe(apiAddr[7:], nil)) } func main() { var port int var api bool flag.IntVar(&port, "port", 8001, "Geecache server port") flag.BoolVar(&api, "api", false, "Start a api server?") flag.Parse() apiAddr := "http://localhost:9999" addrMap := map[int]string{ 8001: "http://localhost:8001", 8002: "http://localhost:8002", 8003: "http://localhost:8003", } var addrs []string for _, v := range addrMap { addrs = append(addrs, v) } gee := createGroup() if api { go startAPIServer(apiAddr, gee) } startCacheServer(addrMap[port], addrs, gee) }

各部分职责：

• createGroup()：创建名为scores的缓存组，容量 2<<10（2048 字节，LRU 淘汰），数据源是内存 map 模拟的「慢数据库」；
• startCacheServer()：创建HTTPPool，Set(addrs...)注册全部 3 个节点，RegisterPeers注入到 Group，最后用http.ListenAndServe(addr[7:], peers)启动缓存服务——addr[7:]截掉http://前缀得到监听地址，peers同时充当 Handler（即ServeHTTP）；
• startAPIServer()：在 9999 端口注册/api路由，解析?key=查询参数后调用gee.Get(key)，成功则返回字节流，失败返回 500；
• main()：通过命令行参数-port指定缓存端口、-api=1决定是否同时启动 API 服务；三台缓存节点共用同一个addrs列表，因此每个节点都知道完整的集群拓扑。

需要注意模块组织：day5 采用独立 module 结构，go.mod 中通过replace geecache => ./geecache将geecache包指向本地子目录，主程序与缓存库分离。

5.2 run.sh：一键启动集群并压测

为了方便，将启动命令封装为 shell 脚本 run.sh：

#!/bin/bash trap "rm server;kill 0" EXIT go build -o server ./server -port=8001 & ./server -port=8002 & ./server -port=8003 -api=1 & sleep 2 echo ">>> start test" curl "http://localhost:9999/api?key=Tom" & curl "http://localhost:9999/api?key=Tom" & curl "http://localhost:9999/api?key=Tom" & wait

• trap "rm server;kill 0" EXIT用于在 shell 脚本退出时删除临时编译产物并结束全部子进程，避免残留后台服务；
• 依次在 8001、8002、8003 启动缓存节点，其中 8003 同时开启 API 服务（9999）；
• 等待 2 秒后并发发起 3 个?key=Tom请求进行测试。

运行./run.sh，输出如下：

$ ./run.sh 2020/02/16 21:17:43 geecache is running at http://localhost:8001 2020/02/16 21:17:43 geecache is running at http://localhost:8002 2020/02/16 21:17:43 geecache is running at http://localhost:8003 2020/02/16 21:17:43 fontend server is running at http://localhost:9999 >>> start test 2020/02/16 21:17:45 [Server http://localhost:8003] Pick peer http://localhost:8001 2020/02/16 21:17:45 [Server http://localhost:8003] Pick peer http://localhost:8001 2020/02/16 21:17:45 [Server http://localhost:8003] Pick peer http://localhost:8001 ... 630630630

此时可以另开一个 shell 手动验证：

$ curl "http://localhost:9999/api?key=Tom" 630 $ curl "http://localhost:9999/api?key=kkk" kkk not exist

日志清晰地展示了分布式协作过程：API 服务（运行在 8003 节点上）收到 3 个并发请求后，经一致性哈希全部选择了节点 8001，由 8001 的服务端返回缓存值，3 次请求最终输出630630630——节点选择与远程 HTTP 通信已经全链路打通。

5.3 暴露的问题：缓存击穿风险

测试时并发 3 个?key=Tom请求，日志显示三次都选中了节点 8001，这是一致性哈希算法"相同 key 稳定映射到相同节点"的功劳。但这同时暴露了一个隐患：

假如有 10 万个并发请求同一数据，就会向 8001 同时发起 10 万次请求；如果 8001 此时又同时向数据库发起 10 万次查询，极易导致缓存被击穿（cache breakdown）。

三次请求结果一致，对相同的 key 完全可以在第一次请求后就只发一次远程请求，其余请求共享结果。这正是第六天singleflight（请求合并）要解决的问题，本文暂不展开。

6 小结

第五天为 GeeCache 补上了分布式缓存最关键的一环：

至此，GeeCache 已经具备完整的三级查询能力：本地缓存命中 → 远程节点获取 → 回调数据源加载并回填。下一步（第六天）将针对并发重复请求带来的缓存击穿问题，引入 singleflight 机制做请求合并。

进一步阅读：系列前文可参考 geecache-day4.md（一致性哈希）、geecache-day3.md（HTTP 服务端）；后续演进见 geecache-day6.md（singleflight）。

• 示例工程