pro/im_manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: stabilize kafka queue startup

Browse Source
This commit is contained in:
richjohnrambo
2026-08-28 01:23:36 +08:00
parent f93b935ea5
commit 296c9a22a8
4 changed files with 22 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/library/queue/consumer.go
+2 -2
View File
@@ -55,7 +55,7 @@ func consumerListen(ctx context.Context, job Consumer) {
)
if err != nil {
Logger().Fatalf(ctx, "InstanceConsumer %s err:%+v", topic, err)
Logger().Errorf(ctx, "InstanceConsumer %s err:%+v", topic, err)
return
}
@@ -70,6 +70,6 @@ func consumerListen(ctx context.Context, job Consumer) {
// 记录消费队列日志
ConsumerLog(ctx, topic, mqMsg, err)
}); listenErr != nil {
Logger().Fatalf(ctx, "消费队列：%s 监听失败, err:%+v", topic, listenErr)
Logger().Errorf(ctx, "消费队列：%s 监听失败, err:%+v", topic, listenErr)
}
}
server/internal/library/queue/disk.go
+1 -1
View File
@@ -125,7 +125,7 @@ func (d *DiskProducerMq) getProducer(topic string) *disk.Queue {
func NewDiskQueue(topic string, config *disk.Config) *disk.Queue {
conf := &disk.Config{
Path: fmt.Sprintf(config.Path + "/" + config.GroupName + "/" + topic),
Path: fmt.Sprintf("%s/%s/%s", config.Path, config.GroupName, topic),
BatchSize: config.BatchSize,
BatchTime: config.BatchTime * time.Second,
SegmentSize: config.SegmentSize,
server/internal/library/queue/kafkamq.go
+13 -5
View File
@@ -9,7 +9,6 @@ import (
"context"
"crypto/sha256"
"crypto/sha512"
"fmt"
"hash"
"strings"
"time"
@@ -124,11 +123,16 @@ func (r *KafkaMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg))
go func(consumerCtx context.Context) {
for {
if err = r.consumerIns.Consume(consumerCtx, []string{topic}, &consumer); err != nil {
Logger().Fatalf(ctx, "kafka Error from consumer, err%+v", err)
Logger().Errorf(ctx, "kafka error from consumer, topic:%s err:%+v", topic, err)
select {
case <-consumerCtx.Done():
return
case <-time.After(3 * time.Second):
}
}
if consumerCtx.Err() != nil {
Logger().Debugf(ctx, fmt.Sprintf("kafka consoumer stop : %v", consumerCtx.Err()))
Logger().Debugf(ctx, "kafka consumer stop : %v", consumerCtx.Err())
return
}
consumer.ready = make(chan bool)
@@ -136,14 +140,18 @@ func (r *KafkaMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg))
}(consumerCtx)
// await till the consumer has been set up
<-consumer.ready
select {
case <-consumer.ready:
case <-time.After(10 * time.Second):
Logger().Warningf(ctx, "kafka consumer setup timeout, topic:%s", topic)
}
Logger().Debug(ctx, "kafka consumer up and running!...")
simple.Event().Register(consts.EventServerClose, func(ctx context.Context, args ...interface{}) {
Logger().Debug(ctx, "kafka consumer close...")
cancel()
if err = r.consumerIns.Close(); err != nil {
Logger().Fatalf(ctx, "kafka Error closing client, err:%+v", err)
Logger().Errorf(ctx, "kafka error closing client, err:%+v", err)
}
})
return
server/manifest/config/config.template
+6 -6
View File
@@ -192,7 +192,7 @@ token:
# 消息队列
queue:
switch: true # 队列开关，可选：true|false，默认为true
driver: "kafka" # 队列驱动，可选：disk|redis|rocketmq|kafka，默认为disk
driver: "$QUEUE_DRIVER" # 队列驱动，可选：disk|redis|rocketmq|kafka，默认为disk
groupName: "hotgo" # mq群组名称
# 磁盘队列
disk:
@@ -212,11 +212,11 @@ queue:
retry: 0 # 重试次数
logLevel: "info" # rocketmq日志输出级别，可选：all|close|debug|info|warn|error|fatal
kafka:
address: ["64.118.137.182:9094"] # kafka地址+端口
version: "2.4.0.0" # kafka专属配置，默认2.4.0.0
username: "kafka-external" # SASL用户名，未开启认证时留空
password: "gr58tNnlZOVGW3AmMbL2" # SASL密码，未开启认证时留空
mechanism: "SCRAM-SHA-512" # SASL机制，可选：SCRAM-SHA-512|SCRAM-SHA-256
address: ["$KAFKA_ADDRESS"] # kafka地址+端口
version: "$KAFKA_VERSION" # kafka专属配置，默认2.4.0.0
username: "$KAFKA_USERNAME" # SASL用户名，未开启认证时留空
password: "$KAFKA_PASSWORD" # SASL密码，未开启认证时留空
mechanism: "$KAFKA_SASL_MECHANISM" # SASL机制，可选：SCRAM-SHA-512|SCRAM-SHA-256
randClient: true # 开启随机生成clientID，可以实现启动多实例同时一起消费相同topic，加速消费能力的特性，默认为true
multiConsumer: true # 是否支持创建多个消费者