上节回顾
在上一篇文章中我们介绍了编写客户端的四个步骤，分别是：
- 读取配置文件，寻找日志路径
- 初始化服务
- 根据日志路径l来收集日志
- 将收集到的日志发送Kafka中
关于上述的内容博主画了一个思维导图(有点丑，大家勉强看看，以前没画过)：
对了，为了画这个思维导图昨天博主找了好久思维导图的软件，最后发现了Vscode上面有一个非常不错的插件：drawio，样子大概是这样的：
大家如果没有合适的思维导图绘制根据，可以试试这个。好了，话不多说，开始今天的内容。
读取配置信息，获取日志信息
前言
这里读取日志信息我们选择的是go-ini这一第三方包，具体的使用方法在我前面的博文这种有所介绍，大家不了解的话可以参考：
go语言并发实战——日志收集系统(五) 基于go-ini包读取日志收集服务的配置文件
需求分析
这里配置文件中我们主要要知道两个消息，一个Kafka的配置信息，一个是日志文件的路径，配置文件应该是这样的：
[kafka]
address=127.0.0.1:9092
topic=web.log
chan_size=100000
[collect]
logfile_path:G:\goproject\-goroutine-\log-agent\log\log1
而为了方便我们利用反射来读取配置文件，我们来创建几个结构体来存储我们读到的配置信息：
- Kafka结构体
type Kafkaddress struct {
Addr []string `ini:"address"`
Topic string `ini:"topic"`
MessageSize int64 `ini:"chan_size"`
}
- tail结构体
type LogFilePath struct {
Path string `ini:"logfile_path"`
}
- 总的结构体
type Config struct {
Kafakaddress Kafkaddress `ini:"kafka"`
LogFilePath LogFilePath `ini:"collect"`
}
然后读取配置信息放入结构体中：
//读取配置文件，获取配置信息
filename := "G:\\goproject\\-goroutine-\\log-agent\\conf\\config.ini"
ConfigObj := new(Config)
err := ini.MapTo(ConfigObj, filename)
if err != nil {
logrus.Error("%s Load failed,err:", filename, err)
}
这样我们就获得我们所需要的配置消息了
初始化服务
前言
这里我们初始服务主要是初始化Kafka以及tail包，利用它们读取日志信息并将其发送Kafka中,具体介绍可以参考前面的几篇文章：
go语言并发实战——日志收集系统(三) 利用sarama包连接KafKa实现消息的生产与消费
go语言并发实战——日志收集系统(四) 利用tail包实现对日志文件的实时监控
Kafka的初始化
//初始化Kafka
err = Kafka.InitKafka(ConfigObj.Kafakaddress.Addr, ConfigObj.Kafakaddress.MessageSize)
if err != nil {
logrus.Error("InitKafka failed, err:%v", err)
return
}
logrus.Infof("InitKafka success")
tail的初始化
func InitTail(filename string) (err error) {
config := tail.Config{
Follow: true,
ReOpen: true,
MustExist: true,
Poll: true,
Location: &tail.SeekInfo{Offset: 0, Whence: 2},
}
TailObj, err = tail.TailFile(filename, config)
if err != nil {
logrus.Error("tail create tailObj for path:%s,err:%v", filename, err)
return
}
return
}
根据路径来读取日志
需求分析
一般我们常见的想法会是我们先将日志消息读取出来然后发送给Kafka但是这样的串行操作无疑会大大增加程序的运行时间，所以这里我们选择将读到的日志信息打包发送到管道中，然后再看起一个协程来发送数据，这样实现了读取与发送的一步操作，可以有效降低程序的运行时间,而上面出现的MessageSiz也就是我们设置的管道大小
func run(config *Config) (err error) {
for {
line, ok := <-tailFile.TailObj.Lines
if !ok {
logrus.Error("read from tail failed,err:", err)
time.Sleep(2 * time.Second)
continue
}
msg := &sarama.ProducerMessage{}
msg.Topic = config.Kafakaddress.Topic
msg.Value = sarama.StringEncoder(line.Text)
Kafka.MsgChan <- msg
}
}
发送消息到KafKa
func SendMsg() {
for {
select {
case msg := <-MsgChan:
pid, offset, err := client.SendMessage(msg)
if err != nil {
logrus.Error("send msg to kafka failed,err:%v", err)
return
}
logrus.Info("send msg to kafka success,pid:%d,offset:%d", pid, offset)
}
}
}
完整代码
- main.go
package main
import (
"github.com/Shopify/sarama"
"github.com/Shopify/toxiproxy/Godeps/_workspace/src/github.com/Sirupsen/logrus"
"github.com/go-ini/ini"
"log-agent/Kafka"
"log-agent/tailFile"
"time"
)
type Config struct {
Kafakaddress Kafkaddress `ini:"kafka"`
LogFilePath LogFilePath `ini:"collect"`
}
type Kafkaddress struct {
Addr []string `ini:"address"`
Topic string `ini:"topic"`
MessageSize int64 `ini:"chan_size"`
}
type LogFilePath struct {
Path string `ini:"logfile_path"`
}
func run(config *Config) (err error) {
for {
line, ok := <-tailFile.TailObj.Lines
if !ok {
logrus.Error("read from tail failed,err:", err)
time.Sleep(2 * time.Second)
continue
}
msg := &sarama.ProducerMessage{}
msg.Topic = config.Kafakaddress.Topic
msg.Value = sarama.StringEncoder(line.Text)
Kafka.MsgChan <- msg
}
}
func main() {
//读取配置文件，获取配置信息
filename := "G:\\goproject\\-goroutine-\\log-agent\\conf\\config.ini"
ConfigObj := new(Config)
err := ini.MapTo(ConfigObj, filename)
if err != nil {
logrus.Error("%s Load failed,err:", filename, err)
}
//初始化Kafka
err = Kafka.InitKafka(ConfigObj.Kafakaddress.Addr, ConfigObj.Kafakaddress.MessageSize)
if err != nil {
logrus.Error("InitKafka failed, err:%v", err)
return
}
logrus.Infof("InitKafka success")
//初始化tail
err = tailFile.InitTail(ConfigObj.LogFilePath.Path)
if err != nil {
logrus.Error("InitTail failed, err:%v", err)
return
}
logrus.Infof("InitTail success")
//利用sarama报发送消息到Kafka中
err = run(ConfigObj)
}
- Kafka.go
package Kafka
import (
"github.com/Shopify/sarama"
"github.com/Shopify/toxiproxy/Godeps/_workspace/src/github.com/Sirupsen/logrus"
)
var (
client sarama.SyncProducer
MsgChan chan *sarama.ProducerMessage
)
func InitKafka(address []string, Chan_size int64) (err error) {
//初始化MsgChan
MsgChan = make(chan *sarama.ProducerMessage, Chan_size)
//初始化config
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Partitioner = sarama.NewRandomPartitioner
config.Producer.Return.Successes = true
//连接Kafka
client, err = sarama.NewSyncProducer(address, config)
if err != nil {
logrus.Error("kafka connect error,err:%v", err)
return
}
go SendMsg()
return
}
func SendMsg() {
for {
select {
case msg := <-MsgChan:
pid, offset, err := client.SendMessage(msg)
if err != nil {
logrus.Error("send msg to kafka failed,err:%v", err)
return
}
logrus.Info("send msg to kafka success,pid:%d,offset:%d", pid, offset)
}
}
}
- tailFile.go
package tailFile
import (
"github.com/Shopify/toxiproxy/Godeps/_workspace/src/github.com/Sirupsen/logrus"
"github.com/hpcloud/tail"
)
var TailObj *tail.Tail
func InitTail(filename string) (err error) {
config := tail.Config{
Follow: true,
ReOpen: true,
MustExist: true,
Poll: true,
Location: &tail.SeekInfo{Offset: 0, Whence: 2},
}
TailObj, err = tail.TailFile(filename, config)
if err != nil {
logrus.Error("tail create tailObj for path:%s,err:%v", filename, err)
return
}
return
}
运行结果
在运行前打开ZooKeeper与Kafka，然后对日志文件进行操作，会出现：
出现
2024/04/22 20:26:34 Seeked G:\goproject\-goroutine-\log-agent\log\log1 - &{Offset:0 Whence:2}
INFO[0013] send msg to kafka success,pid:%d,offset:%d0 3
INFO[0013] send msg to kafka success,pid:%d,offset:%d0 4
就代表运行成功了。
结语
今天的有关内容就到此为止啦，有问题的话欢迎在评论区评论，大家可以集思广益，如果你觉得博主的内容对你有帮助，欢迎三连一下和订阅专栏
如果博主文章里面有什么错误页欢迎斧正(毕竟博主页只是个小蒟蒻鸡)，下篇文章我们要进入etcd的有关学习了，好了，大家下篇文章见！
