go-etcd

1.etcd简介

高可用的分布式key-val存储，可以用于配置共享和服务发现
类似项目：zookeeper(java)和consul(go)
开发语言：go
接口：提供restful的http接口，使用简单
实现算法：基于raft一致性算法的强一致性、高可用（多个备份）的服务存储目录。

etcd 与redis:
redis集群一般是一主两从，主写入数据，从读出数据，故可能存在信息更新延迟；etcd的强一致性确保热和任何节点的数据都是一致的。故对与强一致性要求高的选用etcd，一致性要求不是很严的可选用redis,redis的读取速度优秀。

etcd 使用场景：
1.服务注册和服务发现
2.配置中心
3.分布式锁（强一致性）
4.master选举

2. etcd安装

安装etcd+etcdkeeper合集
启动etcd，使用etcdctl执行命令操作,默认端口"2379"

3.go 使用etcd的基础知识

3.1.context包的使用：

context包定义了context. Context 类型，它携带跨越 API 边界和进程之间的截止日期、取消信号和其他请求范围的值。对服务器的传入请求应该创建一个上下文，对服务器的传出调用应该接受一个上下文。它们之间的函数调用链必须传播 Context，可选择将其替换为使用 WithCancel、WithDeadline、WithTimeout 或 WithValue 创建的派生 Context。当一个上下文被取消时，所有从它派生的上下文也被取消。

3.1.1 WithCancel

package main

import (
	"context"
	"fmt"
	"time"
)

