Golang：將日誌以Json格式輸出到Kafka
在上一篇文章中我實現了一個支援Debug、Info、Error等多個級別的日誌庫，並將日誌寫到了磁碟文件中，程式碼比較簡單，適合練手。有興趣的可以通過這個鏈接前往：//github.com/bosima/ylog/releases/tag/v1.0.1
工程實踐中，我們往往還需要對日誌進行採集，將日誌歸集到一起，然後用於各種處理分析，比如生產環境上的錯誤分析、異常告警等等。在日誌消息系統領域，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
lastHour int64
Path string
}
包含四個欄位：
- file 要輸出的文件對象。
- lastHour 按照小時創建文件的需要。
- Path 日誌文件的根路徑。
再看其實現的介面：
func (w *fileWriter) Ensure(entry *logEntry) (err error) {
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 nil
}
currentHour := w.getTimeHour(entry.Ts)
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了一下。非同步還會存在的一個問題就是不會返回錯誤，可能丟失數據，不過對於日誌這種數據，沒有那麼嚴格的要求，也可以接受。
如果採用同步發送，因為批量發送比較有效率，這裡可以攢幾條再發，但日誌比較稀疏時，可能短時間很難攢夠，就會出現長時間等不到日誌的情況，所以還要有個超時機制，這有點麻煩，不過我也寫了一個版本，有興趣的可以去看看：//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可能是一個不錯的選擇，不過看似還不太穩定，過段時間再嘗試下。有興趣的可以去看看：//github.com/alphadose/ZenQ 。
好了，以上就是本文的主要內容。關於ylog的介紹也告一段落了，後續會在Github上持續更新，增加更多有用的功能，並不斷優化處理性能，歡迎關註：//github.com/bosima/ylog 。
收穫更多架構知識，請關注微信公眾號 螢火架構。原創內容，轉載請註明出處。
