Skip to content

refactor(config): 重构 Etcd 配置监听机制 #369

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
2451965602 wants to merge 5 commits into
base: main
Choose a base branch
from
Open

refactor(config): 重构 Etcd 配置监听机制 #369

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ec24c47
♻️ refactor(config): 重构 Etcd 配置监听机制
2451965602 Dec 13, 2025
be5d80e
chore: sync Go modules
2451965602 Dec 13, 2025
7b9bb12
✨ feat(config): 添加 Etcd 配置动态监听和优雅关闭
2451965602 Dec 15, 2025
6eae822
Merge remote-tracking branch 'origin/main'
2451965602 Jan 17, 2026
c4261bd
♻️ refactor(config): 优化环境变量获取与资源管理
2451965602 Jan 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion cmd/academic/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ limitations under the License.
package main

import (
"context"
"os"

"github.com/cloudwego/kitex/pkg/limit"
"github.com/cloudwego/kitex/pkg/rpcinfo"
"github.com/cloudwego/kitex/server"
Expand Down Expand Up @@ -47,6 +50,13 @@ func init() {
}

func main() {
var watcherCancel context.CancelFunc
if os.Getenv(constants.DeployEnv) != "k8s" {
watcherCtx, cancel := context.WithCancel(context.Background())
watcherCancel = cancel
go config.StartEtcdWatcher(watcherCtx, serviceName)
Copy link
Copy Markdown
Member

@ozline ozline Jan 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

我重新看了一下整个链条，我尝试问一个设计的修改：

这里对于每一个模块（需要注意的是这个 pr 由于比较久了，缺少了新增的 captcha 模块），都使用了一模一样的代码，我们能否尝试把这些代码挪到 config.go 配置模块里呢？

这样在每一个模块的 main 函数（或者说更前面的init()函数，init()执行顺序是优先于 main()的）里只需要调用一个 config 的初始化函数就可以了？

因为我看现在的逻辑，会在init()执行时初始化配置，再一直到 main()时开始监听，这个我觉得可以合并一下

你可以看看方案是否有必要

}

r, err := etcd.NewEtcdRegistry([]string{config.Etcd.Addr})
if err != nil {
logger.Fatalf("Academic: etcd registry failed, error: %v", err)
Expand All @@ -73,7 +83,14 @@ func main() {
MaxQPS: constants.MaxQPS,
}),
)
server.RegisterShutdownHook(clientSet.Close)
server.RegisterShutdownHook(func() {
if watcherCancel != nil {
logger.Info("Shutting down etcd config watcher...")
watcherCancel()
}
logger.Info("Closing client resources...")
clientSet.Close()
})

taskQueue.Start()
if err = svr.Run(); err != nil {
Expand Down
19 changes: 18 additions & 1 deletion cmd/api/main.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

17 changes: 16 additions & 1 deletion cmd/classroom/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package main
import (
"context"
"fmt"
"os"
"strconv"
"time"

Expand Down Expand Up @@ -55,6 +56,13 @@ func init() {
}

func main() {
var watcherCancel context.CancelFunc
if os.Getenv(constants.DeployEnv) != "k8s" {
watcherCtx, cancel := context.WithCancel(context.Background())
watcherCancel = cancel
go config.StartEtcdWatcher(watcherCtx, serviceName)
}

r, err := etcd.NewEtcdRegistry([]string{config.Etcd.Addr})
if err != nil {
logger.Fatalf("Classroom: etcd registry failed, error: %v", err)
Expand Down Expand Up @@ -82,7 +90,14 @@ func main() {
MaxQPS: constants.MaxQPS,
}),
)
server.RegisterShutdownHook(clientSet.Close)
server.RegisterShutdownHook(func() {
if watcherCancel != nil {
logger.Info("Shutting down etcd config watcher...")
watcherCancel()
}
logger.Info("Closing client resources...")
clientSet.Close()
})

taskQueue.AddSchedule("update", taskqueue.ScheduleQueueTask{
Execute: func() error {
Expand Down
17 changes: 16 additions & 1 deletion cmd/common/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
"io"
"net/http"
"net/url"
"os"
"strings"
"time"

Expand Down Expand Up @@ -93,6 +94,13 @@ func loadNotice(db *db.Database) {
}

func main() {
var watcherCancel context.CancelFunc
if os.Getenv(constants.DeployEnv) != "k8s" {
watcherCtx, cancel := context.WithCancel(context.Background())
watcherCancel = cancel
go config.StartEtcdWatcher(watcherCtx, serviceName)
}

r, err := etcd.NewEtcdRegistry([]string{config.Etcd.Addr})
if err != nil {
logger.Fatalf("Common: etcd registry failed, error: %v", err)
Expand All @@ -119,7 +127,14 @@ func main() {
MaxQPS: constants.MaxQPS,
}),
)
server.RegisterShutdownHook(clientSet.Close)
server.RegisterShutdownHook(func() {
if watcherCancel != nil {
logger.Info("Shutting down etcd config watcher...")
watcherCancel()
}
logger.Info("Closing client resources...")
clientSet.Close()
})

taskQueue.AddSchedule(constants.NoticeTaskKey, taskqueue.ScheduleQueueTask{
Execute: syncNoticeTask,
Expand Down
17 changes: 16 additions & 1 deletion cmd/course/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ package main

import (
"context"
"os"
"time"

"github.com/cloudwego/kitex/pkg/limit"
Expand Down Expand Up @@ -54,6 +55,13 @@ func init() {
}

func main() {
var watcherCancel context.CancelFunc
if os.Getenv(constants.DeployEnv) != "k8s" {
watcherCtx, cancel := context.WithCancel(context.Background())
watcherCancel = cancel
go config.StartEtcdWatcher(watcherCtx, serviceName)
}

r, err := etcd.NewEtcdRegistry([]string{config.Etcd.Addr})
if err != nil {
// 如果无法解析 etcd 的地址，则无法连接到其他的微服务，说明整个服务无法运行，直接 panic
Expand Down Expand Up @@ -82,7 +90,14 @@ func main() {
MaxQPS: constants.MaxQPS,
}),
)
server.RegisterShutdownHook(clientSet.Close)
server.RegisterShutdownHook(func() {
if watcherCancel != nil {
logger.Info("Shutting down etcd config watcher...")
watcherCancel()
}
logger.Info("Closing client resources...")
clientSet.Close()
})
taskQueue.AddSchedule(constants.LocateDateTaskKey, taskqueue.ScheduleQueueTask{
Execute: func() error {
locateDate, err := jwch.NewStudent().GetLocateDate()
Expand Down
19 changes: 18 additions & 1 deletion cmd/launch_screen/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ limitations under the License.
package main

import (
"context"
"os"

"github.com/cloudwego/kitex/pkg/limit"
"github.com/cloudwego/kitex/pkg/rpcinfo"
"github.com/cloudwego/kitex/server"
Expand Down Expand Up @@ -50,6 +53,13 @@ func init() {
}

func main() {
var watcherCancel context.CancelFunc
if os.Getenv(constants.DeployEnv) != "k8s" {
watcherCtx, cancel := context.WithCancel(context.Background())
watcherCancel = cancel
go config.StartEtcdWatcher(watcherCtx, serviceName)
}

r, err := etcd.NewEtcdRegistry([]string{config.Etcd.Addr})
if err != nil {
logger.Fatalf("launchScreen: etcd registry failed, error: %v", err)
Expand Down Expand Up @@ -80,7 +90,14 @@ func main() {
},
),
)
server.RegisterShutdownHook(clientSet.Close)
server.RegisterShutdownHook(func() {
if watcherCancel != nil {
logger.Info("Shutting down etcd config watcher...")
watcherCancel()
}
logger.Info("Closing client resources...")
clientSet.Close()
})

err = svr.Run()
if err != nil {
Expand Down
20 changes: 20 additions & 0 deletions cmd/oa/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ limitations under the License.
package main

import (
"context"
"os"

"github.com/cloudwego/kitex/pkg/limit"
"github.com/cloudwego/kitex/pkg/rpcinfo"
"github.com/cloudwego/kitex/server"
Expand Down Expand Up @@ -48,6 +51,13 @@ func init() {
}

func main() {
var watcherCancel context.CancelFunc
if os.Getenv(constants.DeployEnv) != "k8s" {
watcherCtx, cancel := context.WithCancel(context.Background())
watcherCancel = cancel
go config.StartEtcdWatcher(watcherCtx, serviceName)
}

r, err := etcd.NewEtcdRegistry([]string{config.Etcd.Addr})
if err != nil {
logger.Fatalf("OA: new etcd registry failed, err: %v", err)
Expand All @@ -73,6 +83,16 @@ func main() {
MaxQPS: constants.MaxQPS,
}),
)

server.RegisterShutdownHook(func() {
if watcherCancel != nil {
logger.Info("Shutting down etcd config watcher...")
watcherCancel()
}
logger.Info("Closing client resources...")
clientSet.Close()
})

if err = svr.Run(); err != nil {
logger.Fatalf("OA: run server failed, err: %v", err)
}
Expand Down
17 changes: 17 additions & 0 deletions cmd/paper/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ limitations under the License.
package main

import (
"context"
"os"

"github.com/cloudwego/kitex/pkg/limit"
"github.com/cloudwego/kitex/pkg/rpcinfo"
"github.com/cloudwego/kitex/server"
Expand Down Expand Up @@ -47,6 +50,12 @@ func init() {
}

func main() {
var watcherCancel context.CancelFunc
if os.Getenv(constants.DeployEnv) != "k8s" {
watcherCtx, cancel := context.WithCancel(context.Background())
watcherCancel = cancel
go config.StartEtcdWatcher(watcherCtx, serviceName)
}
r, err := etcd.NewEtcdRegistry([]string{config.Etcd.Addr})
if err != nil {
logger.Fatalf("Paper: etcd registry failed, error: %v", err)
Expand Down Expand Up @@ -74,6 +83,14 @@ func main() {
MaxQPS: constants.MaxQPS,
}),
)
server.RegisterShutdownHook(func() {
if watcherCancel != nil {
logger.Info("Shutting down etcd config watcher...")
watcherCancel()
}
logger.Info("Closing client resources...")
clientSet.Close()
})

if err = svr.Run(); err != nil {
logger.Fatalf("Paper: server run failed: %v", err)
Expand Down
20 changes: 20 additions & 0 deletions cmd/user/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ limitations under the License.
package main

import (
"context"
"os"

"github.com/cloudwego/kitex/pkg/limit"
"github.com/cloudwego/kitex/pkg/rpcinfo"
"github.com/cloudwego/kitex/server"
Expand Down Expand Up @@ -48,6 +51,13 @@ func init() {
}

func main() {
var watcherCancel context.CancelFunc
if os.Getenv(constants.DeployEnv) != "k8s" {
watcherCtx, cancel := context.WithCancel(context.Background())
watcherCancel = cancel
go config.StartEtcdWatcher(watcherCtx, serviceName)
}

r, err := etcd.NewEtcdRegistry([]string{config.Etcd.Addr})
if err != nil {
logger.Fatalf("User: new etcd registry failed, err: %v", err)
Expand All @@ -73,6 +83,16 @@ func main() {
MaxQPS: constants.MaxQPS,
}),
)

server.RegisterShutdownHook(func() {
if watcherCancel != nil {
logger.Info("Shutting down etcd config watcher...")
watcherCancel()
}
logger.Info("Closing client resources...")
clientSet.Close()
})

if err = svr.Run(); err != nil {
logger.Fatalf("User: run server failed, err: %v", err)
}
Expand Down
18 changes: 18 additions & 0 deletions cmd/version/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package main
import (
"context"
"fmt"
"os"
"time"

"github.com/cloudwego/kitex/pkg/limit"
Expand Down Expand Up @@ -56,6 +57,13 @@ func init() {
}

func main() {
var watcherCancel context.CancelFunc
if os.Getenv(constants.DeployEnv) != "k8s" {
watcherCtx, cancel := context.WithCancel(context.Background())
watcherCancel = cancel
go config.StartEtcdWatcher(watcherCtx, serviceName)
}

r, err := etcd.NewEtcdRegistry([]string{config.Etcd.Addr})
if err != nil {
logger.Fatalf("Version: etcd registry failed, error: %v", err)
Expand All @@ -82,6 +90,16 @@ func main() {
MaxQPS: constants.MaxQPS,
}),
)

server.RegisterShutdownHook(func() {
if watcherCancel != nil {
logger.Info("Shutting down etcd config watcher...")
watcherCancel()
}
logger.Info("Closing client resources...")
clientSet.Close()
})

taskQueue.AddSchedule(constants.VersionVisitedTaskKey, taskqueue.ScheduleQueueTask{
Execute: syncVersionVisitDailyTask,
GetScheduleTime: func() time.Duration {
Expand Down
Loading