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

chore: seed migration objects batch 5

Browse Source
This commit is contained in:
dev
2026-10-18 13:37:24 +08:00
parent 53c9c0afd8
commit c8c40b1dd2
239 changed files with 16941 additions and 1544 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/library/queue/kafkamq.go
+112 -65
View File
@@ -7,7 +7,10 @@ package queue
import (
"context"
"crypto/tls"
"crypto/sha256"
"crypto/sha512"
"errors"
"hash"
"strings"
"time"
@@ -15,6 +18,7 @@ import (
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
"github.com/xdg-go/scram"
"hotgo/internal/consts"
"hotgo/utility/simple"
)
@@ -26,17 +30,37 @@ type KafkaMq struct {
}
type KafkaConfig struct {
ClientId string
Brokers []string
GroupID string
Partitions int32
Replication int16
Version string
UserName string
Password string
SASLEnabled bool
SASLMechanism string
TLSEnabled bool
ClientId string
Brokers []string
GroupID string
Partitions int32
Replication int16
Version string
UserName string
Password string
Mechanism string
}
type xdgSCRAMClient struct {
*scram.ClientConversation
HashGeneratorFcn scram.HashGeneratorFcn
}
func (x *xdgSCRAMClient) Begin(userName, password, authzID string) error {
client, err := x.HashGeneratorFcn.NewClient(userName, password, authzID)
if err != nil {
return err
}
x.ClientConversation = client.NewConversation()
return nil
}
func (x *xdgSCRAMClient) Step(challenge string) (string, error) {
return x.ClientConversation.Step(challenge)
}
func (x *xdgSCRAMClient) Done() bool {
return x.ClientConversation.Done()
}
// SendMsg 按字符串类型生产数据
@@ -100,11 +124,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, "kafka consoumer stop : %v", consumerCtx.Err())
Logger().Debugf(ctx, "kafka consumer stop : %v", consumerCtx.Err())
return
}
consumer.ready = make(chan bool)
@@ -112,14 +141,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
@@ -128,7 +161,7 @@ func (r *KafkaMq) ListenReceiveMsgDo(topic string, receiveDo func(mqMsg MqMsg))
// RegisterKafkaMqConsumer 注册消费者
func RegisterKafkaMqConsumer(connOpt KafkaConfig) (client MqConsumer, err error) {
mqIns := &KafkaMq{}
kfkVersion, err := sarama.ParseKafkaVersion(connOpt.Version)
kfkVersion, err := parseKafkaVersion(connOpt.Version)
if err != nil {
return
}
@@ -140,9 +173,7 @@ func RegisterKafkaMqConsumer(connOpt KafkaConfig) (client MqConsumer, err error)
conf := sarama.NewConfig()
conf.Consumer.Return.Errors = true
conf.Version = kfkVersion
if err = configureKafkaSecurity(conf, connOpt); err != nil {
return
}
configureKafkaSASL(conf, connOpt)
// 默认按随机方式消费
conf.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
@@ -161,9 +192,7 @@ func RegisterKafkaMqConsumer(connOpt KafkaConfig) (client MqConsumer, err error)
// RegisterKafkaProducer 注册并启动生产者接口实现
func RegisterKafkaProducer(connOpt KafkaConfig) (client MqProducer, err error) {
mqIns := &KafkaMq{}
if connOpt.ClientId == "" {
connOpt.ClientId = "hotgo-producer"
}
connOpt.ClientId = "HOTGO-Producer"
// 这里如果使用go程需要处理chan同步问题
if err = doRegisterKafkaProducer(connOpt, mqIns); err != nil {
@@ -175,7 +204,7 @@ func RegisterKafkaProducer(connOpt KafkaConfig) (client MqProducer, err error) {
// doRegisterKafkaProducer 注册同步类型实例
func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
kfkVersion, err := sarama.ParseKafkaVersion(connOpt.Version)
kfkVersion, err := parseKafkaVersion(connOpt.Version)
if err != nil {
return
}
@@ -197,9 +226,7 @@ func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
conf.ClientID = connOpt.ClientId
conf.Version = kfkVersion
if err = configureKafkaSecurity(conf, connOpt); err != nil {
return
}
configureKafkaSASL(conf, connOpt)
mqIns.producerIns, err = sarama.NewAsyncProducer(brokers, conf)
if err != nil {
@@ -213,61 +240,81 @@ func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) {
return
}
func configureKafkaSecurity(conf *sarama.Config, connOpt KafkaConfig) error {
if connOpt.TLSEnabled {
conf.Net.TLS.Enable = true
conf.Net.TLS.Config = &tls.Config{MinVersion: tls.VersionTLS12}
func parseKafkaVersion(version string) (sarama.KafkaVersion, error) {
parts := strings.Split(version, ".")
if len(parts) == 4 && parts[0] != "0" && parts[3] == "0" {
version = strings.Join(parts[:3], ".")
}
if !connOpt.SASLEnabled && connOpt.UserName == "" {
return nil
}
if connOpt.UserName == "" || connOpt.Password == "" {
return gerror.New("queue kafka sasl username or password is empty")
return sarama.ParseKafkaVersion(version)
}
func configureKafkaSASL(conf *sarama.Config, connOpt KafkaConfig) {
if connOpt.UserName == "" {
return
}
conf.Net.SASL.Enable = true
conf.Net.SASL.User = connOpt.UserName
conf.Net.SASL.Password = connOpt.Password
mechanism := strings.ToUpper(strings.ReplaceAll(connOpt.SASLMechanism, "_", "-"))
mechanism := strings.ToUpper(strings.ReplaceAll(connOpt.Mechanism, "_", "-"))
if mechanism == "" {
mechanism = sarama.SASLTypeSCRAMSHA512
}
switch mechanism {
case "", "PLAIN":
conf.Net.SASL.Mechanism = sarama.SASLTypePlaintext
case "SCRAM-SHA-256":
case sarama.SASLTypeSCRAMSHA512:
conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
conf.Net.SASL.SCRAMClientGeneratorFunc = scramClientGenerator(sha512.New)
case sarama.SASLTypeSCRAMSHA256:
conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA256
conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
return &xdgSCRAMClient{HashGeneratorFcn: scram.SHA256}
}
case "SCRAM-SHA-512":
conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
conf.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
return &xdgSCRAMClient{HashGeneratorFcn: scram.SHA512}
}
conf.Net.SASL.SCRAMClientGeneratorFunc = scramClientGenerator(sha256.New)
default:
return gerror.New("queue kafka sasl mechanism is not support")
conf.Net.SASL.Mechanism = sarama.SASLMechanism(mechanism)
}
return nil
}
type xdgSCRAMClient struct {
*scram.Client
*scram.ClientConversation
scram.HashGeneratorFcn
}
func (x *xdgSCRAMClient) Begin(userName, password, authzID string) (err error) {
x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID)
// KafkaHealthCheck verifies broker connectivity by creating a short-lived Sarama client.
func KafkaHealthCheck(ctx context.Context) error {
if len(config.Kafka.Address) == 0 {
return errors.New("queue kafka address is empty")
}
kfkVersion, err := parseKafkaVersion(config.Kafka.Version)
if err != nil {
return err
}
x.ClientConversation = x.Client.NewConversation()
if !validateVersion(kfkVersion) {
kfkVersion = sarama.V2_4_0_0
}
conf := sarama.NewConfig()
conf.Version = kfkVersion
conf.ClientID = "hotgo-healthz"
conf.Net.DialTimeout = 2 * time.Second
conf.Net.ReadTimeout = 2 * time.Second
conf.Net.WriteTimeout = 2 * time.Second
configureKafkaSASL(conf, KafkaConfig{
UserName: config.Kafka.Username,
Password: config.Kafka.Password,
Mechanism: config.Kafka.Mechanism,
})
select {
case <-ctx.Done():
return ctx.Err()
default:
}
client, err := sarama.NewClient(config.Kafka.Address, conf)
if err != nil {
return err
}
defer func() { _ = client.Close() }()
if len(client.Brokers()) == 0 {
return errors.New("no kafka brokers available")
}
return nil
}
func (x *xdgSCRAMClient) Step(challenge string) (response string, err error) {
return x.ClientConversation.Step(challenge)
}
func (x *xdgSCRAMClient) Done() bool {
return x.ClientConversation.Done()
func scramClientGenerator(hashFunc func() hash.Hash) func() sarama.SCRAMClient {
return func() sarama.SCRAMClient {
return &xdgSCRAMClient{HashGeneratorFcn: scram.HashGeneratorFcn(hashFunc)}
}
}
// validateVersion 验证版本是否有效