go-etcd

1.etcd brief introduction

Highly available distributed key-val Storage , It can be used to configure sharing and service discovery
Similar projects ：zookeeper(java) and consul(go)
development language ：go
Interface ： Provide restful Of http Interface , Easy to use
Implementation algorithm ： be based on raft Strong consistency of consistency algorithm 、 High availability （ Multiple backups ） Service store directory for .

etcd And redis:
redis Clusters are generally one master and two slaves , Master write data , Read data from , Therefore, there may be information update delay ;etcd The strong consistency of ensures that the heat is consistent with the data of any node . Therefore, the selection with high requirements for strong consistency etcd, Optional if the consistency requirement is not very strict redis,redis Excellent reading speed .

etcd Use scenarios ：
1. Service registration and service discovery
2. Configuration center
3. Distributed lock （ Strong consistency ）
4.master The election

2. etcd install

install etcd+etcdkeeper Collection
start-up etcd, Use etcdctl Execute command operation , Default port "2379"

3.go Use etcd Basic knowledge of

3.1.context Use of the bag ：

context The package defines context. Context type , It carries across API Deadlines between boundaries and processes 、 Cancel the value of the signal and other request ranges . Incoming requests to the server should create a context , Outgoing calls to the server should accept a context . The function call chain between them must propagate Context, You can choose to replace it with WithCancel、WithDeadline、WithTimeout or WithValue Created derivation Context. When a context is cancelled , All contexts derived from it are also cancelled .

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() {
     // The background continues to run 
			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 ")
		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)
}

Output
return
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() {
    
	//  Basics ctx
	baseCtx := context.Background()
	//  Set up 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) // Overtime , Cancel http request ,
		res := <-c            //client.Do(), An error was returned due to the request cancellation 
		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  Inheritable tree structure 
	process(ctx)
}

3.2 etcd Use

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 Use

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
Listening to the configuration / Update node configuration information

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 {
     // Continuous monitoring , Whether the configuration has changed 　watch  Proactively inform that the configuration has changed 
		rch := cli.Watch(context.Background(), "/logagent/conf/") // Block here , There are new changes , Pipeline output information 
		for wresp := range rch {
    
			for _, ev := range wresp.Events {
    
				fmt.Printf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
			}
		}
	}

}