Go语言实现海量日志收集系统 _Go语言

再次整理了一下这个日志收集系统的框，如下图

文章插图

这次要实现的代码的整体逻辑为：

文章插图

完整代码地址为: https://github.com/Pythonsite/logagent
etcd介绍高可用的分布式key-value存储，可以用于配置共享和服务发现
类似的项目：zookeeper和consul
开发语言：go
接口：提供restful的接口，使用简单
实现算法：基于raft算法的强一致性，高可用的服务存储目录
etcd的应用场景：

服务发现和服务注册
配置中心(我们实现的日志收集客户端需要用到)
分布式锁
master选举

官网对etcd的有一个非常简明的介绍：

文章插图

etcd搭建：
下载地址：https://github.com/coreos/etcd/releases/
根据自己的环境下载对应的版本然后启动起来就可以了
启动之后可以通过如下命令验证一下：

[root@localhost etcd-v3.2.18-linux-amd64]# ./etcdctl set name zhaofan zhaofan[root@localhost etcd-v3.2.18-linux-amd64]# ./etcdctl get namezhaofan[root@localhost etcd-v3.2.18-linux-amd64]#

context 介绍和使用其实这个东西翻译过来就是上下文管理，那么context的作用是做什么，主要有如下两个作用：

控制goroutine的超时
保存上下文数据

通过下面一个简单的例子进行理解：

package mainimport ("fmt""time""net/http""context""io/ioutil")type Result struct{r *http.Responseerr error}func process(){ctx,cancel := context.WithTimeout(context.Background(),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.google.com",nil)if err != nil{fmt.Println("http request failed,err:",err)return}// 如果请求成功了会将数据存入到管道中go func(){resp,err := client.Do(req)pack := Result{resp,err}c <- pack}()select{case <- ctx.Done():tr.CancelRequest(req)fmt.Println("timeout!")case res := <-c:defer res.r.Body.Close()out,_:= ioutil.ReadAll(res.r.Body)fmt.Printf("server response:%s",out)}return}func main() {process()}

写一个通过context保存上下文，代码例子如：

package mainimport ("github.com/Go-zh/net/context""fmt")func add(ctx context.Context,a,b int) int {traceId := ctx.Value("trace_id").(string)fmt.Printf("trace_id:%vn",traceId)return a+b}func calc(ctx context.Context,a, b int) int{traceId := ctx.Value("trace_id").(string)fmt.Printf("trace_id:%vn",traceId)//再将ctx传入到add中return add(ctx,a,b)}func main() {//将ctx传递到calc中ctx := context.WithValue(context.Background(),"trace_id","123456")calc(ctx,20,30)}

结合etcd和context使用关于通过go连接etcd的简单例子：(这里有个小问题需要注意就是etcd的启动方式，默认启动可能会连接不上，尤其你是在虚拟你安装，所以需要通过如下命令启动：
./etcd --listen-client-urls http://0.0.0.0:2371 --advertise-client-urls http://0.0.0.0:2371 --listen-peer-urls http://0.0.0.0:2381
)

package mainimport (etcd_client "github.com/coreos/etcd/clientv3""time""fmt")func main() {cli, err := etcd_client.New(etcd_client.Config{Endpoints:[]string{"192.168.0.118:2371"},DialTimeout:5*time.Second,})if err != nil{fmt.Println("connect failed,err:",err)return}fmt.Println("connect success")defer cli.Close()}

下面一个例子是通过连接etcd，存值并取值

package mainimport ("github.com/coreos/etcd/clientv3""time""fmt""context")func main() {cli,err := clientv3.New(clientv3.Config{Endpoints:[]string{"192.168.0.118:2371"},DialTimeout:5*time.Second,})if err != nil{fmt.Println("connect failed,err:",err)return}fmt.Println("connect succ")defer cli.Close()ctx,cancel := context.WithTimeout(context.Background(),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(context.Background(),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:%sn",ev.Key,ev.Value)}}

关于context官网也有一个例子非常有用，用于控制开启的goroutine的退出，代码如下：