Golang监听日志文件并发送到kafka中
作者:zhijie 时间:2024-04-25 13:20:44
前言
日志收集项目的准备中,本文主要讲的是利用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
,监听管道消息,收到消息后,将消息组装,发送到kafkaDestruct
, 关闭管道
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 := <-ks.MsgChan:
//
msg := &sarama.ProducerMessage{
Topic: "weblog",
Value: sarama.StringEncoder(content),
}
partition, offset, err := kafkaClient.SendMessage(msg)
if err != nil {
fmt.Println(err)
}
fmt.Println("分区,偏移量:")
fmt.Println(partition, offset)
fmt.Println("___")
}
}
}
func (ks *KafukaServe) Destruct() {
close(ks.MsgChan)
}
tail.go
主要包括了两个方法:
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 := <-ts.tails.Lines
if !ok {
// todo
fmt.Println("数据接收失败")
return
}
fmt.Println(msg.Text)
MsgChan <- msg.Text
}
}
// 测试案例
func Demo() {
filename := `E:\xx.log`
config := tail.Config{
ReOpen: true,
Follow: true,
Location: &tail.SeekInfo{Offset: 0, Whence: 2},
MustExist: false,
Poll: true,
}
// 打开文件开始读取数据
tails, err := tail.TailFile(filename, config)
if err != nil {
fmt.Println("tails %s failed,err:%v\n", filename, err)
return
}
var (
msg *tail.Line
ok bool
)
fmt.Println("启动")
for {
msg, ok = <-tails.Lines
if !ok {
fmt.Println("tails file close reopen,filename:$s\n", tails.Filename)
}
fmt.Println("msg:", msg.Text)
}
}
来源:https://juejin.cn/post/7086105702057377828
![](/images/zang.png)
![](/images/jiucuo.png)
猜你喜欢
解决python3.x安装numpy成功但import出错的问题
优化 SQL Server 索引的小技巧
opencv python统计及绘制直方图的方法
![](https://img.aspxhome.com/file/2023/0/81450_0s.png)
Vue2 响应式系统之深度响应
![](https://img.aspxhome.com/file/2023/6/133136_0s.png)
使用pandas对两个dataframe进行join的实例
在ASP中使用SQL语句之2:用WHERE子句设置查询条件
教你利用pygame模块制作跳跃小球小游戏
![](https://img.aspxhome.com/file/2023/7/123557_0s.png)
python+numpy实现的基本矩阵操作示例
python开发入门——列表生成式
Python流程控制if条件选择与for循环
![](https://img.aspxhome.com/file/2023/8/131648_0s.png)
MySQL触发器学习总结
python实战串口助手_解决8串口多个发送的问题
MySql批量插入时如何不重复插入数据
![](https://img.aspxhome.com/file/2023/1/122921_0s.png)
MySQL 集群配置
centos系统升级python 2.7.3
Mysql 忘记root密码的完美解决方法
![](https://img.aspxhome.com/file/2023/5/102355_0s.png)
win10下Python3.6安装、配置以及pip安装包教程
![](https://img.aspxhome.com/file/2023/3/101303_0s.png)
Python 调用C++封装的进一步探索交流
![](https://img.aspxhome.com/file/2023/9/120309_0s.jpg)
将不规则的Python多维数组拉平到一维的方法实现
![](https://img.aspxhome.com/file/2023/2/63872_0s.png)