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
猜你喜欢
- 我确定有很多关于Unicode和Python的说明,但为了方便自己的理解使用,我还是打算再写一些关于它们的东西。 字节流 vs U
- 1.新建四个层,放入相应图片,模特层的z-index值设为0。2.把第一个层移到模特身上,找出衣服刚好穿上时层的top和left值,记下来,
- 目录前言limit深分页为什么会变慢?通过子查询优化回顾B+ 树结构把条件转移到主键索引树INNER JOIN 延迟关联标签记录法使用bet
- 服务端:#!/usr/bin/envimport SocketServerclass myMonitorHandler(SocketServ
- 看到类似C:\>是在Windows提供的命令行模式,看到>>>是在Python交互式环境下。在命令行模式下,可以执行
- 今天交流会上,分享前端的开发经验,有一条虽然很快带过,但是我倒是印象蛮深刻的,就写点小结来分享一下吧。不知道是标准害了大家还是大家害了标准,
- 打开php.ini,首先找到file_uploads = on ;是否允许通过HTTP上传文件的开关。默认为ON即是开upload_tmp_
- 本文给大家分享的是查看MySQL连接的root密码的方法,下面话不多说来来看正文:1.首先我们进到MySQL的bin目录下➜ cd /usr
- 本文内容皆为作者原创,码字不易,如需转载,请注明出处:https://www.cnblogs.com/temari/p/13048977.h
- 需求:因需要将一json文件中大量的信息填入一固定格式的Excel表格,单纯的复制粘贴肯定也能完成,但是想偷懒一下,于是借助Python解决
- 项目说明开发php项目管理系统,由于是新项目且已经部署在生产环境,导致需要根据实际使用情况,进行及时的功能升级或bug修复。每次升级,进行程
- 工欲善其事必先利其器,Pycharm 是最受欢迎的Python开发工具,它提供的功能非常强大,是构建大型项目的理想工具之一,如果能挖掘出里面
- 简介Github:https://github.com/spf13/cobraStar:26.5KCobra是一个用Go语言实现的命令行工具
- #! /usr/bin/env python ##python2.7-批量下载壁纸 ##壁纸来自桌酷网站,所有权归属其网站 ##本代码仅做为
- 本文实例为大家分享了linux采用binary方式安装mysql的具体步骤,供大家参考,具体内容如下1、下载binary文件在官网上下载 m
- 假如页面上有很多条记录,很多情况下,对这些信息按照字母表降序排序会比传统的升序排序显示效率更高。采用你熟悉的ORDER BY 子句,你可以很
- 代码如下:from PIL import Image #图像处理模块import numpy as npa = np.asarr
- 情景一:不需要原有的数据库数据删除数据库所有的表删除项目的migration模块中的所有 文件,除了__init__.py 文件执行脚本py
- 1 报错类似如下数据库错误: Error querying database. Cause: java.sql.SQLSynta
- 1.0 获取浏览器窗口坐标 python目录可找到Webdriver.py 文件定义了get_window_rect()函数,可获取窗口的坐