uber go zap 日志框架支持异步日志输出

// A WriteSyncer is an io.Writer that can also flush any buffered data. Note
// that *os.File (and thus, os.Stderr and os.Stdout) implement WriteSyncer.
type WriteSyncer interface {
	io.Writer
	Sync() error
}

// AddSync converts an io.Writer to a WriteSyncer. It attempts to be
// intelligent: if the concrete type of the io.Writer implements WriteSyncer,
// we'll use the existing Sync method. If it doesn't, we'll add a no-op Sync.
func AddSync(w io.Writer) WriteSyncer {
	switch w := w.(type) {
	case WriteSyncer:
		return w
	default:
		return writerWrapper{w}
	}
}
type writerWrapper struct {
	io.Writer
}
func (w writerWrapper) Sync() error {
	return nil
}

// Writer is the interface that wraps the basic Write method.
//
// Write writes len(p) bytes from p to the underlying data stream.
// It returns the number of bytes written from p (0 <= n <= len(p))
// and any error encountered that caused the write to stop early.
// Write must return a non-nil error if it returns n < len(p).
// Write must not modify the slice data, even temporarily.
//
// Implementations must not retain p.
type Writer interface {
	Write(p []byte) (n int, err error)
}

const defaultQueueCap = math.MaxUint16 * 8
var QueueIsFullError = errors.New("queue is full")
var DropWriteMessageError = errors.New("message writing failure and drop it")
type Writer struct {
	name        string
	bufferPool  *extraBufferPool
	writer      io.Writer
	wg          sync.WaitGroup
	lock        sync.RWMutex
	channel     chan *extraBuffer
}
func NewBufferWriter(name string, w io.Writer, queueCap uint32) *Writer {
	if len(name) <= 0 {
		name = "bw_" + utils.GetRandIdString()
	}
	if queueCap <= 0 {
		queueCap = defaultQueueCap
	}
	if w == nil {
		return nil
	}
	wr := Writer{
		name:          name,
		bufferPool:    newExtraBufferPool(defaultBufferSize),
		writer:        w,
		channel:       make(chan *extraBuffer, queueCap),
	}
	wr.wg.Add(1)
	go wr.poller(utils.GetRandIdString())
	return &wr
}
func (w *Writer) Write(p []byte) (int, error) {
	if w.lock.TryRLock() {
		defer w.lock.RUnlock()
		b := w.bufferPool.Get()
		count, err := b.buff.Write(p)
		if err != nil {
			w.bufferPool.Put(b)
			return count, err
		}
		select {
		case w.channel <- b: // channel 内部传递的是 buffer 的指针，速度比传递对象快。
			break
		default:
			w.bufferPool.Put(b)
			return count, QueueIsFullError
		}
		return len(p), nil
	} else {
		return -1, DropWriteMessageError
	}
}
func (w *Writer) Close() {
	w.lock.Lock()
	close(w.channel)
	w.wg.Wait()
	w.lock.Unlock()
}
func (w *Writer) poller(id string) {
	var (
		eb  *extraBuffer
		err error
	)
	defer w.wg.Done()
	for eb = range w.channel {
		_, err = w.writer.Write(eb.buff.Bytes())
		if err != nil {
			log.Printf("writer: %s, id: %s, error: %s, message: %s", w.name, id,
				err.Error(), utils.BytesToString(eb.buff.Bytes()))
		}
		w.bufferPool.Put(eb)
	}
}

import (
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"os"
	"time"
)
func main() {
	wr := NewBufferWriter("lee", os.Stdout, 0)
	defer wr.Close()
	c := zapcore.NewCore(
		zapcore.NewJSONEncoder(zap.NewProductionEncoderConfig()),
		zapcore.AddSync(wr),
		zap.NewAtomicLevelAt(zap.DebugLevel),
	)
	log := zap.New(c)
	log.Info("demo log")
	time.Sleep(3 * time.Second) // 这里要稍微等待下，因为是异步的输出，log.Info() 执行完毕，日志并没有完全输出到 console
}

$ go run asynclog.go
{"level":"info","ts":1674808100.0148869,"msg":"demo log"}

$ wrk -t 10 -c 1000 http://127.0.0.1:8080/xx/
Running 10s test @ http://127.0.0.1:8080/xx/
  10 threads and 1000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    12.03ms   14.23ms 202.46ms   89.23%
    Req/Sec     2.03k     1.36k    9.49k    59.28%
  202813 requests in 10.10s, 100.58MB read
  Socket errors: connect 757, read 73, write 0, timeout 0
Requests/sec:  20074.24
Transfer/sec:      9.96MB

$ wrk -t 10 -c 1000 http://127.0.0.1:8080/xx/
Running 10s test @ http://127.0.0.1:8080/xx/
  10 threads and 1000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     3.75ms    2.43ms  39.94ms   92.68%
    Req/Sec     6.48k     3.86k   14.78k    57.11%
  648554 requests in 10.10s, 321.62MB read
  Socket errors: connect 757, read 79, write 0, timeout 0
Requests/sec:  64197.08
Transfer/sec:     31.84MB

$ wrk -t 10 -c 1000 http://127.0.0.1:8080/xx/
Running 10s test @ http://127.0.0.1:8080/xx/
  10 threads and 1000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     3.69ms  505.13us   9.29ms   77.36%
    Req/Sec     6.60k     4.25k   15.31k    56.45%
  662381 requests in 10.10s, 328.48MB read
  Socket errors: connect 757, read 64, write 0, timeout 0
Requests/sec:  65551.84
Transfer/sec:     32.51MB