day12课上笔记#

内容回顾#

go module#

依赖管理工具。

context#

goroutine管理。

context.Context

两个根节点：context.Background()、context.TODO()

四个方法：context.WithCancel()、context.WithTimeout()、context.WithDeadline()、context.WithValue()

日志收集项目#

为什么要自己写不用ELK?#

ELK: 部署的时候麻烦每一个filebeat都需要配置一个配置文件

使用etcd来管理被收集的日志项。

项目的架构#

1569030448232

上节课项目进度#

kafka：消息队列
tailf：从文件里读日志
go-ini：解析配置文件

今日内容#

etcd#

etcd介绍#

详见群文件：etcd.pdf

安装etcd#

详见：https://docs.qq.com/doc/DTndrQXdXYUxUU09O

go操作etcd#

命令行操作：使用自带的etcdctl

注意一定要使用V3版本的API

先设置环境变量：

1
SET ETCDCTL_API=3 // Windows
2
export ETCDCTL_API=3 // Mac&Linux

PUT：

1
etcdctl.exe --endpoints=http://127.0.0.1:2379 put key value

GET：

1
etcdctl.exe --endpoints=http://127.0.0.1:2379 GET key

DEL:

1
etcdctl.exe --endpoints=http://127.0.0.1:2379 DEL key

go mod设置代理：

1
SET GOPROXY=https://goproxy.cn // Windows
2
export GOPROXY=https://goproxy.cn // Mac&Linux

PUT/GET#

1
package main
2
import (
3
  "context"
4
  "fmt"
5
  "time"
6
  "go.etcd.io/etcd/clientv3"
7
)
8
// etcd client put/get demo
9
// use etcd/clientv3
10
func main() {
11
  cli, err := clientv3.New(clientv3.Config{
12
    Endpoints: []string{"127.0.0.1:2379"},
13
    DialTimeout: 5 * time.Second,
14
  })
15
  if err != nil {
16
    // handle error!
17
    fmt.Printf("connect to etcd failed, err:%v\n", err)
18
    return
19
  }
20
  fmt.Println("connect to etcd success")
21

22
  defer cli.Close()
23

24
  // put
25
  ctx, cancel := context.WithTimeout(context.Background(), time.Second)
26
  _, err = cli.Put(ctx, "baodelu", "dsb")
27
  cancel()
28
  if err != nil {
29
    fmt.Printf("put to etcd failed, err:%v\n", err)
30
    return
31
  }
32
  // get
33
  ctx, cancel = context.WithTimeout(context.Background(), time.Second)
34
  resp, err := cli.Get(ctx, "baodelu")
35
  cancel()
36
  if err != nil {
37
    fmt.Printf("get from etcd failed, err:%v\n", err)
38
    return
39
  }
40
  for _, ev := range resp.Kvs {
41
    fmt.Printf("%s:%s\n", ev.Key, ev.Value)
42
  }
43
}

watch#

非常重要

1
package main
2

3
import (
4
  "context"
5
  "fmt"
6
  "go.etcd.io/etcd/clientv3"
7
  "time"
8
)
9
// etcd watch
10

11

12
func main() {
13
  cli, err := clientv3.New(clientv3.Config{
14
    Endpoints: []string{"127.0.0.1:2379"},
15
    DialTimeout: 5 * time.Second,
16
  })
17
  if err != nil {
18
    // handle error!
19
    fmt.Printf("connect to etcd failed, err:%v\n", err)
20
    return
21
  }
22
  fmt.Println("connect to etcd success")
23

24
  defer cli.Close()
25
  // watch
26
  // 派一个哨兵 一直监视着 luminghui 这个key的变化（新增、修改、删除）
27
  ch := cli.Watch(context.Background(), "luminghui")
28
  // 从通道尝试取值(监视的信息)
29
  for wresp := range ch{
30
    for _, evt := range wresp.Events{
31
      fmt.Printf("Type:%v key:%v value:%v\n", evt.Type, string(evt.Kv.Key), string(evt.Kv.Value))
32
    }
33
  }
34
}

使用etcd优化日志收集项目#

从etcd拉配置#

详见 day12上午代码.zip

如何开启多个tailObj#

详见下午01_logagent_从etcd取配置创建tailtask.zip

如何实现watch配置变更实现热更新#

详见下午02_logagent热更新新配置.zip

根据IP拉取自己的配置#

详见课上最后的代码。

本周任务#

Raft协议
1. 选举
2. 日志复制机制
3. 异常处理（脑裂）
4. zookeeper的zad协议的区别
etcd的watch
1. etcd底层如何实现watch给客户发通知（websocket）
课上的代码写一遍