Golang监听日志文件并发送到kafka中

本篇文章给大家分享《Golang监听日志文件并发送到kafka中》，覆盖了Golang的常见基础知识，其实一个语言的全部知识点一篇文章是不可能说完的，但希望通过这些问题，让读者对自己的掌握程度有一定的认识(B 数)，从而弥补自己的不足，更好的掌握它。

前言

日志收集项目的准备中，本文主要讲的是利用golang的tail库，监听日志文件的变动，将日志信息发送到kafka中。

涉及的golang库和可视化工具：

go-ini,sarama,tail其中:

• go-ini：用于读取配置文件，统一管理配置项，有利于后其的维护
• sarama：是一个go操作kafka的客户端。目前我用于向kefka发送消息
• tail：类似于linux的tail命令了，读取文件的后几行。如果文件有追加数据，会检测到。就是通过它来监听日志文件

可视化工具：

offsetexplorer:是kafka的可视化工具，这里用来查看消息是否投递成功

工作的流程

• 加载配置，初始化sarama和kafka。
• 起一个的协程，利用tail不断去监听日志文件的变化。
• 主协程中一直阻塞等待tail发送消息，两者通过一个管道通讯。一旦主协程接收到新日志，组装格式，然后发送到kafka中

环境准备

环境的话，确保zookeeper和kafka正常运行。因为还没有使用sarama读取数据，使用offsetexplorer来查看任务是否真的投递成功了。

代码分层

serve来存放写tail服务类和sarama服务类，conf存放ini配置文件

main函数为程序入口

关键的代码

main.go

main函数做的有：构建配置结构体，映射配置文件。调用和初始化tail,srama服务。

package main

import (
	"fmt"
	"sarama/serve"

	"github.com/go-ini/ini"
)

type KafkaConfig struct {
	Address     string `ini:"address"`
	ChannelSize int    `ini:"chan_size"`
}
type TailConfig struct {
	Path     string `ini:"path"`
	Filename string `ini:"fileName"`
	// 如果是结构体，则指明分区名
	Children `ini:"tailfile.children"`
}
type Config struct {
	KafkaConfig `ini:"kafka"`
	TailConfig  `ini:"tailfile"`
}
type Children struct {
	Name string `ini:"name"`
}

func main() {
	// 加载配置
	var cfg = new(Config)
	err := ini.MapTo(cfg, "./conf/go-conf.ini")
	if err != nil {
		fmt.Print(err)
	}
	// 初始化kafka
	ks := &serve.KafukaServe{}
	// 启动kafka消息监听。异步
	ks.InitKafka([]string{cfg.KafkaConfig.Address}, int64(cfg.KafkaConfig.ChannelSize))
	// 关闭主协程时，关闭channel
	defer ks.Destruct()

	// 初始化tail
	ts := &serve.TailServe{}
	ts.TailInit(cfg.TailConfig.Path + "/" + cfg.TailConfig.Filename)
	// 阻塞
	ts.Listener(ks.MsgChan)

}

kafka.go

有3个方法 ：

• InitKafka，组装配置项以及初始化接收消息的管道，
• Listener,监听管道消息，收到消息后，将消息组装，发送到kafka
• Destruct, 关闭管道

package serve

import (
	"fmt"

	"github.com/Shopify/sarama"
)

type KafukaServe struct {
	MsgChan chan string
	//err         error
}

func (ks *KafukaServe) InitKafka(addr []string, chanSize int64) {

	// 读取配置
	config := sarama.NewConfig()
	// 1. 初始化生产者配置
	config.Producer.RequiredAcks = sarama.WaitForAll
	// 选择分区
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	// 成功交付的信息
	config.Producer.Return.Successes = true

	ks.MsgChan = make(chan string, chanSize)

	go ks.Listener(addr, chanSize, config)

}

func (ks *KafukaServe) Listener(addr []string, chanSize int64, config *sarama.Config) {
	//  连接kafka
	var kafkaClient, _ = sarama.NewSyncProducer(addr, config)
	defer kafkaClient.Close()
	for {
		select {
		case content := 
<h3>tail.go</h3>
<p><strong>主要包括了两个方法：</strong></p>

• TailInit初始化，组装tail配置。Listener
• Listener,保存kafka服务类初始化之后的管道。监听日志文件，如果有新日志，就往管道里发送

package serve

import (
	"fmt"

	"github.com/hpcloud/tail"
)

type TailServe struct {
	tails *tail.Tail
}

func (ts *TailServe) TailInit(filenName string) {
	config := tail.Config{
		ReOpen:    true,
		Follow:    true,
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
		MustExist: false,
		Poll:      true,
	}
	// 打开文件开始读取数据

	ts.tails, _ = tail.TailFile(filenName, config)

	// if err != nil {
	// 	fmt.Println("tails %s failed,err:%v\n", filenName, err)
	// 	return nil, err
	// }
	fmt.Println("启动," + filenName + "监听")
}

func (ts *TailServe) Listener(MsgChan chan string) {
	for {
		msg, ok := 
<p>好了，本文到此结束，带大家了解了《Golang监听日志文件并发送到kafka中》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！</p>