本站资源收集于互联网,不提供软件存储服务,每天免费更新优质的软件以及学习资源!

go语言实现日志收集系统图文详解

网络教程 app 1℃

go语言实现日志收集系统图文详解

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

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

完整代码地址为: github./pythonsite/logagent

etcd介绍

高可用的分布式key-value存储,可以用于配置共享和服务发现

类似的项目:zookeeper和consul

开发语言:go

接口:提供restful的接口,使用简单

实现算法:基于raft算法的强一致性,高可用的服务存储目录

etcd的应用场景:

1、服务发现和服务注册

2、配置中心(我们实现的日志收集客户端需要用到)

3、分布式锁

4、master选举

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

etcd搭建:
下载地址:github./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的作用是做什么,主要有如下两个作用:

1、控制goroutine的超时

2、保存上下文数据

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

package mainimport (    "fmt"    "time"    "net/http"    "context"    "io/ioutil")type Result struct{    r *http.Response    err error}func process(){    ctx,cancel := context.WithTimeout(context.Background(),2*time.Second)    defer cancel()    tr := &amp;http.Transport{}    client := &amp;http.Client{Transport:tr}    c := make(chan Result,1)    req,err := http.NewRequest("GET","www.google.",nil)    if err != nil{        fmt.Println("http request failed,err:",err)        return    }    // 如果请求成功了会将数据存入到管道中    go func(){        resp,err := client.Do(req)        pack := Result{resp,err}        c <p>写一个通过context保存上下文,代码例子如:</p><pre class="brush:js;toolbar:false">package mainimport (    "github./Go-zh/net/context"    "fmt")func add(ctx context.Context,a,b int) int {    traceId := ctx.Value("trace_id").(string)    fmt.Printf("trace_id:%v",traceId)    return a+b}func calc(ctx context.Context,a, b int) int{    traceId := ctx.Value("trace_id").(string)    fmt.Printf("trace_id:%v",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 0.0.0.0:2371 –advertise-client-urls 0.0.0.0:2371 –listen-peer-urls 0.0.0.0:2381
)

package mainimport (    etcd_client "github./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./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:%s",ev.Key,ev.Value)    }}

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

