Golang:將日誌以Json格式輸出到Kafka
在上一篇文章中我實現了一個支援Debug、Info、Error等多個級別的日誌庫,並將日誌寫到了磁碟檔案中,程式碼比較簡單,適合練手。有興趣的可以通過這個連結前往: http://github.com/bosima/ylog/releases/tag/v1.0.2
工程實踐中,我們往往還需要對日誌進行採集,將日誌歸集到一起,然後用於各種處理分析,比如生產環境上的錯誤分析、異常告警等等。在日誌訊息系統領域,Kafka久負盛名,這篇文章就以將日誌傳送到Kafka來實現日誌的採集;同時考慮到日誌分析時對結構化資料的需求,這篇文章還會提供一種輸出Json格式日誌的方法。
這個升級版的日誌庫還要保持向前相容,即還能夠使用普通文字格式,以及寫日誌到磁碟檔案,這兩個特性和要新增的兩個功能分別屬於同類處理,因此我這裡對它們進行抽象,形成兩個介面:格式化介面、寫日誌介面。
格式化介面
所謂格式化,就是日誌的格式處理。這個日誌庫目前要支援兩種格式:普通文字和Json。
為了在不同格式之上提供一個統一的抽象,ylog中定義 logEntry 來代表一條日誌:
type logEntry struct { Ts time.Time `json:"ts"` File string `json:"file"` Line int `json:"line"` Level LogLevel `json:"level"` Msg string `json:"msg"` }
格式化介面的能力就是將日誌從logEntry格式轉化為其它某種資料格式。ylog中對它的定義是:
type LoggerFormatter interface { Format(*logEntry, *[]byte) error }
第1個引數是一個logEntry例項,也就是要被格式化的日誌,第2個引數是日誌格式化之後要寫入的容器。
普通文字格式化器
其實現是這樣的:
type textFormatter struct { } func NewTextFormatter() *textFormatter { return &textFormatter{} } func (f *textFormatter) Format(entry *logEntry, buf *[]byte) error { formatTime(buf, entry.Ts) *buf = append(*buf, ' ') file := toShort(entry.File) *buf = append(*buf, file...) *buf = append(*buf, ':') itoa(buf, entry.Line, -1) *buf = append(*buf, ' ') *buf = append(*buf, levelNames[entry.Level]...) *buf = append(*buf, ' ') *buf = append(*buf, entry.Msg...) return nil }
可以看到它的主要功能就是將logEntry中的各個欄位按照某種順序平鋪開來,中間用空格分隔。
其中的很多資料處理方法參考了Golang標準日誌庫中的資料格式化處理程式碼,有興趣的可以去Github中詳細檢視。
這裡對日期時間格式化為字串做了特別的優化,在標準日誌庫中為了將年、月、日、時、分、秒、毫秒、微秒等格式化指定長度的字串,使用了一個函式:
func itoa(buf *[]byte, i int, wid int) { // Assemble decimal in reverse order. var b [20]byte bp := len(b) - 1 for i >= 10 || wid > 1 { wid-- q := i / 10 b[bp] = byte('0' + i - q*10) bp-- i = q } // i < 10 b[bp] = byte('0' + i) *buf = append(*buf, b[bp:]...) }
其邏輯大概就是將數字中的每一位轉換為字元並存入byte中,注意這裡初始化byte陣列的時候是20位,這是int64最大的數字位數。
其實時間字串中的每個部分位數都是固定的,比如年是4位、月日時分秒都是2位,根本不需要20位,所以這個空間可以節省;還有這裡用了迴圈,這對於CPU的分支預測可能有那麼點影響,所以我這裡分別對不同位數寫了專門的格式化方法,以2位數為例:
func itoa2(buf *[]byte, i int) { q := i / 10 s := byte('0' + i - q*10) f := byte('0' + q) *buf = append(*buf, f, s) }
Json文字格式化器
其實現是這樣的:
type jsonFormatter struct { } func NewJsonFormatter() *jsonFormatter { return &jsonFormatter{} } func (f *jsonFormatter) Format(entry *logEntry, buf *[]byte) (err error) { entry.File = toShortFile(entry.File) jsonBuf, err := json.Marshal(entry) *buf = append(*buf, jsonBuf...) return }
程式碼也很簡單,使用標準庫的json序列化方法將logEntry例項轉化為Json格式的資料。
對於Json格式,後續考慮支援使用者自定義Json欄位,這裡暫時先簡單處理。
寫日誌介面
寫日誌就是將日誌輸出到別的目標,比如ylog要支援的輸出到磁碟檔案、輸出到Kafka等。
前邊格式化介面將格式化後的資料封裝到了 []byte 中,寫日誌介面就是將格式化處理的輸出 []byte 寫到某種輸出目標中。參考Golang中各種Writer的定義,ylog中對它的定義是:
type LoggerWriter interface { Ensure(*logEntry) error Write([]byte) error Sync() error Close() error }
這裡有4個方法:
- Ensure 確保輸出目標已經準備好接收資料,比如開啟要寫入的檔案、建立Kafka連線等等。
- Write 向輸出目標寫資料。
- Sync 要求輸出目標將快取持久化,比如寫資料到磁碟時,作業系統會有快取,通過這個方法要求快取資料寫入磁碟。
- Close 寫日誌結束,關閉輸出目標。
寫日誌到檔案
這裡定義一個名為fileWriter的型別,它需要實現LoggerWriter的介面。
先看型別的定義:
type fileWriter struct { file *os.File mu sync.Mutex lastHour int64 Path string }
包含四個欄位:
- file 要輸出的檔案物件。
- mu 保證檔案操作的協程安全。
- lastHour 按照小時建立檔案的需要。
- Path 日誌檔案的根路徑。
再看其實現的介面:
func (w *fileWriter) Ensure(entry *logEntry) (err error) { if w.file == nil { w.mu.Lock() defer w.mu.Unlock() if w.file == nil { f, err := w.createFile(w.Path, entry.Ts) if err != nil { return err } w.lastHour = w.getTimeHour(entry.Ts) w.file = f } return } currentHour := w.getTimeHour(entry.Ts) if w.lastHour != currentHour { w.mu.Lock() defer w.mu.Unlock() if w.lastHour != currentHour { _ = w.file.Close() f, err := w.createFile(w.Path, entry.Ts) if err != nil { return err } w.lastHour = currentHour w.file = f } } return } func (w *fileWriter) Write(buf []byte) (err error) { buf = append(buf, '\n') _, err = w.file.Write(buf) return } func (w *fileWriter) Sync() error { return w.file.Sync() } func (w *fileWriter) Close() error { return w.file.Close() }
Ensure 中的主要邏輯是建立當前要寫入的檔案物件,如果小時數變了,先把之前的關閉,再建立一個新的檔案。
Write 把資料寫入到檔案物件,這裡加了一個換行符,也就是說對於檔案日誌,其每條日誌最後都會有一個換行符,這樣比較方便閱讀。
Sync 呼叫檔案物件的Sync方法,將日誌從作業系統快取刷到磁碟。
Close 關閉當前檔案物件。
寫日誌到Kafka
這裡定義一個名為kafkaWriter的型別,它也需要實現LoggerWriter的介面。
先看其結構體定義:
type kafkaWriter struct { Topic string Address string writer *kafka.Writer batchSize int }
這裡包含四個欄位:
Topic 寫Kafka時需要一個主題,這裡預設當前Logger中所有日誌使用同一個主題。
Address Kafka的訪問地址。
writer 向Kafka寫資料時使用的Writer,這裡整合的是:github.com/segmentio/kafka-go,支援自動重試和重連。
batchSize Kafka寫日誌的批次大小,批量寫可以提高日誌的寫效率。
再看其實現的介面:
func (w *kafkaWriter) Ensure(curTime time.Time) (err error) { if w.writer == nil { w.writer = &kafka.Writer{ Addr: kafka.TCP(w.Address), Topic: w.Topic, BatchSize: w.batchSize, Async: true, } } return } func (w *kafkaWriter) Write(buf []byte) (err error) { // buf will be reused by ylog when this method return, // with aysnc write, we need copy data to a new slice kbuf := append([]byte(nil), buf...) err = w.writer.WriteMessages(context.Background(), kafka.Message{Value: kbuf}, ) return } func (w *kafkaWriter) Sync() error { return nil } func (w *kafkaWriter) Close() error { return w.writer.Close() }
這裡採用的是非同步傳送到Kafka的方式,WriteMessages方法不會阻塞,因為傳入的buf要被ylog重用,所以這裡copy了一下。非同步還會存在的一個問題就是不會返回錯誤,可能丟失資料,不過對於日誌這種資料,沒有那麼嚴格的要求,也可以接受。
如果採用同步傳送,因為批量傳送比較有效率,這裡可以攢幾條再發,但日誌比較稀疏時,可能短時間很難攢夠,就會出現長時間等不到日誌的情況,所以還要有個超時機制,這有點麻煩,不過我也寫了一個版本,有興趣的可以去看看: http://github.com/bosima/ylog/blob/main/examples/kafka-writer.go
介面的組裝
有了格式化介面和寫日誌介面,下一步就是將它們組裝起來,以實現相應的處理能力。
首先是建立它們,因為我這裡也沒有動態配置的需求,所以就放到建立Logger例項的時候了,這樣比較簡單。
func NewYesLogger(opts ...Option) (logger *YesLogger) { logger = &YesLogger{} ... logger.writer = NewFileWriter("logs") logger.formatter = NewTextFormatter() for _, opt := range opts { opt(logger) } ... return }
可以看到預設的formatter是textFormatter,預設的writer是fileWriter。這個函式傳入的Option其實是個函式,在下邊的opt(logger)中會執行它們,所以使用其它的Formatter或者Writer可以這樣做:
logger := ylog.NewYesLogger( ... ylog.Writer(ylog.NewKafkaWriter(address, topic, writeBatchSize)), ylog.Formatter(ylog.NewJsonFormatter()), )
這裡 ylog.Writer 和 ylog.Formatter 就是符合Option型別的函式,呼叫它們可以設定不同的Formatter和Writer。
然後怎麼使用它們呢?
... l.formatter.Format(entry, &buf) l.writer.Ensure(entry) err := l.writer.Write(buf) ...
當 logEntry 進入訊息處理環節後,首先呼叫formatter的Format方法格式化logEntry;然後呼叫了writer的Ensure方法確保writer已經準備好,最後呼叫writer的Write方法將格式化之後的資料輸出到對應的目標。
為什麼不將Ensure方法放到Write中呢?這是因為目前寫文字日誌的時候需要根據logEntry中的日誌時間建立日誌檔案,這樣就需要給Writer傳遞兩個引數,有點彆扭,所以這裡將它們分開了。
如何提高日誌處理的吞吐量
Kafka的吞吐量是很高的,那麼如果放到ylog自身來說,如何提高它的吞吐量呢?
首先想到的就是Channel,可以使用有緩衝的Channel模擬一個佇列,生產者不停的向Channel傳送資料,如果Writer可以一直在緩衝被填滿之前將資料取走,那麼理論上說生產者就是非阻塞的,相比同步輸出到某個Writer,沒有直接磁碟IO、網路IO,日誌處理的吞吐量必將大幅提升。
定義一個Channel,其容量預設為當前機器邏輯處理器的數量:
logger.pipe = make(chan *logEntry, runtime.NumCPU())
傳送資料的程式碼:
entry := &logEntry{ Level: level, Msg: s, File: file, Line: line, Ts: now, } l.pipe <- entry
接收資料的程式碼:
for { select { case entry := <-l.pipe: // reuse the slice memory buf = buf[:0] l.formatter.Format(entry, &buf) l.writer.Ensure(entry.Ts) err := l.writer.Write(buf) ... } }
實際效果怎麼樣呢?看下Benchmark:
goos: darwin goarch: amd64 pkg: github.com/bosima/ylog cpu: Intel(R) Core(TM) i5-8259U CPU @ 2.30GHz BenchmarkInfo-8 1332333 871.6 ns/op 328 B/op 4 allocs/op
這個結果可以和zerolog、zap等高效能日誌庫一較高下了,當然目前可以做的事情要比它們簡單很多。
如果對Java有所瞭解的同學應該聽說過log4j,在log4j2中引入了一個名為Disruptor的元件,它讓日誌處理飛快了起來,受到很多Java開發者的追捧。Disruptor之所以這麼厲害,是因為它使用了無鎖併發、環形佇列、快取行填充等多種高階技術。
相比之下,Golang的Channel雖然也使用了環形緩衝,但是還是使用了鎖,作為佇列來說效能並不是最優的。
Golang中有沒有類似的東西呢?最近出來的ZenQ可能是一個不錯的選擇,不過看似還不太穩定,過段時間再嘗試下。有興趣的可以去看看: http://github.com/alphadose/ZenQ 。
好了,以上就是本文的主要內容。關於ylog的介紹也告一段落了,後續會在Github上持續更新,增加更多有用的功能,並不斷優化處理效能,歡迎關注: http://github.com/bosima/ylog 。
- Golang:將日誌以Json格式輸出到Kafka
- Golang:手擼一個支援六種級別的日誌庫
- sqlx操作MySQL實戰及其ORM原理
- go-micro使用Consul做服務發現的方法和原理
- go-micro開發RPC服務及其執行原理
- 解決go-micro與其它框架之間的gRPC通訊問題
- go-micro開發gRPC應用程式
- 使用gRPCui測試gRPC服務
- ASP.NET Core中使用漏桶演算法限流
- 限流的非正式用途 – 解決重複提交問題
- 服務限流懲罰是怎麼一回事
- 多租戶系統中如何實現分別限流
- 如何使用陣列實現滑動視窗
- ASP.NET Core中使用令牌桶限流
- ASP.NET Core中如何對不同型別的使用者進行區別限流