func main() {
    
	// gen generates integers in a separate goroutine and
	// sends them to the returned channel.
	// The callers of gen need to cancel the context once
	// they are done consuming generated integers not to leak
	// the internal goroutine started by gen.
	gen := func(ctx context.Context) <-chan int {
    
		dst := make(chan int)
		n := 1
		go func() {
     //后台持续运行
			for {
    
				select {
    
				case <-ctx.Done():
					fmt.Println("goroutine exit")
					return // returning not to leak the goroutine
				case dst <- n:
					fmt.Println("goroutine run")
					n++
				}
			}
		}()
		fmt.Println("返回")
		return dst
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel() // cancel when we are finished consuming integers

	for n := range gen(ctx) {
    
		fmt.Println(n)
		if n == 5 {
    
			break
		}
	}

	time.Sleep(time.Second)
}

输出
返回
goroutine run
1
2
goroutine run
goroutine run
3
4
goroutine run
goroutine run
5

3.1.2 WithDeadline

package main

import (
	"context"
	"fmt"
	"time"
)

const shortDuration = 1 * time.Millisecond

func main() {
    
	d := time.Now().Add(shortDuration)
	ctx, cancel := context.WithDeadline(context.Background(), d)

	// Even though ctx will be expired, it is good practice to call its
	// cancellation function in any case. Failure to do so may keep the
	// context and its parent alive longer than necessary.
	defer cancel()

	select {
    
	case <-time.After(1 * time.Second):
		fmt.Println("overslept")
	case <-ctx.Done():
		fmt.Println(ctx.Err())
	}

}

3.1.3 WithTimeout

package main

import (
	"context"
	"fmt"
	"io/ioutil"
	"net/http"
	"time"
)

type Result struct {
    
	r   *http.Response
	err error
}

func process() {
    
	// 基础ctx
	baseCtx := context.Background()
	// 设定context.WithTimeout
	ctx, cancel := context.WithTimeout(baseCtx, 2*time.Second)
	defer cancel()

	tr := &http.Transport{
    }
	client := &http.Client{
    Transport: tr}
	c := make(chan Result, 1)
	req, err := http.NewRequest("Get", "http://www.baidu.com", nil) //http://google.com
	if err != nil {
    
		fmt.Println("http request failed, err:", err)
		return
	}

	go func() {
    
		resp, err := client.Do(req)
		pack := Result{
    r: resp, err: err}
		c <- pack
	}()

	select {
    
	case <-ctx.Done():
		tr.CancelRequest(req) //超时，取消http请求，
		res := <-c            //client.Do(),由于请求取消返回一个错误
		fmt.Println("Timeout!,errr:", res.err)
	case res := <-c:
		if res.err != nil {
    
			fmt.Println("Server Response ,client.Do() err:", err)
		} else {
    
			defer res.r.Body.Close()
			out, _ := ioutil.ReadAll(res.r.Body)
			fmt.Printf("Server Response :%s\n", out)
		}

	}
	return

}

func main() {
    
	process()
}

3.1.4 WithValue

package main

import (
	"context"
	"fmt"
)

type favContextKey string

func process1(ctx context.Context, k favContextKey) {
    
	if ret := ctx.Value(k); ret != nil {
    
		fmt.Println("found value:", ret)
		return
	}
	fmt.Println("Not found key", k)
}

func process(ctx context.Context) {
    
	ret, ok := ctx.Value("trace_id").(int)
	if !ok {
    
		ret = 12161844
	}
	fmt.Printf("ret:%d\n", ret)

	if s := ctx.Value("session"); s != nil {
    
		fmt.Println("found value:", s)
	} else {
    
		fmt.Println("not found key:", "session")
	}

}

func main() {
    
	basectx := context.Background()
	// contex.WithValue()
	var k favContextKey = "trace_id"
	ctx := context.WithValue(basectx, k, 12161811)
	process1(ctx, k)
	ctx = context.WithValue(ctx, "session", "day-1") //ctx 可继承的树状结构
	process(ctx)
}

3.2 etcd 使用

3.2.1　etcd_connect

package main

import (
	"fmt"
	"time"

	etcd_client "github.com/coreos/etcd/clientv3"
)

func main() {
    
	cli, err := etcd_client.New(etcd_client.Config{
    
		Endpoints:   []string{
    "localhost:2379", "localhost:22379", "localhost:32379"},
		DialTimeout: 5 * time.Second,
	})

	if err != nil {
    
		fmt.Println("connect failed,err :", err)
		return
	}
	defer cli.Close()
	fmt.Println("connect succ")

}

3.2.2　etcd_example
get/put使用

package main

import (
	"context"
	"fmt"
	"time"

	etcd_client "github.com/coreos/etcd/clientv3"
)

func main() {
    
	cli, err := etcd_client.New(etcd_client.Config{
    
		Endpoints:   []string{
    "localhost:2379", "localhost:22379", "localhost:32379"},
		DialTimeout: 5 * time.Second,
	})

	if err != nil {
    
		fmt.Println("connect failed,err :", err)
		return
	}
	fmt.Println("connect succ")
	defer cli.Close()

	baseCtx := context.Background()
	ctx, cancel := context.WithTimeout(baseCtx, time.Second)
	_, err = cli.Put(ctx, "/logagent/conf/", "sample_value")
	cancel()
	if err != nil {
    
		fmt.Println("put failed,err:", err)
		return
	}

	ctx, cancel = context.WithTimeout(baseCtx, time.Second)
	resp, err := cli.Get(ctx, "/logagent/conf/")
	cancel()
	if err != nil {
    
		fmt.Println("get failed,err:", err)
		return
	}

	for _, ev := range resp.Kvs {
    
		fmt.Printf("%s:%s\n", ev.Key, ev.Value)
	}
}

3.2.3　etcd_wath
监听配置/节点配置信息的更新

package main

import (
	"fmt"
	"time"

	"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
	etcd_client "github.com/coreos/etcd/clientv3"
)

func main() {
    
	cli, err := etcd_client.New(etcd_client.Config{
    
		Endpoints:   []string{
    "localhost:2379", "localhost:22379", "localhost:32379"},
		DialTimeout: 5 * time.Second,
	})

	if err != nil {
    
		fmt.Println("connect failed,err :", err)
		return
	}

	fmt.Println("connect succ")
	defer cli.Close()

	for {
     //持续监听，配置是否有变化　watch 主动告知配置有变
		rch := cli.Watch(context.Background(), "/logagent/conf/") //阻塞在此处，有新变化，管道输出信息
		for wresp := range rch {
    
			for _, ev := range wresp.Events {
    
				fmt.Printf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
			}
		}
	}

}