我正在创建一个定制的记录器,在这里我们可以登录到std out和std err,但也添加了登录到kafka的可能性(代码示例如下:https://github.com/roppa/kafka-go). 我们有多个主题,所以我们需要多个记录器,但是当我们使用多个时,就会发生一些奇怪的事情。当两个kafka go设置都是异步的时,我不会收到消费者消息,当一个是异步的,另一个是同步的时,我们会得到如下结果:
//consumer topica
{"level":"\u001b[34mINFO\u001b[0m","timeStamp":"2020-12-09T15:31:04.023Z","msg":"topic-a log 1","UID":"abc123","ns":"test-service"}
{"level":"\u001b[34mINFO\u001b[0m","timeStamp":"2020-12-09T15:31:05.078Z","msg":"topic-a log 2","UID":"abc123","ns":"test-service"}
{"level":"\u001b[34mINFO\u001b[0m","timeStamp":"2020-12-09T15:31:06.085Z","msg":"topic-a log 3","UID":"abc123","ns":"test-service"}
//consumer topicb
2020-12-09T15:31:06.085Z INFO topic-a log 3 {"UID": "abc123", "ns": "test-service"}
2","UID":"abc123","ns":"test-service"}
更改同步会产生完全不同的效果。我是新来的。
这是main.go:
package main
import (
"context"
"kafka-log/logger"
)
func main() {
loggerA := logger.Init("test-service", "localhost:9092", "topica", false, false)
loggerB := logger.Init("test-service", "localhost:9092", "topicb", false, true)
ctx := context.Background()
ctx2 := context.WithValue(ctx, logger.UID, "abc123")
loggerA.CInfo(ctx2, "topic-a log 1")
loggerB.CInfo(ctx2, "topic-b log 1")
loggerA.CInfo(ctx2, "topic-a log 2")
loggerB.CInfo(ctx2, "topic-b log 2")
loggerA.CInfo(ctx2, "topic-a log 3")
loggerB.CInfo(ctx2, "topic-b log 3")
}
这是logger/logger.go:
package logger
import (
"context"
"os"
"github.com/segmentio/kafka-go"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
)
type (
key string
// Logger type embeds zap and also contains the current system name (namespace, Ns)
Logger struct {
*zap.Logger
Ns string
}
// KConfig type for creating a new Kafka logger. Takes a Namespace,
// Broker (eg 'localhost:9092'), Topic (eg 'topic-a')
KConfig struct {
Namespace string
Broker string
Topic string
Async bool
}
producerInterface interface {
WriteMessages(ctx context.Context, msgs ...kafka.Message) error
}
// KafkaProducer contains a kafka.Producer and Kafka topic
KafkaProducer struct {
Producer producerInterface
Topic string
}
)
const (
// UID - uniquely request identifier
UID key = "request_id"
)
var customConfig = zapcore.EncoderConfig{
TimeKey: "timeStamp",
LevelKey: "level",
NameKey: "logger",
CallerKey: "caller",
FunctionKey: zapcore.OmitKey,
MessageKey: "msg",
StacktraceKey: "stacktrace",
LineEnding: zapcore.DefaultLineEnding,
EncodeLevel: zapcore.CapitalColorLevelEncoder,
EncodeTime: zapcore.ISO8601TimeEncoder,
EncodeDuration: zapcore.SecondsDurationEncoder,
}
// CInfo this function takes a context as first parameter, extracts specific fields as well as namespace, and calls zap Info
func (l *Logger) CInfo(ctx context.Context, msg string, fields ...zap.Field) {
l.Info(msg, consolidate(ctx, l.Ns, fields...)...)
}
func consolidate(ctx context.Context, namespace string, fields ...zap.Field) []zap.Field {
return append(append(ctxToZapFields(ctx), fields...), zap.String("ns", namespace))
}
// See advanced config example: https://github.com/uber-go/zap/blob/master/example_test.go#L105
var lowPriority = zap.LevelEnablerFunc(func(lvl zapcore.Level) bool {
return lvl < zapcore.ErrorLevel && lvl > zapcore.DebugLevel
})
var debugPriority = zap.LevelEnablerFunc(func(lvl zapcore.Level) bool {
return lvl < zapcore.ErrorLevel
})
var kafkaPriority = zap.LevelEnablerFunc(func(lvl zapcore.Level) bool {
return lvl > zapcore.DebugLevel
})
// Init creates a new instance of a logger. Namespace is the name of the module using the logger. broker and topic are Kafa specific,
// if either of these is not set a default console logger is created.
func Init(namespace, broker, topic string, debug, async bool) *Logger {
var kp *KafkaProducer = nil
if broker != "" && topic != "" {
kp = NewKafkaProducer(&KConfig{
Broker: broker,
Topic: topic,
Async: async,
})
}
logger := getLogger(debug, kp)
// logger.Info("initiated logger", zap.String("ns", namespace), zap.Bool("kafka", kp != nil), zap.Bool("debug", debug))
return &Logger{logger, namespace}
}
func getLogger(debug bool, kp *KafkaProducer) *zap.Logger {
// cores are logger interfaces
var cores []zapcore.Core
// optimise message for console output (human readable)
consoleEncoder := zapcore.NewConsoleEncoder(customConfig)
// Lock wraps a WriteSyncer in a mutex to make it safe for concurrent use.
// See https://godoc.org/go.uber.org/zap/zapcore
cores = append(cores,
zapcore.NewCore(consoleEncoder, zapcore.Lock(os.Stdout), getPriority(debug)),
zapcore.NewCore(consoleEncoder, zapcore.Lock(os.Stderr), zap.ErrorLevel),
)
if kp != nil {
cores = append(cores, zapcore.NewCore(zapcore.NewJSONEncoder(customConfig), zapcore.Lock(zapcore.AddSync(kp)), kafkaPriority))
}
// join inputs, encoders, level-handling functions into cores, then "tee" together
logger := zap.New(zapcore.NewTee(cores...))
defer logger.Sync()
return logger
}
func getPriority(debug bool) zap.LevelEnablerFunc {
if debug {
return debugPriority
}
return lowPriority
}
func ctxToZapFields(ctx context.Context) []zap.Field {
reqID, _ := ctx.Value(UID).(string)
return []zap.Field{
zap.String("UID", reqID),
}
}
// NewKafkaProducer instantiates a kafka.Producer, saves topic, and returns a KafkaProducer
func NewKafkaProducer(c *KConfig) *KafkaProducer {
return &KafkaProducer{
Producer: kafka.NewWriter(kafka.WriterConfig{
Brokers: []string{c.Broker},
Topic: c.Topic,
Balancer: &kafka.Hash{},
Async: c.Async,
RequiredAcks: -1, // -1 = all
}),
Topic: c.Topic,
}
}
// Write takes a message as a byte slice, wraps in a kafka.message and calls kafka Produce
func (kp *KafkaProducer) Write(msg []byte) (int, error) {
return len(msg), kp.Producer.WriteMessages(context.Background(), kafka.Message{
Key: []byte(""),
Value: msg,
})
}
我为消费者使用这些:
docker exec kafka kafka-console-consumer --bootstrap-server localhost:9092 --topic topica
docker exec kafka kafka-console-consumer --bootstrap-server localhost:9092 --topic topicb
这是我的Kafka Docker 作品:
version: '3.8'
services:
zookeeper:
image: confluentinc/cp-zookeeper
networks:
- kafka-net
container_name: zookeeper
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ports:
- 2181:2181
kafka:
image: confluentinc/cp-kafka
networks:
- kafka-net
container_name: kafka
environment:
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
ALLOW_PLAINTEXT_LISTENER: "yes"
KAFKA_LISTENERS-INTERNAL: //kafka:29092,EXTERNAL://localhost:9092
KAFKA_ADVERTISED: INTERNAL://kafka:29092,EXTERNAL://localhost:9092
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://localhost:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
ports:
- 9092:9092
- 29092:29092
depends_on:
- zookeeper
restart: on-failure
networks:
kafka-net:
driver: bridge
1条答案
按热度按时间8ehkhllq1#
我设想您的程序在异步消息有时间发送之前就已经退出了(尽管如果我正确地阅读了您的示例,我很奇怪“topic-a log3”是唯一的日志消息)。与javascript不同,go在退出之前不会等待所有线程/goroutine终止。
还会突出显示kafka go异步配置的docstring:
在解决方案方面:我想你可以打电话解决这个问题
Close
关于作者:https://pkg.go.dev/github.com/segmentio/kafka-go#writer.close
close刷新挂起的写入,并等待所有写入完成后返回。调用close还会阻止新的写操作提交给writer,对writemessages等的进一步调用将因io.errclosedpipe而失败。
你需要把下面的
KafkaProducer.Producer
打电话给我KafkaProducer.Producer.Close
在离开之前。可能有更聪明的方法来组织清理,但我似乎找不到比直接调用writer更简单的方法来清除挂起的消息。