《分布式缓存-原理、架构及Go语言实现》

1
2
3
4
5

type inMemoryCache struct{
    c map[string][]byte // 缓存数据
    mutex sync.RWMutex  // 读写锁
    Stat                // 缓存状态记录
}

1
2
3
4
5
6
7
8
9

HTTP-message=start-line *( header-filed CRLF ) CRLF [ message-body ]
start-line=request-line / status-line
request-line=method SP resquest-status SP HTTP-version CRLF
status-line=HTTP-version SP status-code SP reason-phrase CRL
header-filed=field-name ":" OWS field-value OWS
OWS=*(SP / HTAB)
field-name=token
field-value=*(field-content / obs-fold )
message-body=*OCTET

1
2
3
4
5
6
7
8

command = op key | key-value
op = 'S' | 'G' | 'D'
key = bytes-array
length = 1*DIGIT
content = *OCTET
key-value = length SP length SP content content
response = error | bytes-array
error = '-' bytes-array

1
2
3
4
5
6
7

package main
// #include "test.h"
// #cgo LDFLAGS: libtest.a -lstdc++
import "C"
func main(){
    C.test()
}

1
2
3
4
5

func main(){
    k := C.CString("name") // Go的字符串转C语言的char*
    defer C.free(unsafe.Pointer(k)) // 手动释放内存
    key := C.GoString(k) // C语言的char*转Go语言的字符串
}

1
2
3
4

$ (printf "PING\r\nPING\r\nPING\r\n"; sleep 1) | nc localhost 6379
+PONG
+PONG
+PONG

1
2
3
4
5
6

// 伪代码
for{
    command := connection.ReadCommand()
    result := command.Execute()
    connection.WriteResponse(result)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

resultChan = make(chan result, maxPipelineNum)
// 读取操作
for{
    command := connection.ReadCommand()
    result := command.Execute()
    resultChan <- result
}
// 写入操作
go fun(){
    for{
        result := <-resultChan
        connection.WriteResponse(result)
    }
}()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

writeChan := make(chan request, maxSize);
t := time.NewTimer(time.Second)
for{
    select{
    case p := <-c:
        writeBatch(p)
        t.Reset(time.Second)
    case <-t.C:
        flushBatch(data)
        t.Reset(time.Second)
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// 大致逻辑代码如下
func process(conn net.Conn){
    resultCh := make(chan chan *result, 5000)
    defer close(resultCh)
    go reply(conn, resultCh)            // 异步响应所有的处理结果
    for{
        cmd := readCommand(conn)        // 逐个读取命令
        result := make(chan *result)    // 命令执行的响应结果队列
        resultCh <- result              // 顺序将结果队列放到响应队列
        go func(){
            result <- executeCommand(cmd)   // 异步执行命令并将响应结果放到响应队列
        }()
    }
}

1
2

go get github.com/hashicorp/memberlist
go get stathat.com/c/consistent

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

func NewNode(addr, cluster string) (Node, error){
    // 步骤1： 创建并加入节点
    config := memberlist.DefaultLANConfig()
    config.Name = addr
    config.BindAddr = addr
    l, e := memberlist.Create(config)   // 创建节点
    if cluster == "" {
        cluster = addr
    }
    _, e = l.Join([]string{cluster})    // 加入集群，如果没有设置集群则加入自己，即仅一个节点
    // 步骤2： 获取集群信息并存到本地
    circle := consistent.New()          // 散列环
    circle.NumberOfReplicas = 256       //虚拟节点数量，默认20，改成256
    go fun(){
        for{
            m := l.Members()            // 获取所有节点
            nodes := make([]string ,len(m))
            for i, n := range m{
                nodes[i] = n.Name       // 节点ID
            }
            circle.set(nodes)           // 更新本地散列环的所有节点数据
            time.Sleep(time.Second)
        }
    }()
    return &node(circle, addr), nil
}

1
2
3
4
5

func (n *node) ShouldProcess(key string) (string, bool) {
    addr, _ := n.Get(key)       // 返回目标节点名称（即目标节点的地址）
    return addr, addr == n.addr // 返回目标节点地址，以及是否时当前节点
}
// 使用时判断是否需要处理目标键对应处理，实现拒绝处理该键的逻辑

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

func (h *rebalanceHandler)rebalance() {
    s := h.NewScanner()
    defer s.Close()
    c := &http.Client{}
    for s.Scan() {          // 遍历所有键，Scan方法需要处理Close方法的信号量
        k := s.Key()        // 获取存储数据的键
        n, ok := h.ShouldProcess(k) // 判断键是否在本节点处理
        if !ok {            // 非本节点处理数据则交由目标节点并从本节点删除
            r, _ := http.NewRequest(PUT, url(n), s.Value())
            c.Do(r)
            h.Del(k)
        }
    }
}

1
2
3
4
5
6
7
8

go func(){
    time.Sleep(ttl)
    for k, v := range cached{
        if v.Created.Add(ttl).Before(time.Now()){
            c.Del(k)
        }
    }
}()