Golang如何将日志以Json格式输出到Kafka

type logEntry struct {
	Ts    time.Time `json:"ts"`
	File  string    `json:"file"`
	Line  int       `json:"line"`
	Level LogLevel  `json:"level"`
	Msg   string    `json:"msg"`
}

type LoggerFORMatter interface {
	Format(*logEntry, *[]byte) error
}

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
}

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:]...)
}

func itoa2(buf *[]byte, i int) {
	q := i / 10
	s := byte('0' + i - q*10)
	f := byte('0' + q)
	*buf = append(*buf, f, s)
}

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
}

type LoggerWriter interface {
	Ensure(*logEntry) error
	Write([]byte) error
	Sync() error
	Close() error
}

type fileWriter struct {
	file     *os.File
	lastHour int64
	Path     string
}

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()
}

type kafkaWriter struct {
	Topic     string
	Address   string
	writer    *kafka.Writer
	batchSize int
}

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()
}

func NewYesLogger(opts ...Option) (logger *YesLogger) {
	logger = &YesLogger{}
	...
	logger.writer = NewFileWriter("logs")
	logger.formatter = NewTextFormatter()

	for _, opt := range opts {
		opt(logger)
	}
	...
	return
}

logger := ylog.NewYesLogger(
		...
		ylog.Writer(ylog.NewKafkaWriter(address, topic, writeBatchSize)),
		ylog.Formatter(ylog.NewJsonFormatter()),
)

...
l.formatter.Format(entry, &buf)
l.writer.Ensure(entry)
err := l.writer.Write(buf)
...

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)
		...
		}
	}

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