package mainimport (    "context"    "fmt")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) <p>关于官网文档中的WithDeadline演示的代码例子:</p><pre class="brush:js;toolbar:false">package mainimport (    "context"    "fmt"    "time")func main() {    d := time.Now().Add(50 * time.Millisecond)    ctx, cancel := context.WithDeadline(context.Background(), d)    // Even though ctx will be expired, it is good practice to call its    // cancelation function in any case. Failure to do so may keep the    // context and its parent alive longer than necessary.    defer cancel()    select {    case <p>通过上面的代码有了一个基本的使用,那么如果我们通过etcd来做配置管理,如果配置更改之后,我们如何通知对应的服务器配置更改,通过下面例子演示:</p><pre class="brush:js;toolbar:false">package mainimport (    "github./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    }    defer cli.Close()    // 这里会阻塞    rch := cli.Watch(context.Background(),"logagent/conf/")    for wresp := range rch{        for _,ev := range wresp.Events{            fmt.Printf("%s %q : %q", ev.Type, ev.Kv.Key, ev.Kv.Value)        }    }}

实现一个kafka的消费者代码的简单例子:

package mainimport (    "github./Shopify/sarama"    "strings"    "fmt"    "time")func main() {    consumer,err := sarama.NewConsumer(strings.Split("192.168.0.118:9092",","),nil)    if err != nil{        fmt.Println("failed to start consumer:",err)        return    }    partitionList,err := consumer.Partitions("nginx_log")    if err != nil {        fmt.Println("Failed to get the list of partitions:",err)        return    }    fmt.Println(partitionList)    for partition := range partitionList{        pc,err := consumer.ConsumePartition("nginx_log",int32(partition),sarama.OffsetNewest)        if err != nil {            fmt.Printf("failed to start consumer for partition %d:%s",partition,err)            return        }        defer pc.AsyncClose()        go func(partitionConsumer sarama.PartitionConsumer){            for msg := range pc.Messages(){                fmt.Printf("partition:%d Offset:%d Key:%s Value:%s",msg.Partition,msg.Offset,string(msg.Key),string(msg.Value))            }        }(pc)    }    time.Sleep(time.Hour)    consumer.Close()}

但是上面的代码并不是最佳代码,因为我们最后是通过time.sleep等待goroutine的执行,我们可以更改为通过sync.WaitGroup方式实现

package mainimport (    "github./Shopify/sarama"    "strings"    "fmt"    "sync")var (    wg sync.WaitGroup)func main() {    consumer,err := sarama.NewConsumer(strings.Split("192.168.0.118:9092",","),nil)    if err != nil{        fmt.Println("failed to start consumer:",err)        return    }    partitionList,err := consumer.Partitions("nginx_log")    if err != nil {        fmt.Println("Failed to get the list of partitions:",err)        return    }    fmt.Println(partitionList)    for partition := range partitionList{        pc,err := consumer.ConsumePartition("nginx_log",int32(partition),sarama.OffsetNewest)        if err != nil {            fmt.Printf("failed to start consumer for partition %d:%s",partition,err)            return        }        defer pc.AsyncClose()        go func(partitionConsumer sarama.PartitionConsumer){            wg.Add(1)            for msg := range partitionConsumer.Messages(){                fmt.Printf("partition:%d Offset:%d Key:%s Value:%s",msg.Partition,msg.Offset,string(msg.Key),string(msg.Value))            }            wg.Done()        }(pc)    }    //time.Sleep(time.Hour)    wg.Wait()    consumer.Close()}

将客户端需要收集的日志信息放到etcd中

关于etcd处理的代码为:

package mainimport (    "github./coreos/etcd/clientv3"    "time"    "github./astaxie/beego/logs"    "context"    "fmt")var Client *clientv3.Clientvar logConfChan chan string// 初始化etcdfunc initEtcd(addr []string,keyfmt string,timeout time.Duration)(err error){    var keys []string    for _,ip := range ipArrays{        //keyfmt = /logagent/%s/log_config        keys = append(keys,fmt.Sprintf(keyfmt,ip))    }    logConfChan = make(chan string,10)    logs.Debug("etcd watch key:%v timeout:%v", keys, timeout)    Client,err = clientv3.New(clientv3.Config{        Endpoints:addr,        DialTimeout: timeout,    })    if err != nil{        logs.Error("connect failed,err:%v",err)        return    }    logs.Debug("init etcd success")    waitGroup.Add(1)    for _, key := range keys{        ctx,cancel := context.WithTimeout(context.Background(),2*time.Second)        // 从etcd中获取要收集日志的信息        resp,err := Client.Get(ctx,key)        cancel()        if err != nil {            logs.Warn("get key %s failed,err:%v",key,err)            continue        }        for _, ev := range resp.Kvs{            logs.Debug("%q : %q",  ev.Key, ev.Value)            logConfChan <p>同样的这里增加对了限速的处理,毕竟日志收集程序不能影响了当前业务的性能,所以增加了limit.go用于限制速度:</p><pre class="brush:js;toolbar:false">package mainimport (    "time"    "sync/atomic"    "github./astaxie/beego/logs")type SecondLimit struct {    unixSecond int64    curCount int32    limit int32}func NewSecondLimit(limit int32) *SecondLimit {    secLimit := &amp;SecondLimit{        unixSecond:time.Now().Unix(),        curCount:0,        limit:limit,    }    return secLimit}func (s *SecondLimit) Add(count int) {    sec := time.Now().Unix()    if sec == s.unixSecond {        atomic.AddInt32(&amp;s.curCount,int32(count))        return    }    atomic.StoreInt64(&amp;s.unixSecond,sec)    atomic.StoreInt32(&amp;s.curCount, int32(count))}func (s *SecondLimit) Wait()bool {    for {        sec := time.Now().Unix()        if (sec == atomic.LoadInt64(&amp;s.unixSecond)) &amp;&amp; s.curCount == s.limit {            time.Sleep(time.Microsecond)            logs.Debug("limit is running,limit:%d s.curCount:%d",s.limit,s.curCount)            continue        }        if sec != atomic.LoadInt64(&amp;s.unixSecond) {            atomic.StoreInt64(&amp;s.unixSecond,sec)            atomic.StoreInt32(&amp;s.curCount,0)        }        logs.Debug("limit is exited")        return false    }}

推荐:go语言教程

以上就是go语言实现日志收集系统图文详解的详细内容,更多请关注范的资源库其它相关文章!

转载请注明:范的资源库 » go语言实现日志收集系统图文详解

喜欢 (0)