pet_home/pet_home_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

系统优化

Browse Source
This commit is contained in:
hahwu 2025-03-19 17:28:46 +08:00
parent fcb838fa9a
commit c3798e7a29
1 changed files with 12 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
src/server/middleware/kafka/kafka.go
View File

@ -2,6 +2,7 @@ package kafkaMiddleware
import ( import (
"context" "context"
"fmt"
"server/conf" "server/conf"
"server/pkg/github.com/name5566/leaf/log" "server/pkg/github.com/name5566/leaf/log"
"time" "time"
@ -12,10 +13,15 @@ import (
var KafkaMod *kafka.Conn var KafkaMod *kafka.Conn
var topic string var topic string
const (
HOST = "kafka-server:9092"
PORT = "9092"
)
func init() { func init() {
topic = conf.Server.GameName topic = conf.Server.GameName
// 连接至Kafka集群的Leader节点 // 连接至Kafka集群的Leader节点
conn, err := kafka.DialContext(context.Background(), "tcp", "kafka-server:9092") conn, err := kafka.DialContext(context.Background(), "tcp", fmt.Sprintf("%s:%s", HOST, PORT))
if err != nil { if err != nil {
log.Debug("kafka.DialContext err: %v", err) log.Debug("kafka.DialContext err: %v", err)
go reconnectKafka() go reconnectKafka()
@ -27,8 +33,8 @@ func init() {
return return
} }
for _, p := range Partitions { for _, p := range Partitions {
if p.Leader.Host == "kafka-server" { if p.Leader.Host == HOST {
conn, err := kafka.DialLeader(context.Background(), "tcp", "kafka-server:9092", topic, p.ID) conn, err := kafka.DialLeader(context.Background(), "tcp", fmt.Sprintf("%s:%s", HOST, PORT), topic, p.ID)
if err != nil { if err != nil {
log.Fatal("kafka.DialLeader err: %v", err) log.Fatal("kafka.DialLeader err: %v", err)
return return
@ -49,7 +55,7 @@ func reconnectKafka() {
log.Debug("Reconnected to Kafka failed") log.Debug("Reconnected to Kafka failed")
return return
} }
conn, err := kafka.DialContext(context.Background(), "tcp", "kafka-server:9092") conn, err := kafka.DialContext(context.Background(), "tcp", fmt.Sprintf("%s:%s", HOST, PORT))
if err != nil { if err != nil {
log.Fatal("kafka.DialContext err: %v", err) log.Fatal("kafka.DialContext err: %v", err)
return return
@ -57,8 +63,8 @@ func reconnectKafka() {
Partitions, err := conn.ReadPartitions(topic) Partitions, err := conn.ReadPartitions(topic)
for _, p := range Partitions { for _, p := range Partitions {
if p.Leader.Host == "kafka-server" { if p.Leader.Host == HOST {
conn, err := kafka.DialLeader(context.Background(), "tcp", "kafka-server:9092", topic, p.ID) conn, err := kafka.DialLeader(context.Background(), "tcp", fmt.Sprintf("%s:%s", HOST, PORT), topic, p.ID)
if err != nil { if err != nil {
log.Fatal("kafka.DialLeader err: %v", err) log.Fatal("kafka.DialLeader err: %v", err)
return return