Add DestroyAll() method to syncx.Pool with comprehensive tests

Co-authored-by: kevwan <1918356+kevwan@users.noreply.github.com>

.codecov.yml

@@ -1,13 +0,0 @@
-coverage:
-  status:
-    patch: true
-    project: false # disabled because project coverage is not stable
-comment:
-  layout: "flags, files"
-  behavior: once
-  require_changes: true
-ignore:
-  - "tools"
-  - "**/mock"
-  - "**/*_mock.go"
-  - "**/*test"

.dockerignore

@@ -1 +1,7 @@
 **/.git
+.dockerignore
+Dockerfile
+goctl
+Makefile
+readme.md
+readme-cn.md

.github/FUNDING.yml

@@ -1,13 +1,3 @@
 # These are supported funding model platforms
 
-github: # Replace with up to 4 GitHub Sponsors-enabled usernames e.g., [user1, user2]
-patreon: # Replace with a single Patreon username
-open_collective: # Replace with a single Open Collective username
-ko_fi: # Replace with a single Ko-fi username
-tidelift: # Replace with a single Tidelift platform-name/package-name e.g., npm/babel
-community_bridge: # Replace with a single Community Bridge project-name e.g., cloud-foundry
-liberapay: # Replace with a single Liberapay username
-issuehunt: # Replace with a single IssueHunt username
-otechie: # Replace with a single Otechie username
-custom: # https://gitee.com/kevwan/static/raw/master/images/sponsor.jpg
-ethereum: # 0x5052b7f6B937B02563996D23feb69b38D06Ca150 | kevwan
+github: [zeromicro]

.github/copilot-instructions.md

@@ -0,0 +1,197 @@
+# GitHub Copilot Instructions for go-zero
+
+This document provides guidelines for GitHub Copilot when assisting with development in the go-zero project.
+
+## Project Overview
+
+go-zero is a web and RPC framework with lots of built-in engineering practices designed to ensure the stability of busy services with resilience design. It has been serving sites with tens of millions of users for years.
+
+### Key Architecture Components
+
+- **REST API framework** (`rest/`) - HTTP service framework with middleware support
+- **RPC framework** (`zrpc/`) - gRPC-based RPC framework with service discovery
+- **Core utilities** (`core/`) - Foundational components including:
+  - Circuit breakers, rate limiters, load shedding
+  - Caching, stores (Redis, MongoDB, SQL)
+  - Concurrency control, metrics, tracing
+  - Configuration management
+- **Code generation tool** (`tools/goctl/`) - CLI tool for generating code from API files
+
+## Coding Standards and Conventions
+
+### Code Style
+
+1. **Follow Go conventions**: Use `gofmt` for formatting, follow effective Go practices
+2. **Package naming**: Use lowercase, single-word package names when possible
+3. **Error handling**: Always handle errors explicitly, use `errorx.BatchError` for multiple errors
+4. **Context propagation**: Always pass `context.Context` as the first parameter for functions that may block
+5. **Configuration structures**: Use struct tags with JSON annotations and default values
+
+Example configuration pattern:
+```go
+type Config struct {
+    Host     string `json:",default=0.0.0.0"`
+    Port     int    `json:",default=8080"`
+    Timeout  int    `json:",default=3000"`
+    Optional string `json:",optional"`
+}
+```
+
+### Interface Design
+
+1. **Small interfaces**: Follow Go's preference for small, focused interfaces
+2. **Context methods**: Provide both context and non-context versions of methods
+3. **Options pattern**: Use functional options for complex configuration
+
+Example:
+```go
+func (c *Client) Get(key string, val any) error {
+    return c.GetCtx(context.Background(), key, val)
+}
+
+func (c *Client) GetCtx(ctx context.Context, key string, val any) error {
+    // implementation
+}
+```
+
+### Testing Patterns
+
+1. **Test file naming**: Use `*_test.go` suffix
+2. **Test function naming**: Use `TestFunctionName` pattern
+3. **Use testify/assert**: Prefer `assert` package for assertions
+4. **Table-driven tests**: Use table-driven tests for multiple scenarios
+5. **Mock interfaces**: Use `go.uber.org/mock` for mocking
+6. **Test helpers**: Use `redistest`, `mongtest` helpers for database testing
+
+Example test pattern:
+```go
+func TestSomething(t *testing.T) {
+    tests := []struct {
+        name     string
+        input    string
+        expected string
+        wantErr  bool
+    }{
+        {"valid case", "input", "output", false},
+        {"error case", "bad", "", true},
+    }
+
+    for _, tt := range tests {
+        t.Run(tt.name, func(t *testing.T) {
+            result, err := SomeFunction(tt.input)
+            if tt.wantErr {
+                assert.Error(t, err)
+                return
+            }
+            assert.NoError(t, err)
+            assert.Equal(t, tt.expected, result)
+        })
+    }
+}
+```
+
+## Framework-Specific Guidelines
+
+### REST API Development
+
+1. **API Definition**: Use `.api` files to define REST APIs
+2. **Handler pattern**: Separate business logic into logic packages
+3. **Middleware**: Use built-in middlewares (tracing, logging, metrics, recovery)
+4. **Response handling**: Use `httpx.WriteJson` for JSON responses
+5. **Error handling**: Use `httpx.Error` for HTTP error responses
+
+### RPC Development
+
+1. **Protocol Buffers**: Use protobuf for service definitions
+2. **Service discovery**: Integrate with etcd for service registration
+3. **Load balancing**: Use built-in load balancing strategies
+4. **Interceptors**: Implement interceptors for cross-cutting concerns
+
+### Database Operations
+
+1. **SQL operations**: Use `sqlx` package for database operations
+2. **Caching**: Implement caching patterns with `cache` package
+3. **Transactions**: Use proper transaction handling
+4. **Connection pooling**: Configure appropriate connection pools
+
+Example cache pattern:
+```go
+err := c.QueryRowCtx(ctx, &dest, key, func(ctx context.Context, conn sqlx.SqlConn) error {
+    return conn.QueryRowCtx(ctx, &dest, query, args...)
+})
+```
+
+### Configuration Management
+
+1. **YAML configuration**: Use YAML for configuration files
+2. **Environment variables**: Support environment variable overrides
+3. **Validation**: Include proper validation for configuration parameters
+4. **Sensible defaults**: Provide reasonable default values
+
+## Error Handling Best Practices
+
+1. **Wrap errors**: Use `fmt.Errorf` with `%w` verb to wrap errors
+2. **Custom errors**: Define custom error types when needed
+3. **Error logging**: Log errors appropriately with context
+4. **Graceful degradation**: Implement fallback mechanisms
+
+## Performance Considerations
+
+1. **Resource pools**: Use connection pools and worker pools
+2. **Circuit breakers**: Implement circuit breaker patterns for external calls
+3. **Rate limiting**: Apply rate limiting to protect services
+4. **Load shedding**: Implement adaptive load shedding
+5. **Metrics**: Add appropriate metrics and monitoring
+
+## Security Guidelines
+
+1. **Input validation**: Validate all input parameters
+2. **SQL injection prevention**: Use parameterized queries
+3. **Authentication**: Implement proper JWT token handling
+4. **HTTPS**: Support TLS/HTTPS configurations
+5. **CORS**: Configure CORS appropriately for web APIs
+
+## Documentation Standards
+
+1. **Package documentation**: Include package-level documentation
+2. **Function documentation**: Document exported functions with examples
+3. **API documentation**: Maintain API documentation in sync
+4. **README updates**: Update README for significant changes
+
+## Common Patterns to Follow
+
+### Service Configuration
+```go
+type ServiceConf struct {
+    Name string
+    Log  logx.LogConf
+    Mode string `json:",default=pro,options=[dev,test,pre,pro]"`
+    // ... other common fields
+}
+```
+
+### Middleware Implementation
+```go
+func SomeMiddleware() rest.Middleware {
+    return func(next http.HandlerFunc) http.HandlerFunc {
+        return func(w http.ResponseWriter, r *http.Request) {
+            // Pre-processing
+            next.ServeHTTP(w, r)
+            // Post-processing
+        }
+    }
+}
+```
+
+### Resource Management
+Always implement proper resource cleanup using defer and context cancellation.
+
+## Build and Test Commands
+
+- Build: `go build ./...`
+- Test: `go test ./...`  
+- Test with race detection: `go test -race ./...`
+- Format: `gofmt -w .`
+- Generate code: `goctl api go -api *.api -dir .`
+
+Remember to run tests and ensure all checks pass before submitting changes. The project emphasizes high quality, performance, and reliability, so these should be primary considerations in all development work.

.github/dependabot.yml

@@ -5,6 +5,14 @@
 
 version: 2
 updates:
+  - package-ecosystem: "docker" # Update image tags in Dockerfile
+    directory: "/"
+    schedule:
+      interval: "weekly"
+  - package-ecosystem: "github-actions" # Update GitHub Actions
+    directory: "/"
+    schedule:
+      interval: "weekly"
   - package-ecosystem: "gomod" # See documentation for possible values
     directory: "/" # Location of package manifests
     schedule:

.github/workflows/codeql-analysis.yml

@@ -35,11 +35,11 @@ jobs:
 
     steps:
     - name: Checkout repository
-      uses: actions/checkout@v4
+      uses: actions/checkout@v5
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v2
+      uses: github/codeql-action/init@v3
       with:
         languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
@@ -50,7 +50,7 @@ jobs:
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
     - name: Autobuild
-      uses: github/codeql-action/autobuild@v2
+      uses: github/codeql-action/autobuild@v3
 
     # ℹ️ Command-line programs to run using the OS shell.
     # 📚 https://git.io/JvXDl
@@ -64,4 +64,4 @@ jobs:
     #   make release
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v2
+      uses: github/codeql-action/analyze@v3

.github/workflows/go.yml

@@ -12,12 +12,12 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Check out code into the Go module directory
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Set up Go 1.x
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v6
         with:
-          go-version: 1.19
+          go-version-file: go.mod
           check-latest: true
           cache: true
         id: go
@@ -40,20 +40,25 @@ jobs:
         run: go test -race -coverprofile=coverage.txt -covermode=atomic ./...
 
       - name: Codecov
-        uses: codecov/codecov-action@v3
+        uses: codecov/codecov-action@v5
+        with:
+          files: ./coverage.txt
+          flags: unittests
+          name: codecov-umbrella
+          fail_ci_if_error: false
 
   test-win:
     name: Windows
     runs-on: windows-latest
     steps:
       - name: Checkout codebase
-        uses: actions/checkout@v4
+        uses: actions/checkout@v5
 
       - name: Set up Go 1.x
-        uses: actions/setup-go@v4
+        uses: actions/setup-go@v6
         with:
-          # use 1.19 to guarantee Go 1.19 compatibility
-          go-version: 1.19
+          # make sure Go version compatible with go-zero
+          go-version-file: go.mod
           check-latest: true
           cache: true
 

.github/workflows/issue-translator.yml

@@ -1,18 +0,0 @@
-name: 'issue-translator'
-on: 
-  issue_comment: 
-    types: [created]
-  issues: 
-    types: [opened]
-
-jobs:
-  build:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: usthe/issues-translate-action@v2.7
-        with:
-          IS_MODIFY_TITLE: true
-          # not require, default false, . Decide whether to modify the issue title
-          # if true, the robot account @Issues-translate-bot must have modification permissions, invite @Issues-translate-bot to your project or use your custom bot.
-          CUSTOM_BOT_NOTE: Bot detected the issue body's language is not English, translate it automatically. 👯👭🏻🧑‍🤝‍🧑👫🧑🏿‍🤝‍🧑🏻👩🏾‍🤝‍👨🏿👬🏿
-          # not require. Customize the translation robot prefix message.

.github/workflows/issues.yml

@@ -7,7 +7,7 @@ jobs:
   close-issues:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/stale@v8
+      - uses: actions/stale@v10
         with:
           days-before-issue-stale: 365
           days-before-issue-close: 90

.github/workflows/release.yaml

@@ -16,13 +16,13 @@ jobs:
           - goarch: "386"
             goos: darwin
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
       - uses: zeromicro/go-zero-release-action@master
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
           goos: ${{ matrix.goos }}
           goarch: ${{ matrix.goarch }}
-          goversion: "https://dl.google.com/go/go1.19.13.linux-amd64.tar.gz"
+          goversion: "https://dl.google.com/go/go1.21.13.linux-amd64.tar.gz"
           project_path: "tools/goctl"
           binary_name: "goctl"
-          extra_files: tools/goctl/readme.md tools/goctl/readme-cn.md
+          extra_files: tools/goctl/readme.md tools/goctl/readme-cn.md

.github/workflows/reviewdog.yml

@@ -5,7 +5,7 @@ jobs:
     name: runner / staticcheck
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v5
       - uses: reviewdog/action-staticcheck@v1
         with:
           github_token: ${{ secrets.github_token }}
@@ -14,6 +14,6 @@ jobs:
           # Report all results.
           filter_mode: nofilter
           # Exit with 1 when it find at least one finding.
-          fail_on_error: true
+          fail_level: any
           # Set staticcheck flags
           staticcheck_flags: -checks=inherit,-SA1019,-SA1029,-SA5008

.github/workflows/version-check.yml

@@ -0,0 +1,42 @@
+name: Release Version Check
+
+on:
+  push:
+    tags:
+      - 'tools/goctl/v*'
+  workflow_dispatch:
+
+jobs:
+  version-check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version: '1.21'
+
+      - name: Extract tag version
+        id: get_version
+        run: |
+          # Extract version from tools/goctl/v* format
+          VERSION="${GITHUB_REF#refs/tags/tools/goctl/v}"
+          echo "VERSION=$VERSION" >> $GITHUB_ENV
+          echo "Extracted version: $VERSION"
+
+      - name: Check version in goctl source code
+        run: |
+          # Change to goctl directory
+          cd tools/goctl
+          
+          # Check version in BuildVersion constant
+          VERSION_IN_CODE=$(grep -r "const BuildVersion =" . | grep -o '".*"' | tr -d '"')
+          echo "Version in code: $VERSION_IN_CODE"
+          echo "Expected version: $VERSION"
+          
+          if [ "$VERSION_IN_CODE" != "$VERSION" ]; then
+            echo "Version mismatch: Version in code ($VERSION_IN_CODE) doesn't match tag version ($VERSION)"
+            exit 1
+          fi
+          echo "✅ Version check passed!"

.gitignore

@@ -17,6 +17,7 @@
 **/logs
 **/adhoc
 **/coverage.txt
+**/WARP.md
 
 # for test purpose
 go.work

SECURITY.md

@@ -0,0 +1,16 @@
+# Security Policy
+
+## Supported Versions
+
+We publish releases monthly.
+
+| Version  | Supported          |
+| -------  | ------------------ |
+| >= 1.4.4 | :white_check_mark: |
+| < 1.4.4  | :x:                |
+
+## Reporting a Vulnerability
+
+https://github.com/zeromicro/go-zero/security/advisories
+
+Accepted vulnerabilities are expected to be fixed within a month.

core/bloom/bloom.go

@@ -2,6 +2,7 @@ package bloom
 
 import (
 	"context"
+	_ "embed"
 	"errors"
 	"strconv"
 
@@ -17,19 +18,13 @@ var (
 	// ErrTooLargeOffset indicates the offset is too large in bitset.
 	ErrTooLargeOffset = errors.New("too large offset")
 
-	setScript = redis.NewScript(`
-for _, offset in ipairs(ARGV) do
-	redis.call("setbit", KEYS[1], offset, 1)
-end
-`)
-	testScript = redis.NewScript(`
-for _, offset in ipairs(ARGV) do
-	if tonumber(redis.call("getbit", KEYS[1], offset)) == 0 then
-		return false
-	end
-end
-return true
-`)
+	//go:embed setscript.lua
+	setLuaScript string
+	setScript    = redis.NewScript(setLuaScript)
+
+	//go:embed testscript.lua
+	testLuaScript string
+	testScript    = redis.NewScript(testLuaScript)
 )
 
 type (
@@ -110,7 +105,7 @@ func newRedisBitSet(store *redis.Redis, key string, bits uint) *redisBitSet {
 }
 
 func (r *redisBitSet) buildOffsetArgs(offsets []uint) ([]string, error) {
-	var args []string
+	args := make([]string, 0, len(offsets))
 
 	for _, offset := range offsets {
 		if offset >= r.bits {
@@ -130,7 +125,7 @@ func (r *redisBitSet) check(ctx context.Context, offsets []uint) (bool, error) {
 	}
 
 	resp, err := r.store.ScriptRunCtx(ctx, testScript, []string{r.key}, args)
-	if err == redis.Nil {
+	if errors.Is(err, redis.Nil) {
 		return false, nil
 	} else if err != nil {
 		return false, err
@@ -162,7 +157,7 @@ func (r *redisBitSet) set(ctx context.Context, offsets []uint) error {
 	}
 
 	_, err = r.store.ScriptRunCtx(ctx, setScript, []string{r.key}, args)
-	if err == redis.Nil {
+	if errors.Is(err, redis.Nil) {
 		return nil
 	}
 

core/bloom/setscript.lua

@@ -0,0 +1,3 @@
+for _, offset in ipairs(ARGV) do
+    redis.call("setbit", KEYS[1], offset, 1)
+end

core/bloom/testscript.lua

@@ -0,0 +1,6 @@
+for _, offset in ipairs(ARGV) do
+    if tonumber(redis.call("getbit", KEYS[1], offset)) == 0 then
+        return false
+    end
+end
+return true

core/breaker/breaker.go

@@ -1,22 +1,19 @@
 package breaker
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
 	"sync"
 	"time"
 
-	"github.com/zeromicro/go-zero/core/mathx"
 	"github.com/zeromicro/go-zero/core/proc"
 	"github.com/zeromicro/go-zero/core/stat"
 	"github.com/zeromicro/go-zero/core/stringx"
 )
 
-const (
-	numHistoryReasons = 5
-	timeFormat        = "15:04:05"
-)
+const numHistoryReasons = 5
 
 // ErrServiceUnavailable is returned when the Breaker state is open.
 var ErrServiceUnavailable = errors.New("circuit breaker is open")
@@ -31,16 +28,21 @@ type (
 		Name() string
 
 		// Allow checks if the request is allowed.
-		// If allowed, a promise will be returned, the caller needs to call promise.Accept()
-		// on success, or call promise.Reject() on failure.
-		// If not allow, ErrServiceUnavailable will be returned.
+		// If allowed, a promise will be returned,
+		// otherwise ErrServiceUnavailable will be returned as the error.
+		// The caller needs to call promise.Accept() on success,
+		// or call promise.Reject() on failure.
 		Allow() (Promise, error)
+		// AllowCtx checks if the request is allowed when ctx isn't done.
+		AllowCtx(ctx context.Context) (Promise, error)
 
 		// Do runs the given request if the Breaker accepts it.
 		// Do returns an error instantly if the Breaker rejects the request.
 		// If a panic occurs in the request, the Breaker handles it as an error
 		// and causes the same panic again.
 		Do(req func() error) error
+		// DoCtx runs the given request if the Breaker accepts it when ctx isn't done.
+		DoCtx(ctx context.Context, req func() error) error
 
 		// DoWithAcceptable runs the given request if the Breaker accepts it.
 		// DoWithAcceptable returns an error instantly if the Breaker rejects the request.
@@ -48,21 +50,31 @@ type (
 		// and causes the same panic again.
 		// acceptable checks if it's a successful call, even if the error is not nil.
 		DoWithAcceptable(req func() error, acceptable Acceptable) error
+		// DoWithAcceptableCtx runs the given request if the Breaker accepts it when ctx isn't done.
+		DoWithAcceptableCtx(ctx context.Context, req func() error, acceptable Acceptable) error
 
 		// DoWithFallback runs the given request if the Breaker accepts it.
 		// DoWithFallback runs the fallback if the Breaker rejects the request.
 		// If a panic occurs in the request, the Breaker handles it as an error
 		// and causes the same panic again.
-		DoWithFallback(req func() error, fallback func(err error) error) error
+		DoWithFallback(req func() error, fallback Fallback) error
+		// DoWithFallbackCtx runs the given request if the Breaker accepts it when ctx isn't done.
+		DoWithFallbackCtx(ctx context.Context, req func() error, fallback Fallback) error
 
 		// DoWithFallbackAcceptable runs the given request if the Breaker accepts it.
 		// DoWithFallbackAcceptable runs the fallback if the Breaker rejects the request.
 		// If a panic occurs in the request, the Breaker handles it as an error
 		// and causes the same panic again.
 		// acceptable checks if it's a successful call, even if the error is not nil.
-		DoWithFallbackAcceptable(req func() error, fallback func(err error) error, acceptable Acceptable) error
+		DoWithFallbackAcceptable(req func() error, fallback Fallback, acceptable Acceptable) error
+		// DoWithFallbackAcceptableCtx runs the given request if the Breaker accepts it when ctx isn't done.
+		DoWithFallbackAcceptableCtx(ctx context.Context, req func() error, fallback Fallback,
+			acceptable Acceptable) error
 	}
 
+	// Fallback is the func to be called if the request is rejected.
+	Fallback func(err error) error
+
 	// Option defines the method to customize a Breaker.
 	Option func(breaker *circuitBreaker)
 
@@ -86,12 +98,12 @@ type (
 
 	internalThrottle interface {
 		allow() (internalPromise, error)
-		doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error
+		doReq(req func() error, fallback Fallback, acceptable Acceptable) error
 	}
 
 	throttle interface {
 		allow() (Promise, error)
-		doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error
+		doReq(req func() error, fallback Fallback, acceptable Acceptable) error
 	}
 )
 
@@ -114,23 +126,71 @@ func (cb *circuitBreaker) Allow() (Promise, error) {
 	return cb.throttle.allow()
 }
 
+func (cb *circuitBreaker) AllowCtx(ctx context.Context) (Promise, error) {
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	default:
+		return cb.Allow()
+	}
+}
+
 func (cb *circuitBreaker) Do(req func() error) error {
 	return cb.throttle.doReq(req, nil, defaultAcceptable)
 }
 
+func (cb *circuitBreaker) DoCtx(ctx context.Context, req func() error) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+		return cb.Do(req)
+	}
+}
+
 func (cb *circuitBreaker) DoWithAcceptable(req func() error, acceptable Acceptable) error {
 	return cb.throttle.doReq(req, nil, acceptable)
 }
 
-func (cb *circuitBreaker) DoWithFallback(req func() error, fallback func(err error) error) error {
+func (cb *circuitBreaker) DoWithAcceptableCtx(ctx context.Context, req func() error,
+	acceptable Acceptable) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+		return cb.DoWithAcceptable(req, acceptable)
+	}
+}
+
+func (cb *circuitBreaker) DoWithFallback(req func() error, fallback Fallback) error {
 	return cb.throttle.doReq(req, fallback, defaultAcceptable)
 }
 
-func (cb *circuitBreaker) DoWithFallbackAcceptable(req func() error, fallback func(err error) error,
+func (cb *circuitBreaker) DoWithFallbackCtx(ctx context.Context, req func() error,
+	fallback Fallback) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+		return cb.DoWithFallback(req, fallback)
+	}
+}
+
+func (cb *circuitBreaker) DoWithFallbackAcceptable(req func() error, fallback Fallback,
 	acceptable Acceptable) error {
 	return cb.throttle.doReq(req, fallback, acceptable)
 }
 
+func (cb *circuitBreaker) DoWithFallbackAcceptableCtx(ctx context.Context, req func() error,
+	fallback Fallback, acceptable Acceptable) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+		return cb.DoWithFallbackAcceptable(req, fallback, acceptable)
+	}
+}
+
 func (cb *circuitBreaker) Name() string {
 	return cb.name
 }
@@ -168,7 +228,7 @@ func (lt loggedThrottle) allow() (Promise, error) {
 	}, lt.logError(err)
 }
 
-func (lt loggedThrottle) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error {
+func (lt loggedThrottle) doReq(req func() error, fallback Fallback, acceptable Acceptable) error {
 	return lt.logError(lt.internalThrottle.doReq(req, fallback, func(err error) bool {
 		accept := acceptable(err)
 		if !accept && err != nil {
@@ -198,14 +258,14 @@ type errorWindow struct {
 
 func (ew *errorWindow) add(reason string) {
 	ew.lock.Lock()
-	ew.reasons[ew.index] = fmt.Sprintf("%s %s", time.Now().Format(timeFormat), reason)
+	ew.reasons[ew.index] = fmt.Sprintf("%s %s", time.Now().Format(time.TimeOnly), reason)
 	ew.index = (ew.index + 1) % numHistoryReasons
-	ew.count = mathx.MinInt(ew.count+1, numHistoryReasons)
+	ew.count = min(ew.count+1, numHistoryReasons)
 	ew.lock.Unlock()
 }
 
 func (ew *errorWindow) String() string {
-	var reasons []string
+	reasons := make([]string, 0, ew.count)
 
 	ew.lock.Lock()
 	// reverse order

core/breaker/breaker_test.go

@@ -1,11 +1,13 @@
 package breaker
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/stat"
@@ -16,10 +18,274 @@ func init() {
 }
 
 func TestCircuitBreaker_Allow(t *testing.T) {
-	b := NewBreaker()
-	assert.True(t, len(b.Name()) > 0)
-	_, err := b.Allow()
-	assert.Nil(t, err)
+	t.Run("allow", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		_, err := b.Allow()
+		assert.Nil(t, err)
+	})
+
+	t.Run("allow with ctx", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		_, err := b.AllowCtx(context.Background())
+		assert.Nil(t, err)
+	})
+
+	t.Run("allow with ctx timeout", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Microsecond)
+		defer cancel()
+		time.Sleep(time.Millisecond)
+		_, err := b.AllowCtx(ctx)
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+	})
+
+	t.Run("allow with ctx cancel", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		for i := 0; i < 100; i++ {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			cancel()
+			_, err := b.AllowCtx(ctx)
+			assert.ErrorIs(t, err, context.Canceled)
+		}
+		_, err := b.AllowCtx(context.Background())
+		assert.NoError(t, err)
+	})
+}
+
+func TestCircuitBreaker_Do(t *testing.T) {
+	t.Run("do", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		err := b.Do(func() error {
+			return nil
+		})
+		assert.Nil(t, err)
+	})
+
+	t.Run("do with ctx", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		err := b.DoCtx(context.Background(), func() error {
+			return nil
+		})
+		assert.Nil(t, err)
+	})
+
+	t.Run("do with ctx timeout", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Microsecond)
+		defer cancel()
+		time.Sleep(time.Millisecond)
+		err := b.DoCtx(ctx, func() error {
+			return nil
+		})
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+	})
+
+	t.Run("do with ctx cancel", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		for i := 0; i < 100; i++ {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			cancel()
+			err := b.DoCtx(ctx, func() error {
+				return nil
+			})
+			assert.ErrorIs(t, err, context.Canceled)
+		}
+		assert.NoError(t, b.DoCtx(context.Background(), func() error {
+			return nil
+		}))
+	})
+}
+
+func TestCircuitBreaker_DoWithAcceptable(t *testing.T) {
+	t.Run("doWithAcceptable", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		err := b.DoWithAcceptable(func() error {
+			return nil
+		}, func(err error) bool {
+			return true
+		})
+		assert.Nil(t, err)
+	})
+
+	t.Run("doWithAcceptable with ctx", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		err := b.DoWithAcceptableCtx(context.Background(), func() error {
+			return nil
+		}, func(err error) bool {
+			return true
+		})
+		assert.Nil(t, err)
+	})
+
+	t.Run("doWithAcceptable with ctx timeout", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Microsecond)
+		defer cancel()
+		time.Sleep(time.Millisecond)
+		err := b.DoWithAcceptableCtx(ctx, func() error {
+			return nil
+		}, func(err error) bool {
+			return true
+		})
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+	})
+
+	t.Run("doWithAcceptable with ctx cancel", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		for i := 0; i < 100; i++ {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			cancel()
+			err := b.DoWithAcceptableCtx(ctx, func() error {
+				return nil
+			}, func(err error) bool {
+				return true
+			})
+			assert.ErrorIs(t, err, context.Canceled)
+		}
+		assert.NoError(t, b.DoWithAcceptableCtx(context.Background(), func() error {
+			return nil
+		}, func(err error) bool {
+			return true
+		}))
+	})
+}
+
+func TestCircuitBreaker_DoWithFallback(t *testing.T) {
+	t.Run("doWithFallback", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		err := b.DoWithFallback(func() error {
+			return nil
+		}, func(err error) error {
+			return err
+		})
+		assert.Nil(t, err)
+	})
+
+	t.Run("doWithFallback with ctx", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		err := b.DoWithFallbackCtx(context.Background(), func() error {
+			return nil
+		}, func(err error) error {
+			return err
+		})
+		assert.Nil(t, err)
+	})
+
+	t.Run("doWithFallback with ctx timeout", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Microsecond)
+		defer cancel()
+		time.Sleep(time.Millisecond)
+		err := b.DoWithFallbackCtx(ctx, func() error {
+			return nil
+		}, func(err error) error {
+			return err
+		})
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+	})
+
+	t.Run("doWithFallback with ctx cancel", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		for i := 0; i < 100; i++ {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			cancel()
+			err := b.DoWithFallbackCtx(ctx, func() error {
+				return nil
+			}, func(err error) error {
+				return err
+			})
+			assert.ErrorIs(t, err, context.Canceled)
+		}
+		assert.NoError(t, b.DoWithFallbackCtx(context.Background(), func() error {
+			return nil
+		}, func(err error) error {
+			return err
+		}))
+	})
+}
+
+func TestCircuitBreaker_DoWithFallbackAcceptable(t *testing.T) {
+	t.Run("doWithFallbackAcceptable", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		err := b.DoWithFallbackAcceptable(func() error {
+			return nil
+		}, func(err error) error {
+			return err
+		}, func(err error) bool {
+			return true
+		})
+		assert.Nil(t, err)
+	})
+
+	t.Run("doWithFallbackAcceptable with ctx", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		err := b.DoWithFallbackAcceptableCtx(context.Background(), func() error {
+			return nil
+		}, func(err error) error {
+			return err
+		}, func(err error) bool {
+			return true
+		})
+		assert.Nil(t, err)
+	})
+
+	t.Run("doWithFallbackAcceptable with ctx timeout", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Microsecond)
+		defer cancel()
+		time.Sleep(time.Millisecond)
+		err := b.DoWithFallbackAcceptableCtx(ctx, func() error {
+			return nil
+		}, func(err error) error {
+			return err
+		}, func(err error) bool {
+			return true
+		})
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+	})
+
+	t.Run("doWithFallbackAcceptable with ctx cancel", func(t *testing.T) {
+		b := NewBreaker()
+		assert.True(t, len(b.Name()) > 0)
+		for i := 0; i < 100; i++ {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			cancel()
+			err := b.DoWithFallbackAcceptableCtx(ctx, func() error {
+				return nil
+			}, func(err error) error {
+				return err
+			}, func(err error) bool {
+				return true
+			})
+			assert.ErrorIs(t, err, context.Canceled)
+		}
+		assert.NoError(t, b.DoWithFallbackAcceptableCtx(context.Background(), func() error {
+			return nil
+		}, func(err error) error {
+			return err
+		}, func(err error) bool {
+			return true
+		}))
+	})
 }
 
 func TestLogReason(t *testing.T) {

core/breaker/breakers.go

@@ -1,6 +1,9 @@
 package breaker
 
-import "sync"
+import (
+	"context"
+	"sync"
+)
 
 var (
 	lock     sync.RWMutex
@@ -14,6 +17,13 @@ func Do(name string, req func() error) error {
 	})
 }
 
+// DoCtx calls Breaker.DoCtx on the Breaker with given name.
+func DoCtx(ctx context.Context, name string, req func() error) error {
+	return do(name, func(b Breaker) error {
+		return b.DoCtx(ctx, req)
+	})
+}
+
 // DoWithAcceptable calls Breaker.DoWithAcceptable on the Breaker with given name.
 func DoWithAcceptable(name string, req func() error, acceptable Acceptable) error {
 	return do(name, func(b Breaker) error {
@@ -21,21 +31,44 @@ func DoWithAcceptable(name string, req func() error, acceptable Acceptable) erro
 	})
 }
 
+// DoWithAcceptableCtx calls Breaker.DoWithAcceptableCtx on the Breaker with given name.
+func DoWithAcceptableCtx(ctx context.Context, name string, req func() error,
+	acceptable Acceptable) error {
+	return do(name, func(b Breaker) error {
+		return b.DoWithAcceptableCtx(ctx, req, acceptable)
+	})
+}
+
 // DoWithFallback calls Breaker.DoWithFallback on the Breaker with given name.
-func DoWithFallback(name string, req func() error, fallback func(err error) error) error {
+func DoWithFallback(name string, req func() error, fallback Fallback) error {
 	return do(name, func(b Breaker) error {
 		return b.DoWithFallback(req, fallback)
 	})
 }
 
+// DoWithFallbackCtx calls Breaker.DoWithFallbackCtx on the Breaker with given name.
+func DoWithFallbackCtx(ctx context.Context, name string, req func() error, fallback Fallback) error {
+	return do(name, func(b Breaker) error {
+		return b.DoWithFallbackCtx(ctx, req, fallback)
+	})
+}
+
 // DoWithFallbackAcceptable calls Breaker.DoWithFallbackAcceptable on the Breaker with given name.
-func DoWithFallbackAcceptable(name string, req func() error, fallback func(err error) error,
+func DoWithFallbackAcceptable(name string, req func() error, fallback Fallback,
 	acceptable Acceptable) error {
 	return do(name, func(b Breaker) error {
 		return b.DoWithFallbackAcceptable(req, fallback, acceptable)
 	})
 }
 
+// DoWithFallbackAcceptableCtx calls Breaker.DoWithFallbackAcceptableCtx on the Breaker with given name.
+func DoWithFallbackAcceptableCtx(ctx context.Context, name string, req func() error,
+	fallback Fallback, acceptable Acceptable) error {
+	return do(name, func(b Breaker) error {
+		return b.DoWithFallbackAcceptableCtx(ctx, req, fallback, acceptable)
+	})
+}
+
 // GetBreaker returns the Breaker with the given name.
 func GetBreaker(name string) Breaker {
 	lock.RLock()
@@ -59,7 +92,7 @@ func GetBreaker(name string) Breaker {
 // NoBreakerFor disables the circuit breaker for the given name.
 func NoBreakerFor(name string) {
 	lock.Lock()
-	breakers[name] = newNopBreaker()
+	breakers[name] = NopBreaker()
 	lock.Unlock()
 }
 

core/breaker/breakers_test.go

@@ -1,6 +1,7 @@
 package breaker
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"testing"
@@ -22,6 +23,9 @@ func TestBreakersDo(t *testing.T) {
 	assert.Equal(t, errDummy, Do("any", func() error {
 		return errDummy
 	}))
+	assert.Equal(t, errDummy, DoCtx(context.Background(), "any", func() error {
+		return errDummy
+	}))
 }
 
 func TestBreakersDoWithAcceptable(t *testing.T) {
@@ -38,6 +42,13 @@ func TestBreakersDoWithAcceptable(t *testing.T) {
 			return nil
 		}) == nil
 	})
+	verify(t, func() bool {
+		return DoWithAcceptableCtx(context.Background(), "anyone", func() error {
+			return nil
+		}, func(err error) bool {
+			return true
+		}) == nil
+	})
 
 	for i := 0; i < 10000; i++ {
 		err := DoWithAcceptable("another", func() error {
@@ -76,6 +87,12 @@ func TestBreakersFallback(t *testing.T) {
 			return nil
 		})
 		assert.True(t, err == nil || errors.Is(err, errDummy))
+		err = DoWithFallbackCtx(context.Background(), "fallback", func() error {
+			return errDummy
+		}, func(err error) error {
+			return nil
+		})
+		assert.True(t, err == nil || errors.Is(err, errDummy))
 	}
 	verify(t, func() bool {
 		return errors.Is(Do("fallback", func() error {
@@ -86,7 +103,7 @@ func TestBreakersFallback(t *testing.T) {
 
 func TestBreakersAcceptableFallback(t *testing.T) {
 	errDummy := errors.New("any")
-	for i := 0; i < 10000; i++ {
+	for i := 0; i < 5000; i++ {
 		err := DoWithFallbackAcceptable("acceptablefallback", func() error {
 			return errDummy
 		}, func(err error) error {
@@ -95,6 +112,14 @@ func TestBreakersAcceptableFallback(t *testing.T) {
 			return err == nil
 		})
 		assert.True(t, err == nil || errors.Is(err, errDummy))
+		err = DoWithFallbackAcceptableCtx(context.Background(), "acceptablefallback", func() error {
+			return errDummy
+		}, func(err error) error {
+			return nil
+		}, func(err error) bool {
+			return err == nil
+		})
+		assert.True(t, err == nil || errors.Is(err, errDummy))
 	}
 	verify(t, func() bool {
 		return errors.Is(Do("acceptablefallback", func() error {
@@ -110,5 +135,5 @@ func verify(t *testing.T, fn func() bool) {
 			count++
 		}
 	}
-	assert.True(t, count >= 80, fmt.Sprintf("should be greater than 80, actual %d", count))
+	assert.True(t, count >= 75, fmt.Sprintf("should be greater than 75, actual %d", count))
 }

core/breaker/bucket.go

@@ -0,0 +1,48 @@
+package breaker
+
+const (
+	success = iota
+	fail
+	drop
+)
+
+// bucket defines the bucket that holds sum and num of additions.
+type bucket struct {
+	Sum     int64
+	Success int64
+	Failure int64
+	Drop    int64
+}
+
+func (b *bucket) Add(v int64) {
+	switch v {
+	case fail:
+		b.fail()
+	case drop:
+		b.drop()
+	default:
+		b.succeed()
+	}
+}
+
+func (b *bucket) Reset() {
+	b.Sum = 0
+	b.Success = 0
+	b.Failure = 0
+	b.Drop = 0
+}
+
+func (b *bucket) drop() {
+	b.Sum++
+	b.Drop++
+}
+
+func (b *bucket) fail() {
+	b.Sum++
+	b.Failure++
+}
+
+func (b *bucket) succeed() {
+	b.Sum++
+	b.Success++
+}

core/breaker/bucket_test.go

@@ -0,0 +1,43 @@
+package breaker
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBucketAdd(t *testing.T) {
+	b := &bucket{}
+
+	// Test succeed
+	b.Add(0) // Using 0 for success
+	assert.Equal(t, int64(1), b.Sum, "Sum should be incremented")
+	assert.Equal(t, int64(1), b.Success, "Success should be incremented")
+	assert.Equal(t, int64(0), b.Failure, "Failure should not be incremented")
+	assert.Equal(t, int64(0), b.Drop, "Drop should not be incremented")
+
+	// Test failure
+	b.Add(fail)
+	assert.Equal(t, int64(2), b.Sum, "Sum should be incremented")
+	assert.Equal(t, int64(1), b.Failure, "Failure should be incremented")
+	assert.Equal(t, int64(0), b.Drop, "Drop should not be incremented")
+
+	// Test drop
+	b.Add(drop)
+	assert.Equal(t, int64(3), b.Sum, "Sum should be incremented")
+	assert.Equal(t, int64(1), b.Drop, "Drop should be incremented")
+}
+
+func TestBucketReset(t *testing.T) {
+	b := &bucket{
+		Sum:     3,
+		Success: 1,
+		Failure: 1,
+		Drop:    1,
+	}
+	b.Reset()
+	assert.Equal(t, int64(0), b.Sum, "Sum should be reset to 0")
+	assert.Equal(t, int64(0), b.Success, "Success should be reset to 0")
+	assert.Equal(t, int64(0), b.Failure, "Failure should be reset to 0")
+	assert.Equal(t, int64(0), b.Drop, "Drop should be reset to 0")
+}

core/breaker/googlebreaker.go

@@ -1,57 +1,87 @@
 package breaker
 
 import (
-	"math"
 	"time"
 
 	"github.com/zeromicro/go-zero/core/collection"
 	"github.com/zeromicro/go-zero/core/mathx"
+	"github.com/zeromicro/go-zero/core/syncx"
+	"github.com/zeromicro/go-zero/core/timex"
 )
 
 const (
 	// 250ms for bucket duration
-	window     = time.Second * 10
-	buckets    = 40
-	k          = 1.5
-	protection = 5
+	window            = time.Second * 10
+	buckets           = 40
+	forcePassDuration = time.Second
+	k                 = 1.5
+	minK              = 1.1
+	protection        = 5
 )
 
 // googleBreaker is a netflixBreaker pattern from google.
 // see Client-Side Throttling section in https://landing.google.com/sre/sre-book/chapters/handling-overload/
-type googleBreaker struct {
-	k     float64
-	stat  *collection.RollingWindow
-	proba *mathx.Proba
-}
+type (
+	googleBreaker struct {
+		k        float64
+		stat     *collection.RollingWindow[int64, *bucket]
+		proba    *mathx.Proba
+		lastPass *syncx.AtomicDuration
+	}
+
+	windowResult struct {
+		accepts        int64
+		total          int64
+		failingBuckets int64
+		workingBuckets int64
+	}
+)
 
 func newGoogleBreaker() *googleBreaker {
 	bucketDuration := time.Duration(int64(window) / int64(buckets))
-	st := collection.NewRollingWindow(buckets, bucketDuration)
+	st := collection.NewRollingWindow[int64, *bucket](func() *bucket {
+		return new(bucket)
+	}, buckets, bucketDuration)
 	return &googleBreaker{
-		stat:  st,
-		k:     k,
-		proba: mathx.NewProba(),
+		stat:     st,
+		k:        k,
+		proba:    mathx.NewProba(),
+		lastPass: syncx.NewAtomicDuration(),
 	}
 }
 
 func (b *googleBreaker) accept() error {
-	accepts, total := b.history()
-	weightedAccepts := b.k * float64(accepts)
+	var w float64
+	history := b.history()
+	w = b.k - (b.k-minK)*float64(history.failingBuckets)/buckets
+	weightedAccepts := mathx.AtLeast(w, minK) * float64(history.accepts)
 	// https://landing.google.com/sre/sre-book/chapters/handling-overload/#eq2101
-	dropRatio := math.Max(0, (float64(total-protection)-weightedAccepts)/float64(total+1))
+	// for better performance, no need to care about the negative ratio
+	dropRatio := (float64(history.total-protection) - weightedAccepts) / float64(history.total+1)
 	if dropRatio <= 0 {
 		return nil
 	}
 
+	lastPass := b.lastPass.Load()
+	if lastPass > 0 && timex.Since(lastPass) > forcePassDuration {
+		b.lastPass.Set(timex.Now())
+		return nil
+	}
+
+	dropRatio *= float64(buckets-history.workingBuckets) / buckets
+
 	if b.proba.TrueOnProba(dropRatio) {
 		return ErrServiceUnavailable
 	}
 
+	b.lastPass.Set(timex.Now())
+
 	return nil
 }
 
 func (b *googleBreaker) allow() (internalPromise, error) {
 	if err := b.accept(); err != nil {
+		b.markDrop()
 		return nil, err
 	}
 
@@ -60,8 +90,9 @@ func (b *googleBreaker) allow() (internalPromise, error) {
 	}, nil
 }
 
-func (b *googleBreaker) doReq(req func() error, fallback func(err error) error, acceptable Acceptable) error {
+func (b *googleBreaker) doReq(req func() error, fallback Fallback, acceptable Acceptable) error {
 	if err := b.accept(); err != nil {
+		b.markDrop()
 		if fallback != nil {
 			return fallback(err)
 		}
@@ -69,38 +100,55 @@ func (b *googleBreaker) doReq(req func() error, fallback func(err error) error,
 		return err
 	}
 
+	var succ bool
 	defer func() {
-		if e := recover(); e != nil {
+		// if req() panic, success is false, mark as failure
+		if succ {
+			b.markSuccess()
+		} else {
 			b.markFailure()
-			panic(e)
 		}
 	}()
 
 	err := req()
 	if acceptable(err) {
-		b.markSuccess()
-	} else {
-		b.markFailure()
+		succ = true
 	}
 
 	return err
 }
 
-func (b *googleBreaker) markSuccess() {
-	b.stat.Add(1)
+func (b *googleBreaker) markDrop() {
+	b.stat.Add(drop)
 }
 
 func (b *googleBreaker) markFailure() {
-	b.stat.Add(0)
+	b.stat.Add(fail)
 }
 
-func (b *googleBreaker) history() (accepts, total int64) {
-	b.stat.Reduce(func(b *collection.Bucket) {
-		accepts += int64(b.Sum)
-		total += b.Count
+func (b *googleBreaker) markSuccess() {
+	b.stat.Add(success)
+}
+
+func (b *googleBreaker) history() windowResult {
+	var result windowResult
+
+	b.stat.Reduce(func(b *bucket) {
+		result.accepts += b.Success
+		result.total += b.Sum
+		if b.Failure > 0 {
+			result.workingBuckets = 0
+		} else if b.Success > 0 {
+			result.workingBuckets++
+		}
+		if b.Success > 0 {
+			result.failingBuckets = 0
+		} else if b.Failure > 0 {
+			result.failingBuckets++
+		}
 	})
 
-	return
+	return result
 }
 
 type googlePromise struct {

core/breaker/googlebreaker_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/zeromicro/go-zero/core/collection"
 	"github.com/zeromicro/go-zero/core/mathx"
 	"github.com/zeromicro/go-zero/core/stat"
+	"github.com/zeromicro/go-zero/core/syncx"
 )
 
 const (
@@ -22,11 +23,14 @@ func init() {
 }
 
 func getGoogleBreaker() *googleBreaker {
-	st := collection.NewRollingWindow(testBuckets, testInterval)
+	st := collection.NewRollingWindow[int64, *bucket](func() *bucket {
+		return new(bucket)
+	}, testBuckets, testInterval)
 	return &googleBreaker{
-		stat:  st,
-		k:     5,
-		proba: mathx.NewProba(),
+		stat:     st,
+		k:        5,
+		proba:    mathx.NewProba(),
+		lastPass: syncx.NewAtomicDuration(),
 	}
 }
 
@@ -63,6 +67,33 @@ func TestGoogleBreakerOpen(t *testing.T) {
 	})
 }
 
+func TestGoogleBreakerRecover(t *testing.T) {
+	st := collection.NewRollingWindow[int64, *bucket](func() *bucket {
+		return new(bucket)
+	}, testBuckets*2, testInterval)
+	b := &googleBreaker{
+		stat:     st,
+		k:        k,
+		proba:    mathx.NewProba(),
+		lastPass: syncx.NewAtomicDuration(),
+	}
+	for i := 0; i < testBuckets; i++ {
+		for j := 0; j < 100; j++ {
+			b.stat.Add(1)
+		}
+		time.Sleep(testInterval)
+	}
+	for i := 0; i < testBuckets; i++ {
+		for j := 0; j < 100; j++ {
+			b.stat.Add(0)
+		}
+		time.Sleep(testInterval)
+	}
+	verify(t, func() bool {
+		return b.accept() == nil
+	})
+}
+
 func TestGoogleBreakerFallback(t *testing.T) {
 	b := getGoogleBreaker()
 	markSuccess(b, 1)
@@ -89,6 +120,43 @@ func TestGoogleBreakerReject(t *testing.T) {
 	}, nil, defaultAcceptable))
 }
 
+func TestGoogleBreakerMoreFallingBuckets(t *testing.T) {
+	t.Parallel()
+
+	t.Run("more falling buckets", func(t *testing.T) {
+		b := getGoogleBreaker()
+
+		func() {
+			stopChan := time.After(testInterval * 6)
+			for {
+				time.Sleep(time.Millisecond)
+				select {
+				case <-stopChan:
+					return
+				default:
+					assert.Error(t, b.doReq(func() error {
+						return errors.New("foo")
+					}, func(err error) error {
+						return err
+					}, func(err error) bool {
+						return err == nil
+					}))
+				}
+			}
+		}()
+
+		var count int
+		for i := 0; i < 100; i++ {
+			if errors.Is(b.doReq(func() error {
+				return ErrServiceUnavailable
+			}, nil, defaultAcceptable), ErrServiceUnavailable) {
+				count++
+			}
+		}
+		assert.True(t, count > 90)
+	})
+}
+
 func TestGoogleBreakerAcceptable(t *testing.T) {
 	b := getGoogleBreaker()
 	errAcceptable := errors.New("any")
@@ -164,41 +232,38 @@ func TestGoogleBreakerSelfProtection(t *testing.T) {
 }
 
 func TestGoogleBreakerHistory(t *testing.T) {
-	var b *googleBreaker
-	var accepts, total int64
-
 	sleep := testInterval
 	t.Run("accepts == total", func(t *testing.T) {
-		b = getGoogleBreaker()
+		b := getGoogleBreaker()
 		markSuccessWithDuration(b, 10, sleep/2)
-		accepts, total = b.history()
-		assert.Equal(t, int64(10), accepts)
-		assert.Equal(t, int64(10), total)
+		result := b.history()
+		assert.Equal(t, int64(10), result.accepts)
+		assert.Equal(t, int64(10), result.total)
 	})
 
 	t.Run("fail == total", func(t *testing.T) {
-		b = getGoogleBreaker()
+		b := getGoogleBreaker()
 		markFailedWithDuration(b, 10, sleep/2)
-		accepts, total = b.history()
-		assert.Equal(t, int64(0), accepts)
-		assert.Equal(t, int64(10), total)
+		result := b.history()
+		assert.Equal(t, int64(0), result.accepts)
+		assert.Equal(t, int64(10), result.total)
 	})
 
 	t.Run("accepts = 1/2 * total, fail = 1/2 * total", func(t *testing.T) {
-		b = getGoogleBreaker()
+		b := getGoogleBreaker()
 		markFailedWithDuration(b, 5, sleep/2)
 		markSuccessWithDuration(b, 5, sleep/2)
-		accepts, total = b.history()
-		assert.Equal(t, int64(5), accepts)
-		assert.Equal(t, int64(10), total)
+		result := b.history()
+		assert.Equal(t, int64(5), result.accepts)
+		assert.Equal(t, int64(10), result.total)
 	})
 
 	t.Run("auto reset rolling counter", func(t *testing.T) {
-		b = getGoogleBreaker()
+		b := getGoogleBreaker()
 		time.Sleep(testInterval * testBuckets)
-		accepts, total = b.history()
-		assert.Equal(t, int64(0), accepts)
-		assert.Equal(t, int64(0), total)
+		result := b.history()
+		assert.Equal(t, int64(0), result.accepts)
+		assert.Equal(t, int64(0), result.total)
 	})
 }
 
@@ -206,7 +271,7 @@ func BenchmarkGoogleBreakerAllow(b *testing.B) {
 	breaker := getGoogleBreaker()
 	b.ResetTimer()
 	for i := 0; i <= b.N; i++ {
-		breaker.accept()
+		_ = breaker.accept()
 		if i%2 == 0 {
 			breaker.markSuccess()
 		} else {
@@ -215,6 +280,16 @@ func BenchmarkGoogleBreakerAllow(b *testing.B) {
 	}
 }
 
+func BenchmarkGoogleBreakerDoReq(b *testing.B) {
+	breaker := getGoogleBreaker()
+	b.ResetTimer()
+	for i := 0; i <= b.N; i++ {
+		_ = breaker.doReq(func() error {
+			return nil
+		}, nil, defaultAcceptable)
+	}
+}
+
 func markSuccess(b *googleBreaker, count int) {
 	for i := 0; i < count; i++ {
 		p, err := b.allow()

core/breaker/nopbreaker.go

@@ -1,10 +1,13 @@
 package breaker
 
+import "context"
+
 const nopBreakerName = "nopBreaker"
 
 type nopBreaker struct{}
 
-func newNopBreaker() Breaker {
+// NopBreaker returns a breaker that never trigger breaker circuit.
+func NopBreaker() Breaker {
 	return nopBreaker{}
 }
 
@@ -16,20 +19,40 @@ func (b nopBreaker) Allow() (Promise, error) {
 	return nopPromise{}, nil
 }
 
+func (b nopBreaker) AllowCtx(_ context.Context) (Promise, error) {
+	return nopPromise{}, nil
+}
+
 func (b nopBreaker) Do(req func() error) error {
 	return req()
 }
 
+func (b nopBreaker) DoCtx(_ context.Context, req func() error) error {
+	return req()
+}
+
 func (b nopBreaker) DoWithAcceptable(req func() error, _ Acceptable) error {
 	return req()
 }
 
-func (b nopBreaker) DoWithFallback(req func() error, _ func(err error) error) error {
+func (b nopBreaker) DoWithAcceptableCtx(_ context.Context, req func() error, _ Acceptable) error {
 	return req()
 }
 
-func (b nopBreaker) DoWithFallbackAcceptable(req func() error, _ func(err error) error,
-	_ Acceptable) error {
+func (b nopBreaker) DoWithFallback(req func() error, _ Fallback) error {
+	return req()
+}
+
+func (b nopBreaker) DoWithFallbackCtx(_ context.Context, req func() error, _ Fallback) error {
+	return req()
+}
+
+func (b nopBreaker) DoWithFallbackAcceptable(req func() error, _ Fallback, _ Acceptable) error {
+	return req()
+}
+
+func (b nopBreaker) DoWithFallbackAcceptableCtx(_ context.Context, req func() error,
+	_ Fallback, _ Acceptable) error {
 	return req()
 }
 

core/breaker/nopbreaker_test.go

@@ -1,6 +1,7 @@
 package breaker
 
 import (
+	"context"
 	"errors"
 	"testing"
 
@@ -8,9 +9,11 @@ import (
 )
 
 func TestNopBreaker(t *testing.T) {
-	b := newNopBreaker()
+	b := NopBreaker()
 	assert.Equal(t, nopBreakerName, b.Name())
-	p, err := b.Allow()
+	_, err := b.Allow()
+	assert.Nil(t, err)
+	p, err := b.AllowCtx(context.Background())
 	assert.Nil(t, err)
 	p.Accept()
 	for i := 0; i < 1000; i++ {
@@ -21,18 +24,34 @@ func TestNopBreaker(t *testing.T) {
 	assert.Nil(t, b.Do(func() error {
 		return nil
 	}))
+	assert.Nil(t, b.DoCtx(context.Background(), func() error {
+		return nil
+	}))
 	assert.Nil(t, b.DoWithAcceptable(func() error {
 		return nil
 	}, defaultAcceptable))
+	assert.Nil(t, b.DoWithAcceptableCtx(context.Background(), func() error {
+		return nil
+	}, defaultAcceptable))
 	errDummy := errors.New("any")
 	assert.Equal(t, errDummy, b.DoWithFallback(func() error {
 		return errDummy
 	}, func(err error) error {
 		return nil
 	}))
+	assert.Equal(t, errDummy, b.DoWithFallbackCtx(context.Background(), func() error {
+		return errDummy
+	}, func(err error) error {
+		return nil
+	}))
 	assert.Equal(t, errDummy, b.DoWithFallbackAcceptable(func() error {
 		return errDummy
 	}, func(err error) error {
 		return nil
 	}, defaultAcceptable))
+	assert.Equal(t, errDummy, b.DoWithFallbackAcceptableCtx(context.Background(), func() error {
+		return errDummy
+	}, func(err error) error {
+		return nil
+	}, defaultAcceptable))
 }

core/codec/dh.go

@@ -23,7 +23,7 @@ var (
 	zero = big.NewInt(0)
 )
 
-// DhKey defines the Diffie Hellman key.
+// DhKey defines the Diffie-Hellman key.
 type DhKey struct {
 	PriKey *big.Int
 	PubKey *big.Int
@@ -46,7 +46,7 @@ func ComputeKey(pubKey, priKey *big.Int) (*big.Int, error) {
 	return new(big.Int).Exp(pubKey, priKey, p), nil
 }
 
-// GenerateKey returns a Diffie Hellman key.
+// GenerateKey returns a Diffie-Hellman key.
 func GenerateKey() (*DhKey, error) {
 	var err error
 	var x *big.Int

core/collection/cache.go

@@ -128,8 +128,8 @@ func (c *Cache) Take(key string, fetch func() (any, error)) (any, error) {
 
 	var fresh bool
 	val, err := c.barrier.Do(key, func() (any, error) {
-		// because O(1) on map search in memory, and fetch is an IO query
-		// so we do double check, cache might be taken by another call
+		// because O(1) on map search in memory, and fetch is an IO query,
+		// so we do double-check, cache might be taken by another call
 		if val, ok := c.doGet(key); ok {
 			return val, nil
 		}

core/collection/rollingwindow.go

@@ -4,18 +4,28 @@ import (
 	"sync"
 	"time"
 
+	"github.com/zeromicro/go-zero/core/mathx"
 	"github.com/zeromicro/go-zero/core/timex"
 )
 
 type (
-	// RollingWindowOption let callers customize the RollingWindow.
-	RollingWindowOption func(rollingWindow *RollingWindow)
+	// BucketInterface is the interface that defines the buckets.
+	BucketInterface[T Numerical] interface {
+		Add(v T)
+		Reset()
+	}
 
-	// RollingWindow defines a rolling window to calculate the events in buckets with time interval.
-	RollingWindow struct {
+	// Numerical is the interface that restricts the numerical type.
+	Numerical = mathx.Numerical
+
+	// RollingWindowOption let callers customize the RollingWindow.
+	RollingWindowOption[T Numerical, B BucketInterface[T]] func(rollingWindow *RollingWindow[T, B])
+
+	// RollingWindow defines a rolling window to calculate the events in buckets with the time interval.
+	RollingWindow[T Numerical, B BucketInterface[T]] struct {
 		lock          sync.RWMutex
 		size          int
-		win           *window
+		win           *window[T, B]
 		interval      time.Duration
 		offset        int
 		ignoreCurrent bool
@@ -25,14 +35,15 @@ type (
 
 // NewRollingWindow returns a RollingWindow that with size buckets and time interval,
 // use opts to customize the RollingWindow.
-func NewRollingWindow(size int, interval time.Duration, opts ...RollingWindowOption) *RollingWindow {
+func NewRollingWindow[T Numerical, B BucketInterface[T]](newBucket func() B, size int,
+	interval time.Duration, opts ...RollingWindowOption[T, B]) *RollingWindow[T, B] {
 	if size < 1 {
 		panic("size must be greater than 0")
 	}
 
-	w := &RollingWindow{
+	w := &RollingWindow[T, B]{
 		size:     size,
-		win:      newWindow(size),
+		win:      newWindow[T, B](newBucket, size),
 		interval: interval,
 		lastTime: timex.Now(),
 	}
@@ -43,7 +54,7 @@ func NewRollingWindow(size int, interval time.Duration, opts ...RollingWindowOpt
 }
 
 // Add adds value to current bucket.
-func (rw *RollingWindow) Add(v float64) {
+func (rw *RollingWindow[T, B]) Add(v T) {
 	rw.lock.Lock()
 	defer rw.lock.Unlock()
 	rw.updateOffset()
@@ -51,13 +62,13 @@ func (rw *RollingWindow) Add(v float64) {
 }
 
 // Reduce runs fn on all buckets, ignore current bucket if ignoreCurrent was set.
-func (rw *RollingWindow) Reduce(fn func(b *Bucket)) {
+func (rw *RollingWindow[T, B]) Reduce(fn func(b B)) {
 	rw.lock.RLock()
 	defer rw.lock.RUnlock()
 
 	var diff int
 	span := rw.span()
-	// ignore current bucket, because of partial data
+	// ignore the current bucket, because of partial data
 	if span == 0 && rw.ignoreCurrent {
 		diff = rw.size - 1
 	} else {
@@ -69,7 +80,7 @@ func (rw *RollingWindow) Reduce(fn func(b *Bucket)) {
 	}
 }
 
-func (rw *RollingWindow) span() int {
+func (rw *RollingWindow[T, B]) span() int {
 	offset := int(timex.Since(rw.lastTime) / rw.interval)
 	if 0 <= offset && offset < rw.size {
 		return offset
@@ -78,7 +89,7 @@ func (rw *RollingWindow) span() int {
 	return rw.size
 }
 
-func (rw *RollingWindow) updateOffset() {
+func (rw *RollingWindow[T, B]) updateOffset() {
 	span := rw.span()
 	if span <= 0 {
 		return
@@ -97,54 +108,54 @@ func (rw *RollingWindow) updateOffset() {
 }
 
 // Bucket defines the bucket that holds sum and num of additions.
-type Bucket struct {
-	Sum   float64
+type Bucket[T Numerical] struct {
+	Sum   T
 	Count int64
 }
 
-func (b *Bucket) add(v float64) {
+func (b *Bucket[T]) Add(v T) {
 	b.Sum += v
 	b.Count++
 }
 
-func (b *Bucket) reset() {
+func (b *Bucket[T]) Reset() {
 	b.Sum = 0
 	b.Count = 0
 }
 
-type window struct {
-	buckets []*Bucket
+type window[T Numerical, B BucketInterface[T]] struct {
+	buckets []B
 	size    int
 }
 
-func newWindow(size int) *window {
-	buckets := make([]*Bucket, size)
+func newWindow[T Numerical, B BucketInterface[T]](newBucket func() B, size int) *window[T, B] {
+	buckets := make([]B, size)
 	for i := 0; i < size; i++ {
-		buckets[i] = new(Bucket)
+		buckets[i] = newBucket()
 	}
-	return &window{
+	return &window[T, B]{
 		buckets: buckets,
 		size:    size,
 	}
 }
 
-func (w *window) add(offset int, v float64) {
-	w.buckets[offset%w.size].add(v)
+func (w *window[T, B]) add(offset int, v T) {
+	w.buckets[offset%w.size].Add(v)
 }
 
-func (w *window) reduce(start, count int, fn func(b *Bucket)) {
+func (w *window[T, B]) reduce(start, count int, fn func(b B)) {
 	for i := 0; i < count; i++ {
 		fn(w.buckets[(start+i)%w.size])
 	}
 }
 
-func (w *window) resetBucket(offset int) {
-	w.buckets[offset%w.size].reset()
+func (w *window[T, B]) resetBucket(offset int) {
+	w.buckets[offset%w.size].Reset()
 }
 
 // IgnoreCurrentBucket lets the Reduce call ignore current bucket.
-func IgnoreCurrentBucket() RollingWindowOption {
-	return func(w *RollingWindow) {
+func IgnoreCurrentBucket[T Numerical, B BucketInterface[T]]() RollingWindowOption[T, B] {
+	return func(w *RollingWindow[T, B]) {
 		w.ignoreCurrent = true
 	}
 }

core/collection/rollingwindow_test.go

@@ -12,18 +12,24 @@ import (
 const duration = time.Millisecond * 50
 
 func TestNewRollingWindow(t *testing.T) {
-	assert.NotNil(t, NewRollingWindow(10, time.Second))
+	assert.NotNil(t, NewRollingWindow[int64, *Bucket[int64]](func() *Bucket[int64] {
+		return new(Bucket[int64])
+	}, 10, time.Second))
 	assert.Panics(t, func() {
-		NewRollingWindow(0, time.Second)
+		NewRollingWindow[int64, *Bucket[int64]](func() *Bucket[int64] {
+			return new(Bucket[int64])
+		}, 0, time.Second)
 	})
 }
 
 func TestRollingWindowAdd(t *testing.T) {
 	const size = 3
-	r := NewRollingWindow(size, duration)
+	r := NewRollingWindow[float64, *Bucket[float64]](func() *Bucket[float64] {
+		return new(Bucket[float64])
+	}, size, duration)
 	listBuckets := func() []float64 {
 		var buckets []float64
-		r.Reduce(func(b *Bucket) {
+		r.Reduce(func(b *Bucket[float64]) {
 			buckets = append(buckets, b.Sum)
 		})
 		return buckets
@@ -47,10 +53,12 @@ func TestRollingWindowAdd(t *testing.T) {
 
 func TestRollingWindowReset(t *testing.T) {
 	const size = 3
-	r := NewRollingWindow(size, duration, IgnoreCurrentBucket())
+	r := NewRollingWindow[float64, *Bucket[float64]](func() *Bucket[float64] {
+		return new(Bucket[float64])
+	}, size, duration, IgnoreCurrentBucket[float64, *Bucket[float64]]())
 	listBuckets := func() []float64 {
 		var buckets []float64
-		r.Reduce(func(b *Bucket) {
+		r.Reduce(func(b *Bucket[float64]) {
 			buckets = append(buckets, b.Sum)
 		})
 		return buckets
@@ -72,15 +80,19 @@ func TestRollingWindowReset(t *testing.T) {
 func TestRollingWindowReduce(t *testing.T) {
 	const size = 4
 	tests := []struct {
-		win    *RollingWindow
+		win    *RollingWindow[float64, *Bucket[float64]]
 		expect float64
 	}{
 		{
-			win:    NewRollingWindow(size, duration),
+			win: NewRollingWindow[float64, *Bucket[float64]](func() *Bucket[float64] {
+				return new(Bucket[float64])
+			}, size, duration),
 			expect: 10,
 		},
 		{
-			win:    NewRollingWindow(size, duration, IgnoreCurrentBucket()),
+			win: NewRollingWindow[float64, *Bucket[float64]](func() *Bucket[float64] {
+				return new(Bucket[float64])
+			}, size, duration, IgnoreCurrentBucket[float64, *Bucket[float64]]()),
 			expect: 4,
 		},
 	}
@@ -97,7 +109,7 @@ func TestRollingWindowReduce(t *testing.T) {
 				}
 			}
 			var result float64
-			r.Reduce(func(b *Bucket) {
+			r.Reduce(func(b *Bucket[float64]) {
 				result += b.Sum
 			})
 			assert.Equal(t, test.expect, result)
@@ -108,10 +120,12 @@ func TestRollingWindowReduce(t *testing.T) {
 func TestRollingWindowBucketTimeBoundary(t *testing.T) {
 	const size = 3
 	interval := time.Millisecond * 30
-	r := NewRollingWindow(size, interval)
+	r := NewRollingWindow[float64, *Bucket[float64]](func() *Bucket[float64] {
+		return new(Bucket[float64])
+	}, size, interval)
 	listBuckets := func() []float64 {
 		var buckets []float64
-		r.Reduce(func(b *Bucket) {
+		r.Reduce(func(b *Bucket[float64]) {
 			buckets = append(buckets, b.Sum)
 		})
 		return buckets
@@ -138,7 +152,9 @@ func TestRollingWindowBucketTimeBoundary(t *testing.T) {
 
 func TestRollingWindowDataRace(t *testing.T) {
 	const size = 3
-	r := NewRollingWindow(size, duration)
+	r := NewRollingWindow[float64, *Bucket[float64]](func() *Bucket[float64] {
+		return new(Bucket[float64])
+	}, size, duration)
 	stop := make(chan bool)
 	go func() {
 		for {
@@ -157,7 +173,7 @@ func TestRollingWindowDataRace(t *testing.T) {
 			case <-stop:
 				return
 			default:
-				r.Reduce(func(b *Bucket) {})
+				r.Reduce(func(b *Bucket[float64]) {})
 			}
 		}
 	}()

core/collection/set.go

@@ -1,235 +1,53 @@
 package collection
 
-import (
-	"github.com/zeromicro/go-zero/core/lang"
-	"github.com/zeromicro/go-zero/core/logx"
-)
+import "github.com/zeromicro/go-zero/core/lang"
 
-const (
-	unmanaged = iota
-	untyped
-	intType
-	int64Type
-	uintType
-	uint64Type
-	stringType
-)
-
-// Set is not thread-safe, for concurrent use, make sure to use it with synchronization.
-type Set struct {
-	data map[any]lang.PlaceholderType
-	tp   int
+// Set is a type-safe generic set collection.
+// It's not thread-safe, use with synchronization for concurrent access.
+type Set[T comparable] struct {
+	data map[T]lang.PlaceholderType
 }
 
-// NewSet returns a managed Set, can only put the values with the same type.
-func NewSet() *Set {
-	return &Set{
-		data: make(map[any]lang.PlaceholderType),
-		tp:   untyped,
+// NewSet returns a new type-safe set.
+func NewSet[T comparable]() *Set[T] {
+	return &Set[T]{
+		data: make(map[T]lang.PlaceholderType),
 	}
 }
 
-// NewUnmanagedSet returns an unmanaged Set, which can put values with different types.
-func NewUnmanagedSet() *Set {
-	return &Set{
-		data: make(map[any]lang.PlaceholderType),
-		tp:   unmanaged,
+// Add adds items to the set. Duplicates are automatically ignored.
+func (s *Set[T]) Add(items ...T) {
+	for _, item := range items {
+		s.data[item] = lang.Placeholder
 	}
 }
 
-// Add adds i into s.
-func (s *Set) Add(i ...any) {
-	for _, each := range i {
-		s.add(each)
-	}
+// Clear removes all items from the set.
+func (s *Set[T]) Clear() {
+	clear(s.data)
 }
 
-// AddInt adds int values ii into s.
-func (s *Set) AddInt(ii ...int) {
-	for _, each := range ii {
-		s.add(each)
-	}
-}
-
-// AddInt64 adds int64 values ii into s.
-func (s *Set) AddInt64(ii ...int64) {
-	for _, each := range ii {
-		s.add(each)
-	}
-}
-
-// AddUint adds uint values ii into s.
-func (s *Set) AddUint(ii ...uint) {
-	for _, each := range ii {
-		s.add(each)
-	}
-}
-
-// AddUint64 adds uint64 values ii into s.
-func (s *Set) AddUint64(ii ...uint64) {
-	for _, each := range ii {
-		s.add(each)
-	}
-}
-
-// AddStr adds string values ss into s.
-func (s *Set) AddStr(ss ...string) {
-	for _, each := range ss {
-		s.add(each)
-	}
-}
-
-// Contains checks if i is in s.
-func (s *Set) Contains(i any) bool {
-	if len(s.data) == 0 {
-		return false
-	}
-
-	s.validate(i)
-	_, ok := s.data[i]
+// Contains checks if an item exists in the set.
+func (s *Set[T]) Contains(item T) bool {
+	_, ok := s.data[item]
 	return ok
 }
 
-// Keys returns the keys in s.
-func (s *Set) Keys() []any {
-	var keys []any
-
-	for key := range s.data {
-		keys = append(keys, key)
-	}
-
-	return keys
-}
-
-// KeysInt returns the int keys in s.
-func (s *Set) KeysInt() []int {
-	var keys []int
-
-	for key := range s.data {
-		if intKey, ok := key.(int); ok {
-			keys = append(keys, intKey)
-		}
-	}
-
-	return keys
-}
-
-// KeysInt64 returns int64 keys in s.
-func (s *Set) KeysInt64() []int64 {
-	var keys []int64
-
-	for key := range s.data {
-		if intKey, ok := key.(int64); ok {
-			keys = append(keys, intKey)
-		}
-	}
-
-	return keys
-}
-
-// KeysUint returns uint keys in s.
-func (s *Set) KeysUint() []uint {
-	var keys []uint
-
-	for key := range s.data {
-		if intKey, ok := key.(uint); ok {
-			keys = append(keys, intKey)
-		}
-	}
-
-	return keys
-}
-
-// KeysUint64 returns uint64 keys in s.
-func (s *Set) KeysUint64() []uint64 {
-	var keys []uint64
-
-	for key := range s.data {
-		if intKey, ok := key.(uint64); ok {
-			keys = append(keys, intKey)
-		}
-	}
-
-	return keys
-}
-
-// KeysStr returns string keys in s.
-func (s *Set) KeysStr() []string {
-	var keys []string
-
-	for key := range s.data {
-		if strKey, ok := key.(string); ok {
-			keys = append(keys, strKey)
-		}
-	}
-
-	return keys
-}
-
-// Remove removes i from s.
-func (s *Set) Remove(i any) {
-	s.validate(i)
-	delete(s.data, i)
-}
-
-// Count returns the number of items in s.
-func (s *Set) Count() int {
+// Count returns the number of items in the set.
+func (s *Set[T]) Count() int {
 	return len(s.data)
 }
 
-func (s *Set) add(i any) {
-	switch s.tp {
-	case unmanaged:
-		// do nothing
-	case untyped:
-		s.setType(i)
-	default:
-		s.validate(i)
+// Keys returns all elements in the set as a slice.
+func (s *Set[T]) Keys() []T {
+	keys := make([]T, 0, len(s.data))
+	for key := range s.data {
+		keys = append(keys, key)
 	}
-	s.data[i] = lang.Placeholder
+	return keys
 }
 
-func (s *Set) setType(i any) {
-	// s.tp can only be untyped here
-	switch i.(type) {
-	case int:
-		s.tp = intType
-	case int64:
-		s.tp = int64Type
-	case uint:
-		s.tp = uintType
-	case uint64:
-		s.tp = uint64Type
-	case string:
-		s.tp = stringType
-	}
-}
-
-func (s *Set) validate(i any) {
-	if s.tp == unmanaged {
-		return
-	}
-
-	switch i.(type) {
-	case int:
-		if s.tp != intType {
-			logx.Errorf("element is int, but set contains elements with type %d", s.tp)
-		}
-	case int64:
-		if s.tp != int64Type {
-			logx.Errorf("element is int64, but set contains elements with type %d", s.tp)
-		}
-	case uint:
-		if s.tp != uintType {
-			logx.Errorf("element is uint, but set contains elements with type %d", s.tp)
-		}
-	case uint64:
-		if s.tp != uint64Type {
-			logx.Errorf("element is uint64, but set contains elements with type %d", s.tp)
-		}
-	case string:
-		if s.tp != stringType {
-			logx.Errorf("element is string, but set contains elements with type %d", s.tp)
-		}
-	}
+// Remove removes an item from the set.
+func (s *Set[T]) Remove(item T) {
+	delete(s.data, item)
 }

core/collection/set_test.go

@@ -12,6 +12,105 @@ func init() {
 	logx.Disable()
 }
 
+// Set functionality tests
+func TestTypedSetInt(t *testing.T) {
+	set := NewSet[int]()
+	values := []int{1, 2, 3, 2, 1} // Contains duplicates
+
+	// Test adding
+	set.Add(values...)
+	assert.Equal(t, 3, set.Count()) // Should only have 3 elements after deduplication
+
+	// Test contains
+	assert.True(t, set.Contains(1))
+	assert.True(t, set.Contains(2))
+	assert.True(t, set.Contains(3))
+	assert.False(t, set.Contains(4))
+
+	// Test getting all keys
+	keys := set.Keys()
+	sort.Ints(keys)
+	assert.EqualValues(t, []int{1, 2, 3}, keys)
+
+	// Test removal
+	set.Remove(2)
+	assert.False(t, set.Contains(2))
+	assert.Equal(t, 2, set.Count())
+}
+
+func TestTypedSetStringOps(t *testing.T) {
+	set := NewSet[string]()
+	values := []string{"a", "b", "c", "b", "a"}
+
+	set.Add(values...)
+	assert.Equal(t, 3, set.Count())
+
+	assert.True(t, set.Contains("a"))
+	assert.True(t, set.Contains("b"))
+	assert.True(t, set.Contains("c"))
+	assert.False(t, set.Contains("d"))
+
+	keys := set.Keys()
+	sort.Strings(keys)
+	assert.EqualValues(t, []string{"a", "b", "c"}, keys)
+}
+
+func TestTypedSetClear(t *testing.T) {
+	set := NewSet[int]()
+	set.Add(1, 2, 3)
+	assert.Equal(t, 3, set.Count())
+
+	set.Clear()
+	assert.Equal(t, 0, set.Count())
+	assert.False(t, set.Contains(1))
+}
+
+func TestTypedSetEmpty(t *testing.T) {
+	set := NewSet[int]()
+	assert.Equal(t, 0, set.Count())
+	assert.False(t, set.Contains(1))
+	assert.Empty(t, set.Keys())
+}
+
+func TestTypedSetMultipleTypes(t *testing.T) {
+	// Test different typed generic sets
+	intSet := NewSet[int]()
+	int64Set := NewSet[int64]()
+	uintSet := NewSet[uint]()
+	uint64Set := NewSet[uint64]()
+	stringSet := NewSet[string]()
+
+	intSet.Add(1, 2, 3)
+	int64Set.Add(1, 2, 3)
+	uintSet.Add(1, 2, 3)
+	uint64Set.Add(1, 2, 3)
+	stringSet.Add("1", "2", "3")
+
+	assert.Equal(t, 3, intSet.Count())
+	assert.Equal(t, 3, int64Set.Count())
+	assert.Equal(t, 3, uintSet.Count())
+	assert.Equal(t, 3, uint64Set.Count())
+	assert.Equal(t, 3, stringSet.Count())
+}
+
+// Set benchmarks
+func BenchmarkTypedIntSet(b *testing.B) {
+	s := NewSet[int]()
+	for i := 0; i < b.N; i++ {
+		s.Add(i)
+		_ = s.Contains(i)
+	}
+}
+
+func BenchmarkTypedStringSet(b *testing.B) {
+	s := NewSet[string]()
+	for i := 0; i < b.N; i++ {
+		s.Add(string(rune(i)))
+		_ = s.Contains(string(rune(i)))
+	}
+}
+
+// Legacy tests remain unchanged for backward compatibility
 func BenchmarkRawSet(b *testing.B) {
 	m := make(map[any]struct{})
 	for i := 0; i < b.N; i++ {
@@ -20,26 +119,10 @@ func BenchmarkRawSet(b *testing.B) {
 	}
 }
 
-func BenchmarkUnmanagedSet(b *testing.B) {
-	s := NewUnmanagedSet()
-	for i := 0; i < b.N; i++ {
-		s.Add(i)
-		_ = s.Contains(i)
-	}
-}
-
-func BenchmarkSet(b *testing.B) {
-	s := NewSet()
-	for i := 0; i < b.N; i++ {
-		s.AddInt(i)
-		_ = s.Contains(i)
-	}
-}
-
 func TestAdd(t *testing.T) {
 	// given
-	set := NewUnmanagedSet()
-	values := []any{1, 2, 3}
+	set := NewSet[int]()
+	values := []int{1, 2, 3}
 
 	// when
 	set.Add(values...)
@@ -51,82 +134,74 @@ func TestAdd(t *testing.T) {
 
 func TestAddInt(t *testing.T) {
 	// given
-	set := NewSet()
+	set := NewSet[int]()
 	values := []int{1, 2, 3}
 
 	// when
-	set.AddInt(values...)
+	set.Add(values...)
 
 	// then
 	assert.True(t, set.Contains(1) && set.Contains(2) && set.Contains(3))
-	keys := set.KeysInt()
+	keys := set.Keys()
 	sort.Ints(keys)
 	assert.EqualValues(t, values, keys)
 }
 
 func TestAddInt64(t *testing.T) {
 	// given
-	set := NewSet()
+	set := NewSet[int64]()
 	values := []int64{1, 2, 3}
 
 	// when
-	set.AddInt64(values...)
+	set.Add(values...)
 
 	// then
-	assert.True(t, set.Contains(int64(1)) && set.Contains(int64(2)) && set.Contains(int64(3)))
-	assert.Equal(t, len(values), len(set.KeysInt64()))
+	assert.True(t, set.Contains(1) && set.Contains(2) && set.Contains(3))
+	assert.Equal(t, len(values), len(set.Keys()))
 }
 
 func TestAddUint(t *testing.T) {
 	// given
-	set := NewSet()
+	set := NewSet[uint]()
 	values := []uint{1, 2, 3}
 
 	// when
-	set.AddUint(values...)
+	set.Add(values...)
 
 	// then
-	assert.True(t, set.Contains(uint(1)) && set.Contains(uint(2)) && set.Contains(uint(3)))
-	assert.Equal(t, len(values), len(set.KeysUint()))
+	assert.True(t, set.Contains(1) && set.Contains(2) && set.Contains(3))
+	assert.Equal(t, len(values), len(set.Keys()))
 }
 
 func TestAddUint64(t *testing.T) {
 	// given
-	set := NewSet()
+	set := NewSet[uint64]()
 	values := []uint64{1, 2, 3}
 
 	// when
-	set.AddUint64(values...)
+	set.Add(values...)
 
 	// then
-	assert.True(t, set.Contains(uint64(1)) && set.Contains(uint64(2)) && set.Contains(uint64(3)))
-	assert.Equal(t, len(values), len(set.KeysUint64()))
+	assert.True(t, set.Contains(1) && set.Contains(2) && set.Contains(3))
+	assert.Equal(t, len(values), len(set.Keys()))
 }
 
 func TestAddStr(t *testing.T) {
 	// given
-	set := NewSet()
+	set := NewSet[string]()
 	values := []string{"1", "2", "3"}
 
 	// when
-	set.AddStr(values...)
+	set.Add(values...)
 
 	// then
 	assert.True(t, set.Contains("1") && set.Contains("2") && set.Contains("3"))
-	assert.Equal(t, len(values), len(set.KeysStr()))
+	assert.Equal(t, len(values), len(set.Keys()))
 }
 
 func TestContainsWithoutElements(t *testing.T) {
 	// given
-	set := NewSet()
-
-	// then
-	assert.False(t, set.Contains(1))
-}
-
-func TestContainsUnmanagedWithoutElements(t *testing.T) {
-	// given
-	set := NewUnmanagedSet()
+	set := NewSet[int]()
 
 	// then
 	assert.False(t, set.Contains(1))
@@ -134,8 +209,8 @@ func TestContainsUnmanagedWithoutElements(t *testing.T) {
 
 func TestRemove(t *testing.T) {
 	// given
-	set := NewSet()
-	set.Add([]any{1, 2, 3}...)
+	set := NewSet[int]()
+	set.Add([]int{1, 2, 3}...)
 
 	// when
 	set.Remove(2)
@@ -146,57 +221,9 @@ func TestRemove(t *testing.T) {
 
 func TestCount(t *testing.T) {
 	// given
-	set := NewSet()
-	set.Add([]any{1, 2, 3}...)
+	set := NewSet[int]()
+	set.Add([]int{1, 2, 3}...)
 
 	// then
 	assert.Equal(t, set.Count(), 3)
 }
-
-func TestKeysIntMismatch(t *testing.T) {
-	set := NewSet()
-	set.add(int64(1))
-	set.add(2)
-	vals := set.KeysInt()
-	assert.EqualValues(t, []int{2}, vals)
-}
-
-func TestKeysInt64Mismatch(t *testing.T) {
-	set := NewSet()
-	set.add(1)
-	set.add(int64(2))
-	vals := set.KeysInt64()
-	assert.EqualValues(t, []int64{2}, vals)
-}
-
-func TestKeysUintMismatch(t *testing.T) {
-	set := NewSet()
-	set.add(1)
-	set.add(uint(2))
-	vals := set.KeysUint()
-	assert.EqualValues(t, []uint{2}, vals)
-}
-
-func TestKeysUint64Mismatch(t *testing.T) {
-	set := NewSet()
-	set.add(1)
-	set.add(uint64(2))
-	vals := set.KeysUint64()
-	assert.EqualValues(t, []uint64{2}, vals)
-}
-
-func TestKeysStrMismatch(t *testing.T) {
-	set := NewSet()
-	set.add(1)
-	set.add("2")
-	vals := set.KeysStr()
-	assert.EqualValues(t, []string{"2"}, vals)
-}
-
-func TestSetType(t *testing.T) {
-	set := NewUnmanagedSet()
-	set.add(1)
-	set.add("2")
-	vals := set.Keys()
-	assert.ElementsMatch(t, []any{1, "2"}, vals)
-}

core/conf/config.go

@@ -62,7 +62,11 @@ func Load(file string, v any, opts ...Option) error {
 		return loader([]byte(os.ExpandEnv(string(content))), v)
 	}
 
-	return loader(content, v)
+	if err = loader(content, v); err != nil {
+		return err
+	}
+
+	return validate(v)
 }
 
 // LoadConfig loads config into v from file, .json, .yaml and .yml are acceptable.
@@ -85,7 +89,12 @@ func LoadFromJsonBytes(content []byte, v any) error {
 
 	lowerCaseKeyMap := toLowerCaseKeyMap(m, info)
 
-	return mapping.UnmarshalJsonMap(lowerCaseKeyMap, v, mapping.WithCanonicalKeyFunc(toLowerCase))
+	if err = mapping.UnmarshalJsonMap(lowerCaseKeyMap, v,
+		mapping.WithCanonicalKeyFunc(toLowerCase)); err != nil {
+		return err
+	}
+
+	return validate(v)
 }
 
 // LoadConfigFromJsonBytes loads config into v from content json bytes.
@@ -133,7 +142,7 @@ func addOrMergeFields(info *fieldInfo, key string, child *fieldInfo, fullName st
 			return newConflictKeyError(fullName)
 		}
 
-		if err := mergeFields(prev, key, child.children, fullName); err != nil {
+		if err := mergeFields(prev, child.children, fullName); err != nil {
 			return err
 		}
 	} else {
@@ -189,10 +198,10 @@ func buildFieldsInfo(tp reflect.Type, fullName string) (*fieldInfo, error) {
 	switch tp.Kind() {
 	case reflect.Struct:
 		return buildStructFieldsInfo(tp, fullName)
-	case reflect.Array, reflect.Slice:
+	case reflect.Array, reflect.Slice, reflect.Map:
 		return buildFieldsInfo(mapping.Deref(tp.Elem()), fullName)
 	case reflect.Chan, reflect.Func:
-		return nil, fmt.Errorf("unsupported type: %s", tp.Kind())
+		return nil, fmt.Errorf("unsupported type: %s, fullName: %s", tp.Kind(), fullName)
 	default:
 		return &fieldInfo{
 			children: make(map[string]*fieldInfo),
@@ -281,7 +290,7 @@ func getTagName(field reflect.StructField) string {
 	return field.Name
 }
 
-func mergeFields(prev *fieldInfo, key string, children map[string]*fieldInfo, fullName string) error {
+func mergeFields(prev *fieldInfo, children map[string]*fieldInfo, fullName string) error {
 	if len(prev.children) == 0 || len(children) == 0 {
 		return newConflictKeyError(fullName)
 	}
@@ -307,7 +316,7 @@ func toLowerCaseInterface(v any, info *fieldInfo) any {
 	case map[string]any:
 		return toLowerCaseKeyMap(vv, info)
 	case []any:
-		var arr []any
+		arr := make([]any, 0, len(vv))
 		for _, vvv := range vv {
 			arr = append(arr, toLowerCaseInterface(vvv, info))
 		}
@@ -332,6 +341,8 @@ func toLowerCaseKeyMap(m map[string]any, info *fieldInfo) map[string]any {
 			res[lk] = toLowerCaseInterface(v, ti)
 		} else if info.mapField != nil {
 			res[k] = toLowerCaseInterface(v, info.mapField)
+		} else if vv, ok := v.(map[string]any); ok {
+			res[k] = toLowerCaseKeyMap(vv, info)
 		} else {
 			res[k] = v
 		}

core/conf/config_test.go

@@ -1,6 +1,7 @@
 package conf
 
 import (
+	"errors"
 	"os"
 	"reflect"
 	"testing"
@@ -40,9 +41,8 @@ func TestConfigJson(t *testing.T) {
 	for _, test := range tests {
 		test := test
 		t.Run(test, func(t *testing.T) {
-			tmpfile, err := createTempFile(test, text)
+			tmpfile, err := createTempFile(t, test, text)
 			assert.Nil(t, err)
-			defer os.Remove(tmpfile)
 
 			var val struct {
 				A string `json:"a"`
@@ -82,9 +82,8 @@ c = "${FOO}"
 d = "abcd!@#$112"
 `
 	t.Setenv("FOO", "2")
-	tmpfile, err := createTempFile(".toml", text)
+	tmpfile, err := createTempFile(t, ".toml", text)
 	assert.Nil(t, err)
-	defer os.Remove(tmpfile)
 
 	var val struct {
 		A string `json:"a"`
@@ -105,9 +104,8 @@ b = 1
 c = "FOO"
 d = "abcd"
 `
-	tmpfile, err := createTempFile(".toml", text)
+	tmpfile, err := createTempFile(t, ".toml", text)
 	assert.Nil(t, err)
-	defer os.Remove(tmpfile)
 
 	var val struct {
 		A string `json:"a"`
@@ -127,9 +125,8 @@ func TestConfigWithLower(t *testing.T) {
 	text := `a = "foo"
 b = 1
 `
-	tmpfile, err := createTempFile(".toml", text)
+	tmpfile, err := createTempFile(t, ".toml", text)
 	assert.Nil(t, err)
-	defer os.Remove(tmpfile)
 
 	var val struct {
 		A string `json:"a"`
@@ -207,9 +204,8 @@ c = "${FOO}"
 d = "abcd!@#112"
 `
 	t.Setenv("FOO", "2")
-	tmpfile, err := createTempFile(".toml", text)
+	tmpfile, err := createTempFile(t, ".toml", text)
 	assert.Nil(t, err)
-	defer os.Remove(tmpfile)
 
 	var val struct {
 		A string `json:"a"`
@@ -241,9 +237,8 @@ func TestConfigJsonEnv(t *testing.T) {
 	for _, test := range tests {
 		test := test
 		t.Run(test, func(t *testing.T) {
-			tmpfile, err := createTempFile(test, text)
+			tmpfile, err := createTempFile(t, test, text)
 			assert.Nil(t, err)
-			defer os.Remove(tmpfile)
 
 			var val struct {
 				A string `json:"a"`
@@ -1192,6 +1187,42 @@ Email = "bar"`)
 			assert.Len(t, c.Value, 2)
 		}
 	})
+
+	t.Run("multi layer map", func(t *testing.T) {
+		type Value struct {
+			User struct {
+				Name string
+			}
+		}
+
+		type Config struct {
+			Value map[string]map[string]Value
+		}
+
+		var input = []byte(`
+[Value.first.User1.User]
+Name = "foo"
+[Value.second.User2.User]
+Name = "bar"
+`)
+		var c Config
+		if assert.NoError(t, LoadFromTomlBytes(input, &c)) {
+			assert.Len(t, c.Value, 2)
+		}
+	})
+}
+
+func Test_LoadBadConfig(t *testing.T) {
+	type Config struct {
+		Name string `json:"name,options=foo|bar"`
+	}
+
+	file, err := createTempFile(t, ".json", `{"name": "baz"}`)
+	assert.NoError(t, err)
+
+	var c Config
+	err = Load(file, &c)
+	assert.Error(t, err)
 }
 
 func Test_getFullName(t *testing.T) {
@@ -1199,6 +1230,26 @@ func Test_getFullName(t *testing.T) {
 	assert.Equal(t, "a", getFullName("", "a"))
 }
 
+func TestValidate(t *testing.T) {
+	t.Run("normal config", func(t *testing.T) {
+		var c mockConfig
+		err := LoadFromJsonBytes([]byte(`{"val": "hello", "number": 8}`), &c)
+		assert.NoError(t, err)
+	})
+
+	t.Run("error no int", func(t *testing.T) {
+		var c mockConfig
+		err := LoadFromJsonBytes([]byte(`{"val": "hello"}`), &c)
+		assert.Error(t, err)
+	})
+
+	t.Run("error no string", func(t *testing.T) {
+		var c mockConfig
+		err := LoadFromJsonBytes([]byte(`{"number": 8}`), &c)
+		assert.Error(t, err)
+	})
+}
+
 func Test_buildFieldsInfo(t *testing.T) {
 	type ParentSt struct {
 		Name string
@@ -1288,13 +1339,13 @@ func Test_buildFieldsInfo(t *testing.T) {
 	}
 }
 
-func createTempFile(ext, text string) (string, error) {
+func createTempFile(t *testing.T, ext, text string) (string, error) {
 	tmpFile, err := os.CreateTemp(os.TempDir(), hash.Md5Hex([]byte(text))+"*"+ext)
 	if err != nil {
 		return "", err
 	}
 
-	if err := os.WriteFile(tmpFile.Name(), []byte(text), os.ModeTemporary); err != nil {
+	if err = os.WriteFile(tmpFile.Name(), []byte(text), os.ModeTemporary); err != nil {
 		return "", err
 	}
 
@@ -1303,5 +1354,26 @@ func createTempFile(ext, text string) (string, error) {
 		return "", err
 	}
 
+	t.Cleanup(func() {
+		_ = os.Remove(filename)
+	})
+
 	return filename, nil
 }
+
+type mockConfig struct {
+	Val    string
+	Number int
+}
+
+func (m mockConfig) Validate() error {
+	if len(m.Val) == 0 {
+		return errors.New("val is empty")
+	}
+
+	if m.Number == 0 {
+		return errors.New("number is zero")
+	}
+
+	return nil
+}

core/conf/readme.md

@@ -12,7 +12,7 @@ type RestfulConf struct {
 	MaxConns     int           `json:",default=10000"`
 	MaxBytes     int64         `json:",default=1048576"`
 	Timeout      time.Duration `json:",default=3s"`
-	CpuThreshold int64         `json:",default=900,range=[0:1000]"`
+	CpuThreshold int64         `json:",default=900,range=[0:1000)"`
 }
 ```
 

core/conf/validate.go

@@ -0,0 +1,12 @@
+package conf
+
+import "github.com/zeromicro/go-zero/core/validation"
+
+// validate validates the value if it implements the Validator interface.
+func validate(v any) error {
+	if val, ok := v.(validation.Validator); ok {
+		return val.Validate()
+	}
+
+	return nil
+}

core/conf/validate_test.go

@@ -0,0 +1,81 @@
+package conf
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type mockType int
+
+func (m mockType) Validate() error {
+	if m < 10 {
+		return errors.New("invalid value")
+	}
+
+	return nil
+}
+
+type anotherMockType int
+
+func Test_validate(t *testing.T) {
+	tests := []struct {
+		name    string
+		v       any
+		wantErr bool
+	}{
+		{
+			name:    "invalid",
+			v:       mockType(5),
+			wantErr: true,
+		},
+		{
+			name:    "valid",
+			v:       mockType(10),
+			wantErr: false,
+		},
+		{
+			name:    "not validator",
+			v:       anotherMockType(5),
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validate(tt.v)
+			assert.Equal(t, tt.wantErr, err != nil)
+		})
+	}
+}
+
+type mockVal struct {
+}
+
+func (m mockVal) Validate() error {
+	return errors.New("invalid value")
+}
+
+func Test_validateValPtr(t *testing.T) {
+	tests := []struct {
+		name    string
+		v       any
+		wantErr bool
+	}{
+		{
+			name: "invalid",
+			v:    mockVal{},
+		},
+		{
+			name: "invalid value",
+			v:    &mockVal{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Error(t, validate(tt.v))
+		})
+	}
+}

core/configcenter/configurator.go

@@ -0,0 +1,200 @@
+package configurator
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+	"sync"
+	"sync/atomic"
+
+	"github.com/zeromicro/go-zero/core/configcenter/subscriber"
+	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/mapping"
+	"github.com/zeromicro/go-zero/core/threading"
+)
+
+var (
+	errEmptyConfig            = errors.New("empty config value")
+	errMissingUnmarshalerType = errors.New("missing unmarshaler type")
+)
+
+// Configurator is the interface for configuration center.
+type Configurator[T any] interface {
+	// GetConfig returns the subscription value.
+	GetConfig() (T, error)
+	// AddListener adds a listener to the subscriber.
+	AddListener(listener func())
+}
+
+type (
+	// Config is the configuration for Configurator.
+	Config struct {
+		// Type is the value type, yaml, json or toml.
+		Type string `json:",default=yaml,options=[yaml,json,toml]"`
+		// Log is the flag to control logging.
+		Log bool `json:",default=true"`
+	}
+
+	configCenter[T any] struct {
+		conf        Config
+		unmarshaler LoaderFn
+		subscriber  subscriber.Subscriber
+		listeners   []func()
+		lock        sync.Mutex
+		snapshot    atomic.Value
+	}
+
+	value[T any] struct {
+		data        string
+		marshalData T
+		err         error
+	}
+)
+
+// Configurator is the interface for configuration center.
+var _ Configurator[any] = (*configCenter[any])(nil)
+
+// MustNewConfigCenter returns a Configurator, exits on errors.
+func MustNewConfigCenter[T any](c Config, subscriber subscriber.Subscriber) Configurator[T] {
+	cc, err := NewConfigCenter[T](c, subscriber)
+	logx.Must(err)
+	return cc
+}
+
+// NewConfigCenter returns a Configurator.
+func NewConfigCenter[T any](c Config, subscriber subscriber.Subscriber) (Configurator[T], error) {
+	unmarshaler, ok := Unmarshaler(strings.ToLower(c.Type))
+	if !ok {
+		return nil, fmt.Errorf("unknown format: %s", c.Type)
+	}
+
+	cc := &configCenter[T]{
+		conf:        c,
+		unmarshaler: unmarshaler,
+		subscriber:  subscriber,
+	}
+
+	if err := cc.loadConfig(); err != nil {
+		return nil, err
+	}
+
+	if err := cc.subscriber.AddListener(cc.onChange); err != nil {
+		return nil, err
+	}
+
+	if _, err := cc.GetConfig(); err != nil {
+		return nil, err
+	}
+
+	return cc, nil
+}
+
+// AddListener adds listener to s.
+func (c *configCenter[T]) AddListener(listener func()) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.listeners = append(c.listeners, listener)
+}
+
+// GetConfig return structured config.
+func (c *configCenter[T]) GetConfig() (T, error) {
+	v := c.value()
+	if v == nil || len(v.data) == 0 {
+		var empty T
+		return empty, errEmptyConfig
+	}
+
+	return v.marshalData, v.err
+}
+
+// Value returns the subscription value.
+func (c *configCenter[T]) Value() string {
+	v := c.value()
+	if v == nil {
+		return ""
+	}
+	return v.data
+}
+
+func (c *configCenter[T]) loadConfig() error {
+	v, err := c.subscriber.Value()
+	if err != nil {
+		if c.conf.Log {
+			logx.Errorf("ConfigCenter loads changed configuration, error: %v", err)
+		}
+		return err
+	}
+
+	if c.conf.Log {
+		logx.Infof("ConfigCenter loads changed configuration, content [%s]", v)
+	}
+
+	c.snapshot.Store(c.genValue(v))
+	return nil
+}
+
+func (c *configCenter[T]) onChange() {
+	if err := c.loadConfig(); err != nil {
+		return
+	}
+
+	c.lock.Lock()
+	listeners := make([]func(), len(c.listeners))
+	copy(listeners, c.listeners)
+	c.lock.Unlock()
+
+	for _, l := range listeners {
+		threading.GoSafe(l)
+	}
+}
+
+func (c *configCenter[T]) value() *value[T] {
+	content := c.snapshot.Load()
+	if content == nil {
+		return nil
+	}
+	return content.(*value[T])
+}
+
+func (c *configCenter[T]) genValue(data string) *value[T] {
+	v := &value[T]{
+		data: data,
+	}
+	if len(data) == 0 {
+		return v
+	}
+
+	t := reflect.TypeOf(v.marshalData)
+	// if the type is nil, it means that the user has not set the type of the configuration.
+	if t == nil {
+		v.err = errMissingUnmarshalerType
+		return v
+	}
+
+	t = mapping.Deref(t)
+	switch t.Kind() {
+	case reflect.Struct, reflect.Array, reflect.Slice:
+		if err := c.unmarshaler([]byte(data), &v.marshalData); err != nil {
+			v.err = err
+			if c.conf.Log {
+				logx.Errorf("ConfigCenter unmarshal configuration failed, err: %+v, content [%s]",
+					err.Error(), data)
+			}
+		}
+	case reflect.String:
+		if str, ok := any(data).(T); ok {
+			v.marshalData = str
+		} else {
+			v.err = errMissingUnmarshalerType
+		}
+	default:
+		if c.conf.Log {
+			logx.Errorf("ConfigCenter unmarshal configuration missing unmarshaler for type: %s, content [%s]",
+				t.Kind(), data)
+		}
+		v.err = errMissingUnmarshalerType
+	}
+
+	return v
+}

core/configcenter/configurator_test.go

@@ -0,0 +1,233 @@
+package configurator
+
+import (
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewConfigCenter(t *testing.T) {
+	_, err := NewConfigCenter[any](Config{
+		Log: true,
+	}, &mockSubscriber{})
+	assert.Error(t, err)
+
+	_, err = NewConfigCenter[any](Config{
+		Type: "json",
+		Log:  true,
+	}, &mockSubscriber{})
+	assert.Error(t, err)
+}
+
+func TestConfigCenter_GetConfig(t *testing.T) {
+	mock := &mockSubscriber{}
+	type Data struct {
+		Name string `json:"name"`
+	}
+
+	mock.v = `{"name": "go-zero"}`
+	c1, err := NewConfigCenter[Data](Config{
+		Type: "json",
+		Log:  true,
+	}, mock)
+	assert.NoError(t, err)
+
+	data, err := c1.GetConfig()
+	assert.NoError(t, err)
+	assert.Equal(t, "go-zero", data.Name)
+
+	mock.v = `{"name": "111"}`
+	c2, err := NewConfigCenter[Data](Config{Type: "json"}, mock)
+	assert.NoError(t, err)
+
+	mock.v = `{}`
+	c3, err := NewConfigCenter[string](Config{
+		Type: "json",
+		Log:  true,
+	}, mock)
+	assert.NoError(t, err)
+	_, err = c3.GetConfig()
+	assert.NoError(t, err)
+
+	data, err = c2.GetConfig()
+	assert.NoError(t, err)
+
+	mock.lisErr = errors.New("mock error")
+	_, err = NewConfigCenter[Data](Config{
+		Type: "json",
+		Log:  true,
+	}, mock)
+	assert.Error(t, err)
+}
+
+func TestConfigCenter_onChange(t *testing.T) {
+	mock := &mockSubscriber{}
+	type Data struct {
+		Name string `json:"name"`
+	}
+
+	mock.v = `{"name": "go-zero"}`
+	c1, err := NewConfigCenter[Data](Config{Type: "json", Log: true}, mock)
+	assert.NoError(t, err)
+
+	data, err := c1.GetConfig()
+	assert.NoError(t, err)
+	assert.Equal(t, "go-zero", data.Name)
+
+	mock.v = `{"name": "go-zero2"}`
+	mock.change()
+
+	data, err = c1.GetConfig()
+	assert.NoError(t, err)
+	assert.Equal(t, "go-zero2", data.Name)
+
+	mock.valErr = errors.New("mock error")
+	_, err = NewConfigCenter[Data](Config{Type: "json", Log: false}, mock)
+	assert.Error(t, err)
+}
+
+func TestConfigCenter_Value(t *testing.T) {
+	mock := &mockSubscriber{}
+	mock.v = "1234"
+
+	c, err := NewConfigCenter[string](Config{
+		Type: "json",
+		Log:  true,
+	}, mock)
+	assert.NoError(t, err)
+
+	cc := c.(*configCenter[string])
+
+	assert.Equal(t, cc.Value(), "1234")
+
+	mock.valErr = errors.New("mock error")
+
+	_, err = NewConfigCenter[any](Config{
+		Type: "json",
+		Log:  true,
+	}, mock)
+	assert.Error(t, err)
+}
+
+func TestConfigCenter_AddListener(t *testing.T) {
+	mock := &mockSubscriber{}
+	mock.v = "1234"
+	c, err := NewConfigCenter[string](Config{
+		Type: "json",
+		Log:  true,
+	}, mock)
+	assert.NoError(t, err)
+
+	cc := c.(*configCenter[string])
+	var a, b int
+	var mutex sync.Mutex
+	cc.AddListener(func() {
+		mutex.Lock()
+		a = 1
+		mutex.Unlock()
+	})
+	cc.AddListener(func() {
+		mutex.Lock()
+		b = 2
+		mutex.Unlock()
+	})
+
+	assert.Equal(t, 2, len(cc.listeners))
+
+	mock.change()
+
+	time.Sleep(time.Millisecond * 100)
+
+	mutex.Lock()
+	assert.Equal(t, 1, a)
+	assert.Equal(t, 2, b)
+	mutex.Unlock()
+}
+
+func TestConfigCenter_genValue(t *testing.T) {
+	t.Run("data is empty", func(t *testing.T) {
+		c := &configCenter[string]{
+			unmarshaler: registry.unmarshalers["json"],
+			conf:        Config{Log: true},
+		}
+		v := c.genValue("")
+		assert.Equal(t, "", v.data)
+	})
+
+	t.Run("invalid template type", func(t *testing.T) {
+		c := &configCenter[any]{
+			unmarshaler: registry.unmarshalers["json"],
+			conf:        Config{Log: true},
+		}
+		v := c.genValue("xxxx")
+		assert.Equal(t, errMissingUnmarshalerType, v.err)
+	})
+
+	t.Run("unsupported template type", func(t *testing.T) {
+		c := &configCenter[int]{
+			unmarshaler: registry.unmarshalers["json"],
+			conf:        Config{Log: true},
+		}
+		v := c.genValue("1")
+		assert.Equal(t, errMissingUnmarshalerType, v.err)
+	})
+
+	t.Run("supported template string type", func(t *testing.T) {
+		c := &configCenter[string]{
+			unmarshaler: registry.unmarshalers["json"],
+			conf:        Config{Log: true},
+		}
+		v := c.genValue("12345")
+		assert.NoError(t, v.err)
+		assert.Equal(t, "12345", v.data)
+	})
+
+	t.Run("unmarshal fail", func(t *testing.T) {
+		c := &configCenter[struct {
+			Name string `json:"name"`
+		}]{
+			unmarshaler: registry.unmarshalers["json"],
+			conf:        Config{Log: true},
+		}
+		v := c.genValue(`{"name":"new name}`)
+		assert.Equal(t, `{"name":"new name}`, v.data)
+		assert.Error(t, v.err)
+	})
+
+	t.Run("success", func(t *testing.T) {
+		c := &configCenter[struct {
+			Name string `json:"name"`
+		}]{
+			unmarshaler: registry.unmarshalers["json"],
+			conf:        Config{Log: true},
+		}
+		v := c.genValue(`{"name":"new name"}`)
+		assert.Equal(t, `{"name":"new name"}`, v.data)
+		assert.Equal(t, "new name", v.marshalData.Name)
+		assert.NoError(t, v.err)
+	})
+}
+
+type mockSubscriber struct {
+	v              string
+	lisErr, valErr error
+	listener       func()
+}
+
+func (m *mockSubscriber) AddListener(listener func()) error {
+	m.listener = listener
+	return m.lisErr
+}
+
+func (m *mockSubscriber) Value() (string, error) {
+	return m.v, m.valErr
+}
+
+func (m *mockSubscriber) change() {
+	if m.listener != nil {
+		m.listener()
+	}
+}

core/configcenter/subscriber/etcd.go

@@ -0,0 +1,67 @@
+package subscriber
+
+import (
+	"github.com/zeromicro/go-zero/core/discov"
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+type (
+	// etcdSubscriber is a subscriber that subscribes to etcd.
+	etcdSubscriber struct {
+		*discov.Subscriber
+	}
+
+	// EtcdConf is the configuration for etcd.
+	EtcdConf = discov.EtcdConf
+)
+
+// MustNewEtcdSubscriber returns an etcd Subscriber, exits on errors.
+func MustNewEtcdSubscriber(conf EtcdConf) Subscriber {
+	s, err := NewEtcdSubscriber(conf)
+	logx.Must(err)
+	return s
+}
+
+// NewEtcdSubscriber returns an etcd Subscriber.
+func NewEtcdSubscriber(conf EtcdConf) (Subscriber, error) {
+	opts := buildSubOptions(conf)
+	s, err := discov.NewSubscriber(conf.Hosts, conf.Key, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &etcdSubscriber{Subscriber: s}, nil
+}
+
+// buildSubOptions constructs the options for creating a new etcd subscriber.
+func buildSubOptions(conf EtcdConf) []discov.SubOption {
+	opts := []discov.SubOption{
+		discov.WithExactMatch(),
+	}
+
+	if len(conf.User) > 0 {
+		opts = append(opts, discov.WithSubEtcdAccount(conf.User, conf.Pass))
+	}
+	if len(conf.CertFile) > 0 || len(conf.CertKeyFile) > 0 || len(conf.CACertFile) > 0 {
+		opts = append(opts, discov.WithSubEtcdTLS(conf.CertFile, conf.CertKeyFile,
+			conf.CACertFile, conf.InsecureSkipVerify))
+	}
+
+	return opts
+}
+
+// AddListener adds a listener to the subscriber.
+func (s *etcdSubscriber) AddListener(listener func()) error {
+	s.Subscriber.AddListener(listener)
+	return nil
+}
+
+// Value returns the value of the subscriber.
+func (s *etcdSubscriber) Value() (string, error) {
+	vs := s.Subscriber.Values()
+	if len(vs) > 0 {
+		return vs[len(vs)-1], nil
+	}
+
+	return "", nil
+}

core/configcenter/subscriber/subscriber.go

@@ -0,0 +1,9 @@
+package subscriber
+
+// Subscriber is the interface for configcenter subscribers.
+type Subscriber interface {
+	// AddListener adds a listener to the subscriber.
+	AddListener(listener func()) error
+	// Value returns the value of the subscriber.
+	Value() (string, error)
+}

core/configcenter/unmarshaler.go

@@ -0,0 +1,41 @@
+package configurator
+
+import (
+	"sync"
+
+	"github.com/zeromicro/go-zero/core/conf"
+)
+
+var registry = &unmarshalerRegistry{
+	unmarshalers: map[string]LoaderFn{
+		"json": conf.LoadFromJsonBytes,
+		"toml": conf.LoadFromTomlBytes,
+		"yaml": conf.LoadFromYamlBytes,
+	},
+}
+
+type (
+	// LoaderFn is the function type for loading configuration.
+	LoaderFn func([]byte, any) error
+
+	// unmarshalerRegistry is the registry for unmarshalers.
+	unmarshalerRegistry struct {
+		unmarshalers map[string]LoaderFn
+		mu           sync.RWMutex
+	}
+)
+
+// RegisterUnmarshaler registers an unmarshaler.
+func RegisterUnmarshaler(name string, fn LoaderFn) {
+	registry.mu.Lock()
+	defer registry.mu.Unlock()
+	registry.unmarshalers[name] = fn
+}
+
+// Unmarshaler returns the unmarshaler by name.
+func Unmarshaler(name string) (LoaderFn, bool) {
+	registry.mu.RLock()
+	defer registry.mu.RUnlock()
+	fn, ok := registry.unmarshalers[name]
+	return fn, ok
+}

core/configcenter/unmarshaler_test.go

@@ -0,0 +1,28 @@
+package configurator
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRegisterUnmarshaler(t *testing.T) {
+	RegisterUnmarshaler("test", func(data []byte, v interface{}) error {
+		return nil
+	})
+
+	_, ok := Unmarshaler("test")
+	assert.True(t, ok)
+
+	_, ok = Unmarshaler("test2")
+	assert.False(t, ok)
+
+	_, ok = Unmarshaler("json")
+	assert.True(t, ok)
+
+	_, ok = Unmarshaler("toml")
+	assert.True(t, ok)
+
+	_, ok = Unmarshaler("yaml")
+	assert.True(t, ok)
+}

core/discov/internal/etcdclient_mock.go

@@ -1,5 +1,10 @@
 // Code generated by MockGen. DO NOT EDIT.
 // Source: etcdclient.go
+//
+// Generated by this command:
+//
+//	mockgen -package internal -destination etcdclient_mock.go -source etcdclient.go EtcdClient
+//
 
 // Package internal is a generated GoMock package.
 package internal
@@ -8,35 +13,36 @@ import (
 	context "context"
 	reflect "reflect"
 
-	gomock "github.com/golang/mock/gomock"
 	clientv3 "go.etcd.io/etcd/client/v3"
+	gomock "go.uber.org/mock/gomock"
 	grpc "google.golang.org/grpc"
 )
 
-// MockEtcdClient is a mock of EtcdClient interface
+// MockEtcdClient is a mock of EtcdClient interface.
 type MockEtcdClient struct {
 	ctrl     *gomock.Controller
 	recorder *MockEtcdClientMockRecorder
+	isgomock struct{}
 }
 
-// MockEtcdClientMockRecorder is the mock recorder for MockEtcdClient
+// MockEtcdClientMockRecorder is the mock recorder for MockEtcdClient.
 type MockEtcdClientMockRecorder struct {
 	mock *MockEtcdClient
 }
 
-// NewMockEtcdClient creates a new mock instance
+// NewMockEtcdClient creates a new mock instance.
 func NewMockEtcdClient(ctrl *gomock.Controller) *MockEtcdClient {
 	mock := &MockEtcdClient{ctrl: ctrl}
 	mock.recorder = &MockEtcdClientMockRecorder{mock}
 	return mock
 }
 
-// EXPECT returns an object that allows the caller to indicate expected use
+// EXPECT returns an object that allows the caller to indicate expected use.
 func (m *MockEtcdClient) EXPECT() *MockEtcdClientMockRecorder {
 	return m.recorder
 }
 
-// ActiveConnection mocks base method
+// ActiveConnection mocks base method.
 func (m *MockEtcdClient) ActiveConnection() *grpc.ClientConn {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "ActiveConnection")
@@ -44,13 +50,13 @@ func (m *MockEtcdClient) ActiveConnection() *grpc.ClientConn {
 	return ret0
 }
 
-// ActiveConnection indicates an expected call of ActiveConnection
+// ActiveConnection indicates an expected call of ActiveConnection.
 func (mr *MockEtcdClientMockRecorder) ActiveConnection() *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ActiveConnection", reflect.TypeOf((*MockEtcdClient)(nil).ActiveConnection))
 }
 
-// Close mocks base method
+// Close mocks base method.
 func (m *MockEtcdClient) Close() error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Close")
@@ -58,13 +64,13 @@ func (m *MockEtcdClient) Close() error {
 	return ret0
 }
 
-// Close indicates an expected call of Close
+// Close indicates an expected call of Close.
 func (mr *MockEtcdClientMockRecorder) Close() *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockEtcdClient)(nil).Close))
 }
 
-// Ctx mocks base method
+// Ctx mocks base method.
 func (m *MockEtcdClient) Ctx() context.Context {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Ctx")
@@ -72,13 +78,13 @@ func (m *MockEtcdClient) Ctx() context.Context {
 	return ret0
 }
 
-// Ctx indicates an expected call of Ctx
+// Ctx indicates an expected call of Ctx.
 func (mr *MockEtcdClientMockRecorder) Ctx() *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Ctx", reflect.TypeOf((*MockEtcdClient)(nil).Ctx))
 }
 
-// Get mocks base method
+// Get mocks base method.
 func (m *MockEtcdClient) Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
 	m.ctrl.T.Helper()
 	varargs := []any{ctx, key}
@@ -91,14 +97,14 @@ func (m *MockEtcdClient) Get(ctx context.Context, key string, opts ...clientv3.O
 	return ret0, ret1
 }
 
-// Get indicates an expected call of Get
+// Get indicates an expected call of Get.
 func (mr *MockEtcdClientMockRecorder) Get(ctx, key any, opts ...any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	varargs := append([]any{ctx, key}, opts...)
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockEtcdClient)(nil).Get), varargs...)
 }
 
-// Grant mocks base method
+// Grant mocks base method.
 func (m *MockEtcdClient) Grant(ctx context.Context, ttl int64) (*clientv3.LeaseGrantResponse, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Grant", ctx, ttl)
@@ -107,13 +113,13 @@ func (m *MockEtcdClient) Grant(ctx context.Context, ttl int64) (*clientv3.LeaseG
 	return ret0, ret1
 }
 
-// Grant indicates an expected call of Grant
+// Grant indicates an expected call of Grant.
 func (mr *MockEtcdClientMockRecorder) Grant(ctx, ttl any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Grant", reflect.TypeOf((*MockEtcdClient)(nil).Grant), ctx, ttl)
 }
 
-// KeepAlive mocks base method
+// KeepAlive mocks base method.
 func (m *MockEtcdClient) KeepAlive(ctx context.Context, id clientv3.LeaseID) (<-chan *clientv3.LeaseKeepAliveResponse, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "KeepAlive", ctx, id)
@@ -122,13 +128,13 @@ func (m *MockEtcdClient) KeepAlive(ctx context.Context, id clientv3.LeaseID) (<-
 	return ret0, ret1
 }
 
-// KeepAlive indicates an expected call of KeepAlive
+// KeepAlive indicates an expected call of KeepAlive.
 func (mr *MockEtcdClientMockRecorder) KeepAlive(ctx, id any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "KeepAlive", reflect.TypeOf((*MockEtcdClient)(nil).KeepAlive), ctx, id)
 }
 
-// Put mocks base method
+// Put mocks base method.
 func (m *MockEtcdClient) Put(ctx context.Context, key, val string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) {
 	m.ctrl.T.Helper()
 	varargs := []any{ctx, key, val}
@@ -141,14 +147,14 @@ func (m *MockEtcdClient) Put(ctx context.Context, key, val string, opts ...clien
 	return ret0, ret1
 }
 
-// Put indicates an expected call of Put
+// Put indicates an expected call of Put.
 func (mr *MockEtcdClientMockRecorder) Put(ctx, key, val any, opts ...any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	varargs := append([]any{ctx, key, val}, opts...)
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Put", reflect.TypeOf((*MockEtcdClient)(nil).Put), varargs...)
 }
 
-// Revoke mocks base method
+// Revoke mocks base method.
 func (m *MockEtcdClient) Revoke(ctx context.Context, id clientv3.LeaseID) (*clientv3.LeaseRevokeResponse, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "Revoke", ctx, id)
@@ -157,13 +163,13 @@ func (m *MockEtcdClient) Revoke(ctx context.Context, id clientv3.LeaseID) (*clie
 	return ret0, ret1
 }
 
-// Revoke indicates an expected call of Revoke
+// Revoke indicates an expected call of Revoke.
 func (mr *MockEtcdClientMockRecorder) Revoke(ctx, id any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Revoke", reflect.TypeOf((*MockEtcdClient)(nil).Revoke), ctx, id)
 }
 
-// Watch mocks base method
+// Watch mocks base method.
 func (m *MockEtcdClient) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
 	m.ctrl.T.Helper()
 	varargs := []any{ctx, key}
@@ -175,7 +181,7 @@ func (m *MockEtcdClient) Watch(ctx context.Context, key string, opts ...clientv3
 	return ret0
 }
 
-// Watch indicates an expected call of Watch
+// Watch indicates an expected call of Watch.
 func (mr *MockEtcdClientMockRecorder) Watch(ctx, key any, opts ...any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	varargs := append([]any{ctx, key}, opts...)

core/discov/internal/registry.go

@@ -10,27 +10,30 @@ import (
 	"sync"
 	"time"
 
-	"github.com/zeromicro/go-zero/core/contextx"
 	"github.com/zeromicro/go-zero/core/lang"
-	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/logc"
+	"github.com/zeromicro/go-zero/core/mathx"
 	"github.com/zeromicro/go-zero/core/syncx"
 	"github.com/zeromicro/go-zero/core/threading"
 	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
+const coolDownDeviation = 0.05
+
 var (
 	registry = Registry{
 		clusters: make(map[string]*cluster),
 	}
-	connManager = syncx.NewResourceManager()
-	errClosed   = errors.New("etcd monitor chan has been closed")
+	connManager      = syncx.NewResourceManager()
+	coolDownUnstable = mathx.NewUnstable(coolDownDeviation)
+	errClosed        = errors.New("etcd monitor chan has been closed")
 )
 
 // A Registry is a registry that manages the etcd client connections.
 type Registry struct {
 	clusters map[string]*cluster
-	lock     sync.Mutex
+	lock     sync.RWMutex
 }
 
 // GetRegistry returns a global Registry.
@@ -40,60 +43,148 @@ func GetRegistry() *Registry {
 
 // GetConn returns an etcd client connection associated with given endpoints.
 func (r *Registry) GetConn(endpoints []string) (EtcdClient, error) {
-	c, _ := r.getCluster(endpoints)
+	c, _ := r.getOrCreateCluster(endpoints)
 	return c.getClient()
 }
 
 // Monitor monitors the key on given etcd endpoints, notify with the given UpdateListener.
-func (r *Registry) Monitor(endpoints []string, key string, l UpdateListener) error {
-	c, exists := r.getCluster(endpoints)
+func (r *Registry) Monitor(endpoints []string, key string, exactMatch bool, l UpdateListener) error {
+	wkey := watchKey{
+		key:        key,
+		exactMatch: exactMatch,
+	}
+
+	c, exists := r.getOrCreateCluster(endpoints)
 	// if exists, the existing values should be updated to the listener.
 	if exists {
-		kvs := c.getCurrent(key)
-		for _, kv := range kvs {
-			l.OnAdd(kv)
+		c.lock.Lock()
+		watcher, ok := c.watchers[wkey]
+		if ok {
+			watcher.listeners = append(watcher.listeners, l)
+		}
+		c.lock.Unlock()
+
+		if ok {
+			kvs := c.getCurrent(wkey)
+			for _, kv := range kvs {
+				l.OnAdd(kv)
+			}
+
+			return nil
 		}
 	}
 
-	return c.monitor(key, l)
+	return c.monitor(wkey, l)
 }
 
-func (r *Registry) getCluster(endpoints []string) (c *cluster, exists bool) {
-	clusterKey := getClusterKey(endpoints)
-	r.lock.Lock()
-	defer r.lock.Unlock()
-	c, exists = r.clusters[clusterKey]
+func (r *Registry) Unmonitor(endpoints []string, key string, exactMatch bool, l UpdateListener) {
+	c, exists := r.getCluster(endpoints)
 	if !exists {
-		c = newCluster(endpoints)
-		r.clusters[clusterKey] = c
+		return
+	}
+
+	wkey := watchKey{
+		key:        key,
+		exactMatch: exactMatch,
+	}
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	watcher, ok := c.watchers[wkey]
+	if !ok {
+		return
+	}
+
+	for i, listener := range watcher.listeners {
+		if listener == l {
+			watcher.listeners = append(watcher.listeners[:i], watcher.listeners[i+1:]...)
+			break
+		}
+	}
+
+	if len(watcher.listeners) == 0 {
+		if watcher.cancel != nil {
+			watcher.cancel()
+		}
+		delete(c.watchers, wkey)
+	}
+}
+
+func (r *Registry) getCluster(endpoints []string) (*cluster, bool) {
+	clusterKey := getClusterKey(endpoints)
+
+	r.lock.RLock()
+	c, ok := r.clusters[clusterKey]
+	r.lock.RUnlock()
+
+	return c, ok
+}
+
+func (r *Registry) getOrCreateCluster(endpoints []string) (c *cluster, exists bool) {
+	c, exists = r.getCluster(endpoints)
+	if !exists {
+		clusterKey := getClusterKey(endpoints)
+
+		r.lock.Lock()
+		defer r.lock.Unlock()
+
+		// double-check locking
+		c, exists = r.clusters[clusterKey]
+		if !exists {
+			c = newCluster(endpoints)
+			r.clusters[clusterKey] = c
+		}
 	}
 
 	return
 }
 
-type cluster struct {
-	endpoints  []string
-	key        string
-	values     map[string]map[string]string
-	listeners  map[string][]UpdateListener
-	watchGroup *threading.RoutineGroup
-	done       chan lang.PlaceholderType
-	lock       sync.Mutex
-}
+type (
+	watchKey struct {
+		key        string
+		exactMatch bool
+	}
+
+	watchValue struct {
+		listeners []UpdateListener
+		values    map[string]string
+		cancel    context.CancelFunc
+	}
+
+	cluster struct {
+		endpoints  []string
+		key        string
+		watchers   map[watchKey]*watchValue
+		watchGroup *threading.RoutineGroup
+		done       chan lang.PlaceholderType
+		lock       sync.RWMutex
+	}
+)
 
 func newCluster(endpoints []string) *cluster {
 	return &cluster{
 		endpoints:  endpoints,
 		key:        getClusterKey(endpoints),
-		values:     make(map[string]map[string]string),
-		listeners:  make(map[string][]UpdateListener),
+		watchers:   make(map[watchKey]*watchValue),
 		watchGroup: threading.NewRoutineGroup(),
 		done:       make(chan lang.PlaceholderType),
 	}
 }
 
-func (c *cluster) context(cli EtcdClient) context.Context {
-	return contextx.ValueOnlyFrom(cli.Ctx())
+func (c *cluster) addListener(key watchKey, l UpdateListener) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	watcher, ok := c.watchers[key]
+	if ok {
+		watcher.listeners = append(watcher.listeners, l)
+		return
+	}
+
+	val := newWatchValue()
+	val.listeners = []UpdateListener{l}
+	c.watchers[key] = val
 }
 
 func (c *cluster) getClient() (EtcdClient, error) {
@@ -107,12 +198,17 @@ func (c *cluster) getClient() (EtcdClient, error) {
 	return val.(EtcdClient), nil
 }
 
-func (c *cluster) getCurrent(key string) []KV {
-	c.lock.Lock()
-	defer c.lock.Unlock()
+func (c *cluster) getCurrent(key watchKey) []KV {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
 
-	var kvs []KV
-	for k, v := range c.values[key] {
+	watcher, ok := c.watchers[key]
+	if !ok {
+		return nil
+	}
+
+	kvs := make([]KV, 0, len(watcher.values))
+	for k, v := range watcher.values {
 		kvs = append(kvs, KV{
 			Key: k,
 			Val: v,
@@ -122,42 +218,23 @@ func (c *cluster) getCurrent(key string) []KV {
 	return kvs
 }
 
-func (c *cluster) handleChanges(key string, kvs []KV) {
-	var add []KV
-	var remove []KV
+func (c *cluster) handleChanges(key watchKey, kvs []KV) {
 	c.lock.Lock()
-	listeners := append([]UpdateListener(nil), c.listeners[key]...)
-	vals, ok := c.values[key]
+	watcher, ok := c.watchers[key]
 	if !ok {
-		add = kvs
-		vals = make(map[string]string)
-		for _, kv := range kvs {
-			vals[kv.Key] = kv.Val
-		}
-		c.values[key] = vals
-	} else {
-		m := make(map[string]string)
-		for _, kv := range kvs {
-			m[kv.Key] = kv.Val
-		}
-		for k, v := range vals {
-			if val, ok := m[k]; !ok || v != val {
-				remove = append(remove, KV{
-					Key: k,
-					Val: v,
-				})
-			}
-		}
-		for k, v := range m {
-			if val, ok := vals[k]; !ok || v != val {
-				add = append(add, KV{
-					Key: k,
-					Val: v,
-				})
-			}
-		}
-		c.values[key] = m
+		c.lock.Unlock()
+		return
 	}
+
+	listeners := append([]UpdateListener(nil), watcher.listeners...)
+	// watcher.values cannot be nil
+	vals := watcher.values
+	newVals := make(map[string]string, len(kvs)+len(vals))
+	for _, kv := range kvs {
+		newVals[kv.Key] = kv.Val
+	}
+	add, remove := calculateChanges(vals, newVals)
+	watcher.values = newVals
 	c.lock.Unlock()
 
 	for _, kv := range add {
@@ -172,20 +249,22 @@ func (c *cluster) handleChanges(key string, kvs []KV) {
 	}
 }
 
-func (c *cluster) handleWatchEvents(key string, events []*clientv3.Event) {
-	c.lock.Lock()
-	listeners := append([]UpdateListener(nil), c.listeners[key]...)
-	c.lock.Unlock()
+func (c *cluster) handleWatchEvents(ctx context.Context, key watchKey, events []*clientv3.Event) {
+	c.lock.RLock()
+	watcher, ok := c.watchers[key]
+	if !ok {
+		c.lock.RUnlock()
+		return
+	}
+
+	listeners := append([]UpdateListener(nil), watcher.listeners...)
+	c.lock.RUnlock()
 
 	for _, ev := range events {
 		switch ev.Type {
 		case clientv3.EventTypePut:
 			c.lock.Lock()
-			if vals, ok := c.values[key]; ok {
-				vals[string(ev.Kv.Key)] = string(ev.Kv.Value)
-			} else {
-				c.values[key] = map[string]string{string(ev.Kv.Key): string(ev.Kv.Value)}
-			}
+			watcher.values[string(ev.Kv.Key)] = string(ev.Kv.Value)
 			c.lock.Unlock()
 			for _, l := range listeners {
 				l.OnAdd(KV{
@@ -195,9 +274,7 @@ func (c *cluster) handleWatchEvents(key string, events []*clientv3.Event) {
 			}
 		case clientv3.EventTypeDelete:
 			c.lock.Lock()
-			if vals, ok := c.values[key]; ok {
-				delete(vals, string(ev.Kv.Key))
-			}
+			delete(watcher.values, string(ev.Kv.Key))
 			c.lock.Unlock()
 			for _, l := range listeners {
 				l.OnDelete(KV{
@@ -206,27 +283,32 @@ func (c *cluster) handleWatchEvents(key string, events []*clientv3.Event) {
 				})
 			}
 		default:
-			logx.Errorf("Unknown event type: %v", ev.Type)
+			logc.Errorf(ctx, "Unknown event type: %v", ev.Type)
 		}
 	}
 }
 
-func (c *cluster) load(cli EtcdClient, key string) int64 {
+func (c *cluster) load(cli EtcdClient, key watchKey) int64 {
 	var resp *clientv3.GetResponse
 	for {
 		var err error
-		ctx, cancel := context.WithTimeout(c.context(cli), RequestTimeout)
-		resp, err = cli.Get(ctx, makeKeyPrefix(key), clientv3.WithPrefix())
+		ctx, cancel := context.WithTimeout(cli.Ctx(), RequestTimeout)
+		if key.exactMatch {
+			resp, err = cli.Get(ctx, key.key)
+		} else {
+			resp, err = cli.Get(ctx, makeKeyPrefix(key.key), clientv3.WithPrefix())
+		}
+
 		cancel()
 		if err == nil {
 			break
 		}
 
-		logx.Error(err)
-		time.Sleep(coolDownInterval)
+		logc.Errorf(cli.Ctx(), "%s, key: %s, exactMatch: %t", err.Error(), key.key, key.exactMatch)
+		time.Sleep(coolDownUnstable.AroundDuration(coolDownInterval))
 	}
 
-	var kvs []KV
+	kvs := make([]KV, 0, len(resp.Kvs))
 	for _, ev := range resp.Kvs {
 		kvs = append(kvs, KV{
 			Key: string(ev.Key),
@@ -239,16 +321,13 @@ func (c *cluster) load(cli EtcdClient, key string) int64 {
 	return resp.Header.Revision
 }
 
-func (c *cluster) monitor(key string, l UpdateListener) error {
-	c.lock.Lock()
-	c.listeners[key] = append(c.listeners[key], l)
-	c.lock.Unlock()
-
+func (c *cluster) monitor(key watchKey, l UpdateListener) error {
 	cli, err := c.getClient()
 	if err != nil {
 		return err
 	}
 
+	c.addListener(key, l)
 	rev := c.load(cli, key)
 	c.watchGroup.Run(func() {
 		c.watch(cli, key, rev)
@@ -270,16 +349,22 @@ func (c *cluster) newClient() (EtcdClient, error) {
 
 func (c *cluster) reload(cli EtcdClient) {
 	c.lock.Lock()
+	// cancel the previous watches
 	close(c.done)
 	c.watchGroup.Wait()
+	keys := make([]watchKey, 0, len(c.watchers))
+	for wk, wval := range c.watchers {
+		keys = append(keys, wk)
+		if wval.cancel != nil {
+			wval.cancel()
+		}
+	}
+
 	c.done = make(chan lang.PlaceholderType)
 	c.watchGroup = threading.NewRoutineGroup()
-	var keys []string
-	for k := range c.listeners {
-		keys = append(keys, k)
-	}
 	c.lock.Unlock()
 
+	// start new watches
 	for _, key := range keys {
 		k := key
 		c.watchGroup.Run(func() {
@@ -289,7 +374,7 @@ func (c *cluster) reload(cli EtcdClient) {
 	}
 }
 
-func (c *cluster) watch(cli EtcdClient, key string, rev int64) {
+func (c *cluster) watch(cli EtcdClient, key watchKey, rev int64) {
 	for {
 		err := c.watchStream(cli, key, rev)
 		if err == nil {
@@ -297,24 +382,17 @@ func (c *cluster) watch(cli EtcdClient, key string, rev int64) {
 		}
 
 		if rev != 0 && errors.Is(err, rpctypes.ErrCompacted) {
-			logx.Errorf("etcd watch stream has been compacted, try to reload, rev %d", rev)
+			logc.Errorf(cli.Ctx(), "etcd watch stream has been compacted, try to reload, rev %d", rev)
 			rev = c.load(cli, key)
 		}
 
 		// log the error and retry
-		logx.Error(err)
+		logc.Error(cli.Ctx(), err)
 	}
 }
 
-func (c *cluster) watchStream(cli EtcdClient, key string, rev int64) error {
-	var rch clientv3.WatchChan
-	if rev != 0 {
-		rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key),
-			clientv3.WithPrefix(), clientv3.WithRev(rev+1))
-	} else {
-		rch = cli.Watch(clientv3.WithRequireLeader(c.context(cli)), makeKeyPrefix(key),
-			clientv3.WithPrefix())
-	}
+func (c *cluster) watchStream(cli EtcdClient, key watchKey, rev int64) error {
+	ctx, rch := c.setupWatch(cli, key, rev)
 
 	for {
 		select {
@@ -329,13 +407,47 @@ func (c *cluster) watchStream(cli EtcdClient, key string, rev int64) error {
 				return fmt.Errorf("etcd monitor chan error: %w", wresp.Err())
 			}
 
-			c.handleWatchEvents(key, wresp.Events)
+			c.handleWatchEvents(ctx, key, wresp.Events)
+		case <-ctx.Done():
+			return nil
 		case <-c.done:
 			return nil
 		}
 	}
 }
 
+func (c *cluster) setupWatch(cli EtcdClient, key watchKey, rev int64) (context.Context, clientv3.WatchChan) {
+	var (
+		rch  clientv3.WatchChan
+		ops  []clientv3.OpOption
+		wkey = key.key
+	)
+
+	if !key.exactMatch {
+		wkey = makeKeyPrefix(key.key)
+		ops = append(ops, clientv3.WithPrefix())
+	}
+	if rev != 0 {
+		ops = append(ops, clientv3.WithRev(rev+1))
+	}
+
+	ctx, cancel := context.WithCancel(cli.Ctx())
+	if watcher, ok := c.watchers[key]; ok {
+		watcher.cancel = cancel
+	} else {
+		val := newWatchValue()
+		val.cancel = cancel
+
+		c.lock.Lock()
+		c.watchers[key] = val
+		c.lock.Unlock()
+	}
+
+	rch = cli.Watch(clientv3.WithRequireLeader(ctx), wkey, ops...)
+
+	return ctx, rch
+}
+
 func (c *cluster) watchConnState(cli EtcdClient) {
 	watcher := newStateWatcher()
 	watcher.addListener(func() {
@@ -364,6 +476,28 @@ func DialClient(endpoints []string) (EtcdClient, error) {
 	return clientv3.New(cfg)
 }
 
+func calculateChanges(oldVals, newVals map[string]string) (add, remove []KV) {
+	for k, v := range newVals {
+		if val, ok := oldVals[k]; !ok || v != val {
+			add = append(add, KV{
+				Key: k,
+				Val: v,
+			})
+		}
+	}
+
+	for k, v := range oldVals {
+		if val, ok := newVals[k]; !ok || v != val {
+			remove = append(remove, KV{
+				Key: k,
+				Val: v,
+			})
+		}
+	}
+
+	return add, remove
+}
+
 func getClusterKey(endpoints []string) string {
 	sort.Strings(endpoints)
 	return strings.Join(endpoints, endpointsSeparator)
@@ -372,3 +506,10 @@ func getClusterKey(endpoints []string) string {
 func makeKeyPrefix(key string) string {
 	return fmt.Sprintf("%s%c", key, Delimiter)
 }
+
+// NewClient returns a watchValue that make sure values are not nil.
+func newWatchValue() *watchValue {
+	return &watchValue{
+		values: make(map[string]string),
+	}
+}

core/discov/internal/registry_test.go

@@ -7,16 +7,17 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/contextx"
 	"github.com/zeromicro/go-zero/core/lang"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stringx"
+	"github.com/zeromicro/go-zero/core/threading"
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/client/v3/mock/mockserver"
+	"go.uber.org/mock/gomock"
 )
 
 var mockLock sync.Mutex
@@ -38,9 +39,9 @@ func setMockClient(cli EtcdClient) func() {
 
 func TestGetCluster(t *testing.T) {
 	AddAccount([]string{"first"}, "foo", "bar")
-	c1, _ := GetRegistry().getCluster([]string{"first"})
-	c2, _ := GetRegistry().getCluster([]string{"second"})
-	c3, _ := GetRegistry().getCluster([]string{"first"})
+	c1, _ := GetRegistry().getOrCreateCluster([]string{"first"})
+	c2, _ := GetRegistry().getOrCreateCluster([]string{"second"})
+	c3, _ := GetRegistry().getOrCreateCluster([]string{"first"})
 	assert.Equal(t, c1, c3)
 	assert.NotEqual(t, c1, c2)
 }
@@ -50,6 +51,36 @@ func TestGetClusterKey(t *testing.T) {
 		getClusterKey([]string{"remotehost:5678", "localhost:1234"}))
 }
 
+func TestUnmonitor(t *testing.T) {
+	t.Run("no listener", func(t *testing.T) {
+		reg := &Registry{
+			clusters: map[string]*cluster{},
+		}
+		assert.NotPanics(t, func() {
+			reg.Unmonitor([]string{"any"}, "any", false, nil)
+		})
+	})
+
+	t.Run("no value", func(t *testing.T) {
+		reg := &Registry{
+			clusters: map[string]*cluster{
+				"any": {
+					watchers: map[watchKey]*watchValue{
+						{
+							key: "any",
+						}: {
+							values: map[string]string{},
+						},
+					},
+				},
+			},
+		}
+		assert.NotPanics(t, func() {
+			reg.Unmonitor([]string{"any"}, "another", false, nil)
+		})
+	})
+}
+
 func TestCluster_HandleChanges(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	l := NewMockUpdateListener(ctrl)
@@ -78,8 +109,14 @@ func TestCluster_HandleChanges(t *testing.T) {
 		Val: "4",
 	})
 	c := newCluster([]string{"any"})
-	c.listeners["any"] = []UpdateListener{l}
-	c.handleChanges("any", []KV{
+	key := watchKey{
+		key:        "any",
+		exactMatch: false,
+	}
+	c.watchers[key] = &watchValue{
+		listeners: []UpdateListener{l},
+	}
+	c.handleChanges(key, []KV{
 		{
 			Key: "first",
 			Val: "1",
@@ -92,8 +129,8 @@ func TestCluster_HandleChanges(t *testing.T) {
 	assert.EqualValues(t, map[string]string{
 		"first":  "1",
 		"second": "2",
-	}, c.values["any"])
-	c.handleChanges("any", []KV{
+	}, c.watchers[key].values)
+	c.handleChanges(key, []KV{
 		{
 			Key: "third",
 			Val: "3",
@@ -106,7 +143,7 @@ func TestCluster_HandleChanges(t *testing.T) {
 	assert.EqualValues(t, map[string]string{
 		"third":  "3",
 		"fourth": "4",
-	}, c.values["any"])
+	}, c.watchers[key].values)
 }
 
 func TestCluster_Load(t *testing.T) {
@@ -126,9 +163,11 @@ func TestCluster_Load(t *testing.T) {
 	}, nil)
 	cli.EXPECT().Ctx().Return(context.Background())
 	c := &cluster{
-		values: make(map[string]map[string]string),
+		watchers: make(map[watchKey]*watchValue),
 	}
-	c.load(cli, "any")
+	c.load(cli, watchKey{
+		key: "any",
+	})
 }
 
 func TestCluster_Watch(t *testing.T) {
@@ -160,11 +199,16 @@ func TestCluster_Watch(t *testing.T) {
 			var wg sync.WaitGroup
 			wg.Add(1)
 			c := &cluster{
-				listeners: make(map[string][]UpdateListener),
-				values:    make(map[string]map[string]string),
+				watchers: make(map[watchKey]*watchValue),
+			}
+			key := watchKey{
+				key: "any",
 			}
 			listener := NewMockUpdateListener(ctrl)
-			c.listeners["any"] = []UpdateListener{listener}
+			c.watchers[key] = &watchValue{
+				listeners: []UpdateListener{listener},
+				values:    make(map[string]string),
+			}
 			listener.EXPECT().OnAdd(gomock.Any()).Do(func(kv KV) {
 				assert.Equal(t, "hello", kv.Key)
 				assert.Equal(t, "world", kv.Val)
@@ -173,7 +217,7 @@ func TestCluster_Watch(t *testing.T) {
 			listener.EXPECT().OnDelete(gomock.Any()).Do(func(_ any) {
 				wg.Done()
 			}).MaxTimes(1)
-			go c.watch(cli, "any", 0)
+			go c.watch(cli, key, 0)
 			ch <- clientv3.WatchResponse{
 				Events: []*clientv3.Event{
 					{
@@ -211,17 +255,111 @@ func TestClusterWatch_RespFailures(t *testing.T) {
 			ch := make(chan clientv3.WatchResponse)
 			cli.EXPECT().Watch(gomock.Any(), "any/", gomock.Any()).Return(ch).AnyTimes()
 			cli.EXPECT().Ctx().Return(context.Background()).AnyTimes()
-			c := new(cluster)
+			c := &cluster{
+				watchers: make(map[watchKey]*watchValue),
+			}
 			c.done = make(chan lang.PlaceholderType)
 			go func() {
 				ch <- resp
 				close(c.done)
 			}()
-			c.watch(cli, "any", 0)
+			key := watchKey{
+				key: "any",
+			}
+			c.watch(cli, key, 0)
 		})
 	}
 }
 
+func TestCluster_getCurrent(t *testing.T) {
+	t.Run("no value", func(t *testing.T) {
+		c := &cluster{
+			watchers: map[watchKey]*watchValue{
+				{
+					key: "any",
+				}: {
+					values: map[string]string{},
+				},
+			},
+		}
+		assert.Nil(t, c.getCurrent(watchKey{
+			key: "another",
+		}))
+	})
+}
+
+func TestCluster_handleWatchEvents(t *testing.T) {
+	t.Run("no value", func(t *testing.T) {
+		c := &cluster{
+			watchers: map[watchKey]*watchValue{
+				{
+					key: "any",
+				}: {
+					values: map[string]string{},
+				},
+			},
+		}
+		assert.NotPanics(t, func() {
+			c.handleWatchEvents(context.Background(), watchKey{
+				key: "another",
+			}, nil)
+		})
+	})
+}
+
+func TestCluster_addListener(t *testing.T) {
+	t.Run("has listener", func(t *testing.T) {
+		c := &cluster{
+			watchers: map[watchKey]*watchValue{
+				{
+					key: "any",
+				}: {
+					listeners: make([]UpdateListener, 0),
+				},
+			},
+		}
+		assert.NotPanics(t, func() {
+			c.addListener(watchKey{
+				key: "any",
+			}, nil)
+		})
+	})
+
+	t.Run("no listener", func(t *testing.T) {
+		c := &cluster{
+			watchers: map[watchKey]*watchValue{
+				{
+					key: "any",
+				}: {
+					listeners: make([]UpdateListener, 0),
+				},
+			},
+		}
+		assert.NotPanics(t, func() {
+			c.addListener(watchKey{
+				key: "another",
+			}, nil)
+		})
+	})
+}
+
+func TestCluster_reload(t *testing.T) {
+	c := &cluster{
+		watchers:   map[watchKey]*watchValue{},
+		watchGroup: threading.NewRoutineGroup(),
+		done:       make(chan lang.PlaceholderType),
+	}
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	cli := NewMockEtcdClient(ctrl)
+	restore := setMockClient(cli)
+	defer restore()
+	assert.NotPanics(t, func() {
+		c.reload(cli)
+	})
+}
+
 func TestClusterWatch_CloseChan(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
@@ -231,13 +369,17 @@ func TestClusterWatch_CloseChan(t *testing.T) {
 	ch := make(chan clientv3.WatchResponse)
 	cli.EXPECT().Watch(gomock.Any(), "any/", gomock.Any()).Return(ch).AnyTimes()
 	cli.EXPECT().Ctx().Return(context.Background()).AnyTimes()
-	c := new(cluster)
+	c := &cluster{
+		watchers: make(map[watchKey]*watchValue),
+	}
 	c.done = make(chan lang.PlaceholderType)
 	go func() {
 		close(ch)
 		close(c.done)
 	}()
-	c.watch(cli, "any", 0)
+	c.watch(cli, watchKey{
+		key: "any",
+	}, 0)
 }
 
 func TestValueOnlyContext(t *testing.T) {
@@ -280,16 +422,59 @@ func TestRegistry_Monitor(t *testing.T) {
 	GetRegistry().lock.Lock()
 	GetRegistry().clusters = map[string]*cluster{
 		getClusterKey(endpoints): {
-			listeners: map[string][]UpdateListener{},
-			values: map[string]map[string]string{
-				"foo": {
-					"bar": "baz",
+			watchers: map[watchKey]*watchValue{
+				{
+					key:        "foo",
+					exactMatch: true,
+				}: {
+					values: map[string]string{
+						"bar": "baz",
+					},
 				},
 			},
 		},
 	}
 	GetRegistry().lock.Unlock()
-	assert.Error(t, GetRegistry().Monitor(endpoints, "foo", new(mockListener)))
+	assert.Error(t, GetRegistry().Monitor(endpoints, "foo", false, new(mockListener)))
+}
+
+func TestRegistry_Unmonitor(t *testing.T) {
+	svr, err := mockserver.StartMockServers(1)
+	assert.NoError(t, err)
+	svr.StartAt(0)
+
+	_, cancel := context.WithCancel(context.Background())
+	endpoints := []string{svr.Servers[0].Address}
+	GetRegistry().lock.Lock()
+	GetRegistry().clusters = map[string]*cluster{
+		getClusterKey(endpoints): {
+			watchers: map[watchKey]*watchValue{
+				{
+					key:        "foo",
+					exactMatch: true,
+				}: {
+					values: map[string]string{
+						"bar": "baz",
+					},
+					cancel: cancel,
+				},
+			},
+		},
+	}
+	GetRegistry().lock.Unlock()
+	l := new(mockListener)
+	assert.NoError(t, GetRegistry().Monitor(endpoints, "foo", true, l))
+	watchVals := GetRegistry().clusters[getClusterKey(endpoints)].watchers[watchKey{
+		key:        "foo",
+		exactMatch: true,
+	}]
+	assert.Equal(t, 1, len(watchVals.listeners))
+	GetRegistry().Unmonitor(endpoints, "foo", true, l)
+	watchVals = GetRegistry().clusters[getClusterKey(endpoints)].watchers[watchKey{
+		key:        "foo",
+		exactMatch: true,
+	}]
+	assert.Nil(t, watchVals)
 }
 
 type mockListener struct {

core/discov/internal/statewatcher_mock.go

@@ -1,5 +1,10 @@
 // Code generated by MockGen. DO NOT EDIT.
 // Source: statewatcher.go
+//
+// Generated by this command:
+//
+//	mockgen -package internal -destination statewatcher_mock.go -source statewatcher.go etcdConn
+//
 
 // Package internal is a generated GoMock package.
 package internal
@@ -8,34 +13,35 @@ import (
 	context "context"
 	reflect "reflect"
 
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 	connectivity "google.golang.org/grpc/connectivity"
 )
 
-// MocketcdConn is a mock of etcdConn interface
+// MocketcdConn is a mock of etcdConn interface.
 type MocketcdConn struct {
 	ctrl     *gomock.Controller
 	recorder *MocketcdConnMockRecorder
+	isgomock struct{}
 }
 
-// MocketcdConnMockRecorder is the mock recorder for MocketcdConn
+// MocketcdConnMockRecorder is the mock recorder for MocketcdConn.
 type MocketcdConnMockRecorder struct {
 	mock *MocketcdConn
 }
 
-// NewMocketcdConn creates a new mock instance
+// NewMocketcdConn creates a new mock instance.
 func NewMocketcdConn(ctrl *gomock.Controller) *MocketcdConn {
 	mock := &MocketcdConn{ctrl: ctrl}
 	mock.recorder = &MocketcdConnMockRecorder{mock}
 	return mock
 }
 
-// EXPECT returns an object that allows the caller to indicate expected use
+// EXPECT returns an object that allows the caller to indicate expected use.
 func (m *MocketcdConn) EXPECT() *MocketcdConnMockRecorder {
 	return m.recorder
 }
 
-// GetState mocks base method
+// GetState mocks base method.
 func (m *MocketcdConn) GetState() connectivity.State {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetState")
@@ -43,13 +49,13 @@ func (m *MocketcdConn) GetState() connectivity.State {
 	return ret0
 }
 
-// GetState indicates an expected call of GetState
+// GetState indicates an expected call of GetState.
 func (mr *MocketcdConnMockRecorder) GetState() *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetState", reflect.TypeOf((*MocketcdConn)(nil).GetState))
 }
 
-// WaitForStateChange mocks base method
+// WaitForStateChange mocks base method.
 func (m *MocketcdConn) WaitForStateChange(ctx context.Context, sourceState connectivity.State) bool {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "WaitForStateChange", ctx, sourceState)
@@ -57,7 +63,7 @@ func (m *MocketcdConn) WaitForStateChange(ctx context.Context, sourceState conne
 	return ret0
 }
 
-// WaitForStateChange indicates an expected call of WaitForStateChange
+// WaitForStateChange indicates an expected call of WaitForStateChange.
 func (mr *MocketcdConnMockRecorder) WaitForStateChange(ctx, sourceState any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WaitForStateChange", reflect.TypeOf((*MocketcdConn)(nil).WaitForStateChange), ctx, sourceState)

core/discov/internal/statewatcher_test.go

@@ -4,7 +4,7 @@ import (
 	"sync"
 	"testing"
 
-	"github.com/golang/mock/gomock"
+	"go.uber.org/mock/gomock"
 	"google.golang.org/grpc/connectivity"
 )
 

core/discov/internal/updatelistener.go

@@ -10,6 +10,7 @@ type (
 	}
 
 	// UpdateListener wraps the OnAdd and OnDelete methods.
+	// The implementation should be thread-safe and idempotent.
 	UpdateListener interface {
 		OnAdd(kv KV)
 		OnDelete(kv KV)

core/discov/internal/updatelistener_mock.go

@@ -1,5 +1,10 @@
 // Code generated by MockGen. DO NOT EDIT.
 // Source: updatelistener.go
+//
+// Generated by this command:
+//
+//	mockgen -package internal -destination updatelistener_mock.go -source updatelistener.go UpdateListener
+//
 
 // Package internal is a generated GoMock package.
 package internal
@@ -7,51 +12,52 @@ package internal
 import (
 	reflect "reflect"
 
-	gomock "github.com/golang/mock/gomock"
+	gomock "go.uber.org/mock/gomock"
 )
 
-// MockUpdateListener is a mock of UpdateListener interface
+// MockUpdateListener is a mock of UpdateListener interface.
 type MockUpdateListener struct {
 	ctrl     *gomock.Controller
 	recorder *MockUpdateListenerMockRecorder
+	isgomock struct{}
 }
 
-// MockUpdateListenerMockRecorder is the mock recorder for MockUpdateListener
+// MockUpdateListenerMockRecorder is the mock recorder for MockUpdateListener.
 type MockUpdateListenerMockRecorder struct {
 	mock *MockUpdateListener
 }
 
-// NewMockUpdateListener creates a new mock instance
+// NewMockUpdateListener creates a new mock instance.
 func NewMockUpdateListener(ctrl *gomock.Controller) *MockUpdateListener {
 	mock := &MockUpdateListener{ctrl: ctrl}
 	mock.recorder = &MockUpdateListenerMockRecorder{mock}
 	return mock
 }
 
-// EXPECT returns an object that allows the caller to indicate expected use
+// EXPECT returns an object that allows the caller to indicate expected use.
 func (m *MockUpdateListener) EXPECT() *MockUpdateListenerMockRecorder {
 	return m.recorder
 }
 
-// OnAdd mocks base method
+// OnAdd mocks base method.
 func (m *MockUpdateListener) OnAdd(kv KV) {
 	m.ctrl.T.Helper()
 	m.ctrl.Call(m, "OnAdd", kv)
 }
 
-// OnAdd indicates an expected call of OnAdd
+// OnAdd indicates an expected call of OnAdd.
 func (mr *MockUpdateListenerMockRecorder) OnAdd(kv any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnAdd", reflect.TypeOf((*MockUpdateListener)(nil).OnAdd), kv)
 }
 
-// OnDelete mocks base method
+// OnDelete mocks base method.
 func (m *MockUpdateListener) OnDelete(kv KV) {
 	m.ctrl.T.Helper()
 	m.ctrl.Call(m, "OnDelete", kv)
 }
 
-// OnDelete indicates an expected call of OnDelete
+// OnDelete indicates an expected call of OnDelete.
 func (mr *MockUpdateListenerMockRecorder) OnDelete(kv any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnDelete", reflect.TypeOf((*MockUpdateListener)(nil).OnDelete), kv)

core/discov/publisher.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/zeromicro/go-zero/core/discov/internal"
 	"github.com/zeromicro/go-zero/core/lang"
+	"github.com/zeromicro/go-zero/core/logc"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/proc"
 	"github.com/zeromicro/go-zero/core/syncx"
@@ -91,12 +92,12 @@ func (p *Publisher) doKeepAlive() error {
 		default:
 			cli, err := p.doRegister()
 			if err != nil {
-				logx.Errorf("etcd publisher doRegister: %s", err.Error())
+				logc.Errorf(cli.Ctx(), "etcd publisher doRegister: %v", err)
 				break
 			}
 
 			if err := p.keepAliveAsync(cli); err != nil {
-				logx.Errorf("etcd publisher keepAliveAsync: %s", err.Error())
+				logc.Errorf(cli.Ctx(), "etcd publisher keepAliveAsync: %v", err)
 				break
 			}
 
@@ -124,23 +125,48 @@ func (p *Publisher) keepAliveAsync(cli internal.EtcdClient) error {
 	}
 
 	threading.GoSafe(func() {
+		wch := cli.Watch(cli.Ctx(), p.fullKey, clientv3.WithFilterPut())
+
 		for {
 			select {
 			case _, ok := <-ch:
 				if !ok {
 					p.revoke(cli)
 					if err := p.doKeepAlive(); err != nil {
-						logx.Errorf("etcd publisher KeepAlive: %s", err.Error())
+						logc.Errorf(cli.Ctx(), "etcd publisher KeepAlive: %v", err)
 					}
 					return
 				}
+
+			case c := <-wch:
+				if c.Err() != nil {
+					logc.Errorf(cli.Ctx(), "etcd publisher watch: %v", c.Err())
+					if err := p.doKeepAlive(); err != nil {
+						logc.Errorf(cli.Ctx(), "etcd publisher KeepAlive: %v", err)
+					}
+					return
+				}
+
+				for _, evt := range c.Events {
+					if evt.Type == clientv3.EventTypeDelete {
+						logc.Infof(cli.Ctx(), "etcd publisher watch: %s, event: %v",
+							evt.Kv.Key, evt.Type)
+						_, err := cli.Put(cli.Ctx(), p.fullKey, p.value, clientv3.WithLease(p.lease))
+						if err != nil {
+							logc.Errorf(cli.Ctx(), "etcd publisher re-put key: %v", err)
+						} else {
+							logc.Infof(cli.Ctx(), "etcd publisher re-put key: %s, value: %s",
+								p.fullKey, p.value)
+						}
+					}
+				}
 			case <-p.pauseChan:
-				logx.Infof("paused etcd renew, key: %s, value: %s", p.key, p.value)
+				logc.Infof(cli.Ctx(), "paused etcd renew, key: %s, value: %s", p.key, p.value)
 				p.revoke(cli)
 				select {
 				case <-p.resumeChan:
 					if err := p.doKeepAlive(); err != nil {
-						logx.Errorf("etcd publisher KeepAlive: %s", err.Error())
+						logc.Errorf(cli.Ctx(), "etcd publisher KeepAlive: %v", err)
 					}
 					return
 				case <-p.quit.Done():
@@ -175,7 +201,7 @@ func (p *Publisher) register(client internal.EtcdClient) (clientv3.LeaseID, erro
 
 func (p *Publisher) revoke(cli internal.EtcdClient) {
 	if _, err := cli.Revoke(cli.Ctx(), p.lease); err != nil {
-		logx.Errorf("etcd publisher revoke: %s", err.Error())
+		logc.Errorf(cli.Ctx(), "etcd publisher revoke: %v", err)
 	}
 }
 

core/discov/publisher_test.go

@@ -9,13 +9,14 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	"github.com/zeromicro/go-zero/core/discov/internal"
 	"github.com/zeromicro/go-zero/core/lang"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stringx"
+	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.uber.org/mock/gomock"
 	"golang.org/x/net/http2"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -211,6 +212,9 @@ func TestPublisher_keepAliveAsyncQuit(t *testing.T) {
 	defer restore()
 	cli.EXPECT().Ctx().AnyTimes()
 	cli.EXPECT().KeepAlive(gomock.Any(), id)
+	// Add Watch expectation for the new watch mechanism
+	watchChan := make(<-chan clientv3.WatchResponse)
+	cli.EXPECT().Watch(gomock.Any(), gomock.Any(), gomock.Any()).Return(watchChan)
 	var wg sync.WaitGroup
 	wg.Add(1)
 	cli.EXPECT().Revoke(gomock.Any(), id).Do(func(_, _ any) {
@@ -232,6 +236,9 @@ func TestPublisher_keepAliveAsyncPause(t *testing.T) {
 	defer restore()
 	cli.EXPECT().Ctx().AnyTimes()
 	cli.EXPECT().KeepAlive(gomock.Any(), id)
+	// Add Watch expectation for the new watch mechanism
+	watchChan := make(<-chan clientv3.WatchResponse)
+	cli.EXPECT().Watch(gomock.Any(), gomock.Any(), gomock.Any()).Return(watchChan)
 	pub := NewPublisher(nil, "thekey", "thevalue")
 	var wg sync.WaitGroup
 	wg.Add(1)
@@ -245,6 +252,112 @@ func TestPublisher_keepAliveAsyncPause(t *testing.T) {
 	wg.Wait()
 }
 
+// Test case for key deletion and re-registration (covers lines 148-155)
+func TestPublisher_keepAliveAsyncKeyDeletion(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	const id clientv3.LeaseID = 1
+	cli := internal.NewMockEtcdClient(ctrl)
+	restore := setMockClient(cli)
+	defer restore()
+	cli.EXPECT().Ctx().AnyTimes()
+	cli.EXPECT().KeepAlive(gomock.Any(), id)
+
+	// Create a watch channel that will send a delete event
+	watchChan := make(chan clientv3.WatchResponse, 1)
+	watchResp := clientv3.WatchResponse{
+		Events: []*clientv3.Event{{
+			Type: clientv3.EventTypeDelete,
+			Kv: &mvccpb.KeyValue{
+				Key: []byte("thekey"),
+			},
+		}},
+	}
+	watchChan <- watchResp
+
+	cli.EXPECT().Watch(gomock.Any(), gomock.Any(), gomock.Any()).Return((<-chan clientv3.WatchResponse)(watchChan))
+
+	var wg sync.WaitGroup
+	wg.Add(1) // Only wait for Revoke call
+
+	// Use a channel to signal when Put has been called
+	putCalled := make(chan struct{})
+
+	// Expect the re-put operation when key is deleted
+	cli.EXPECT().Put(gomock.Any(), "thekey", "thevalue", gomock.Any()).Do(func(_, _, _, _ any) {
+		close(putCalled) // Signal that Put has been called
+	}).Return(nil, nil)
+
+	// Expect revoke when Stop is called
+	cli.EXPECT().Revoke(gomock.Any(), id).Do(func(_, _ any) {
+		wg.Done()
+	})
+
+	pub := NewPublisher(nil, "thekey", "thevalue")
+	pub.lease = id
+	pub.fullKey = "thekey"
+
+	assert.Nil(t, pub.keepAliveAsync(cli))
+
+	// Wait for Put to be called, then stop
+	<-putCalled
+	pub.Stop()
+	wg.Wait()
+}
+
+// Test case for key deletion with re-put error (covers error branch in lines 151-152)
+func TestPublisher_keepAliveAsyncKeyDeletionPutError(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	const id clientv3.LeaseID = 1
+	cli := internal.NewMockEtcdClient(ctrl)
+	restore := setMockClient(cli)
+	defer restore()
+	cli.EXPECT().Ctx().AnyTimes()
+	cli.EXPECT().KeepAlive(gomock.Any(), id)
+
+	// Create a watch channel that will send a delete event
+	watchChan := make(chan clientv3.WatchResponse, 1)
+	watchResp := clientv3.WatchResponse{
+		Events: []*clientv3.Event{{
+			Type: clientv3.EventTypeDelete,
+			Kv: &mvccpb.KeyValue{
+				Key: []byte("thekey"),
+			},
+		}},
+	}
+	watchChan <- watchResp
+
+	cli.EXPECT().Watch(gomock.Any(), gomock.Any(), gomock.Any()).Return((<-chan clientv3.WatchResponse)(watchChan))
+
+	var wg sync.WaitGroup
+	wg.Add(1) // Only wait for Revoke call
+
+	// Use a channel to signal when Put has been called
+	putCalled := make(chan struct{})
+
+	// Expect the re-put operation to fail
+	cli.EXPECT().Put(gomock.Any(), "thekey", "thevalue", gomock.Any()).Do(func(_, _, _, _ any) {
+		close(putCalled) // Signal that Put has been called
+	}).Return(nil, errors.New("put error"))
+
+	// Expect revoke when Stop is called
+	cli.EXPECT().Revoke(gomock.Any(), id).Do(func(_, _ any) {
+		wg.Done()
+	})
+
+	pub := NewPublisher(nil, "thekey", "thevalue")
+	pub.lease = id
+	pub.fullKey = "thekey"
+
+	assert.Nil(t, pub.keepAliveAsync(cli))
+
+	// Wait for Put to be called, then stop
+	<-putCalled
+	pub.Stop()
+	wg.Wait()
+}
+
 func TestPublisher_Resume(t *testing.T) {
 	publisher := new(Publisher)
 	publisher.resumeChan = make(chan lang.PlaceholderType)
@@ -273,6 +386,9 @@ func TestPublisher_keepAliveAsync(t *testing.T) {
 	defer restore()
 	cli.EXPECT().Ctx().AnyTimes()
 	cli.EXPECT().KeepAlive(gomock.Any(), id)
+	// Add Watch expectation for the new watch mechanism
+	watchChan := make(<-chan clientv3.WatchResponse)
+	cli.EXPECT().Watch(gomock.Any(), gomock.Any(), gomock.Any()).Return(watchChan)
 	cli.EXPECT().Grant(gomock.Any(), timeToLive).Return(&clientv3.LeaseGrantResponse{
 		ID: 1,
 	}, nil)

core/discov/subscriber.go

@@ -15,9 +15,11 @@ type (
 
 	// A Subscriber is used to subscribe the given key on an etcd cluster.
 	Subscriber struct {
-		endpoints []string
-		exclusive bool
-		items     *container
+		endpoints  []string
+		exclusive  bool
+		key        string
+		exactMatch bool
+		items      *container
 	}
 )
 
@@ -28,13 +30,14 @@ type (
 func NewSubscriber(endpoints []string, key string, opts ...SubOption) (*Subscriber, error) {
 	sub := &Subscriber{
 		endpoints: endpoints,
+		key:       key,
 	}
 	for _, opt := range opts {
 		opt(sub)
 	}
 	sub.items = newContainer(sub.exclusive)
 
-	if err := internal.GetRegistry().Monitor(endpoints, key, sub.items); err != nil {
+	if err := internal.GetRegistry().Monitor(endpoints, key, sub.exactMatch, sub.items); err != nil {
 		return nil, err
 	}
 
@@ -46,6 +49,11 @@ func (s *Subscriber) AddListener(listener func()) {
 	s.items.addListener(listener)
 }
 
+// Close closes the subscriber.
+func (s *Subscriber) Close() {
+	internal.GetRegistry().Unmonitor(s.endpoints, s.key, s.exactMatch, s.items)
+}
+
 // Values returns all the subscription values.
 func (s *Subscriber) Values() []string {
 	return s.items.getValues()
@@ -59,6 +67,13 @@ func Exclusive() SubOption {
 	}
 }
 
+// WithExactMatch turn off querying using key prefixes.
+func WithExactMatch() SubOption {
+	return func(sub *Subscriber) {
+		sub.exactMatch = true
+	}
+}
+
 // WithSubEtcdAccount provides the etcd username/password.
 func WithSubEtcdAccount(user, pass string) SubOption {
 	return func(sub *Subscriber) {

core/discov/subscriber_test.go

@@ -225,3 +225,28 @@ func TestWithSubEtcdAccount(t *testing.T) {
 	assert.Equal(t, user, account.User)
 	assert.Equal(t, "bar", account.Pass)
 }
+
+func TestWithExactMatch(t *testing.T) {
+	sub := new(Subscriber)
+	WithExactMatch()(sub)
+	sub.items = newContainer(sub.exclusive)
+	var count int32
+	sub.AddListener(func() {
+		atomic.AddInt32(&count, 1)
+	})
+	sub.items.notifyChange()
+	assert.Empty(t, sub.Values())
+	assert.Equal(t, int32(1), atomic.LoadInt32(&count))
+}
+
+func TestSubscriberClose(t *testing.T) {
+	l := newContainer(false)
+	sub := &Subscriber{
+		endpoints: []string{"localhost:12379"},
+		key:       "foo",
+		items:     l,
+	}
+	assert.NotPanics(t, func() {
+		sub.Close()
+	})
+}

core/errorx/batcherror.go

@@ -1,18 +1,21 @@
 package errorx
 
-import "bytes"
-
-type (
-	// A BatchError is an error that can hold multiple errors.
-	BatchError struct {
-		errs errorArray
-	}
-
-	errorArray []error
+import (
+	"errors"
+	"sync"
 )
 
-// Add adds errs to be, nil errors are ignored.
+// BatchError is an error that can hold multiple errors.
+type BatchError struct {
+	errs []error
+	lock sync.RWMutex
+}
+
+// Add adds one or more non-nil errors to the BatchError instance.
 func (be *BatchError) Add(errs ...error) {
+	be.lock.Lock()
+	defer be.lock.Unlock()
+
 	for _, err := range errs {
 		if err != nil {
 			be.errs = append(be.errs, err)
@@ -20,33 +23,20 @@ func (be *BatchError) Add(errs ...error) {
 	}
 }
 
-// Err returns an error that represents all errors.
+// Err returns an error that represents all accumulated errors.
+// It returns nil if there are no errors.
 func (be *BatchError) Err() error {
-	switch len(be.errs) {
-	case 0:
-		return nil
-	case 1:
-		return be.errs[0]
-	default:
-		return be.errs
-	}
+	be.lock.RLock()
+	defer be.lock.RUnlock()
+
+	// If there are no non-nil errors, errors.Join(...) returns nil.
+	return errors.Join(be.errs...)
 }
 
-// NotNil checks if any error inside.
+// NotNil checks if there is at least one error inside the BatchError.
 func (be *BatchError) NotNil() bool {
+	be.lock.RLock()
+	defer be.lock.RUnlock()
+
 	return len(be.errs) > 0
 }
-
-// Error returns a string that represents inside errors.
-func (ea errorArray) Error() string {
-	var buf bytes.Buffer
-
-	for i := range ea {
-		if i > 0 {
-			buf.WriteByte('\n')
-		}
-		buf.WriteString(ea[i].Error())
-	}
-
-	return buf.String()
-}

core/errorx/batcherror_test.go

@@ -3,6 +3,7 @@ package errorx
 import (
 	"errors"
 	"fmt"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -33,7 +34,7 @@ func TestBatchErrorNilFromFunc(t *testing.T) {
 func TestBatchErrorOneError(t *testing.T) {
 	var batch BatchError
 	batch.Add(errors.New(err1))
-	assert.NotNil(t, batch)
+	assert.NotNil(t, batch.Err())
 	assert.Equal(t, err1, batch.Err().Error())
 	assert.True(t, batch.NotNil())
 }
@@ -42,7 +43,105 @@ func TestBatchErrorWithErrors(t *testing.T) {
 	var batch BatchError
 	batch.Add(errors.New(err1))
 	batch.Add(errors.New(err2))
-	assert.NotNil(t, batch)
+	assert.NotNil(t, batch.Err())
 	assert.Equal(t, fmt.Sprintf("%s\n%s", err1, err2), batch.Err().Error())
 	assert.True(t, batch.NotNil())
 }
+
+func TestBatchErrorConcurrentAdd(t *testing.T) {
+	const count = 10000
+	var batch BatchError
+	var wg sync.WaitGroup
+
+	wg.Add(count)
+	for i := 0; i < count; i++ {
+		go func() {
+			defer wg.Done()
+			batch.Add(errors.New(err1))
+		}()
+	}
+	wg.Wait()
+
+	assert.NotNil(t, batch.Err())
+	assert.Equal(t, count, len(batch.errs))
+	assert.True(t, batch.NotNil())
+}
+
+func TestBatchError_Unwrap(t *testing.T) {
+	t.Run("nil", func(t *testing.T) {
+		var be BatchError
+		assert.Nil(t, be.Err())
+		assert.True(t, errors.Is(be.Err(), nil))
+	})
+
+	t.Run("one error", func(t *testing.T) {
+		var errFoo = errors.New("foo")
+		var errBar = errors.New("bar")
+		var be BatchError
+		be.Add(errFoo)
+		assert.True(t, errors.Is(be.Err(), errFoo))
+		assert.False(t, errors.Is(be.Err(), errBar))
+	})
+
+	t.Run("two errors", func(t *testing.T) {
+		var errFoo = errors.New("foo")
+		var errBar = errors.New("bar")
+		var errBaz = errors.New("baz")
+		var be BatchError
+		be.Add(errFoo)
+		be.Add(errBar)
+		assert.True(t, errors.Is(be.Err(), errFoo))
+		assert.True(t, errors.Is(be.Err(), errBar))
+		assert.False(t, errors.Is(be.Err(), errBaz))
+	})
+}
+
+func TestBatchError_Add(t *testing.T) {
+	var be BatchError
+
+	// Test adding nil errors
+	be.Add(nil, nil)
+	assert.False(t, be.NotNil(), "Expected BatchError to be empty after adding nil errors")
+
+	// Test adding non-nil errors
+	err1 := errors.New("error 1")
+	err2 := errors.New("error 2")
+	be.Add(err1, err2)
+	assert.True(t, be.NotNil(), "Expected BatchError to be non-empty after adding errors")
+
+	// Test adding a mix of nil and non-nil errors
+	err3 := errors.New("error 3")
+	be.Add(nil, err3, nil)
+	assert.True(t, be.NotNil(), "Expected BatchError to be non-empty after adding a mix of nil and non-nil errors")
+}
+
+func TestBatchError_Err(t *testing.T) {
+	var be BatchError
+
+	// Test Err() on empty BatchError
+	assert.Nil(t, be.Err(), "Expected nil error for empty BatchError")
+
+	// Test Err() with multiple errors
+	err1 := errors.New("error 1")
+	err2 := errors.New("error 2")
+	be.Add(err1, err2)
+
+	combinedErr := be.Err()
+	assert.NotNil(t, combinedErr, "Expected nil error for BatchError with multiple errors")
+
+	// Check if the combined error contains both error messages
+	errString := combinedErr.Error()
+	assert.Truef(t, errors.Is(combinedErr, err1), "Combined error doesn't contain first error: %s", errString)
+	assert.Truef(t, errors.Is(combinedErr, err2), "Combined error doesn't contain second error: %s", errString)
+}
+
+func TestBatchError_NotNil(t *testing.T) {
+	var be BatchError
+
+	// Test NotNil() on empty BatchError
+	assert.Nil(t, be.Err(), "Expected nil error for empty BatchError")
+
+	// Test NotNil() after adding an error
+	be.Add(errors.New("test error"))
+	assert.NotNil(t, be.Err(), "Expected non-nil error after adding an error")
+}

core/errorx/check.go

@@ -0,0 +1,14 @@
+package errorx
+
+import "errors"
+
+// In checks if the given err is one of errs.
+func In(err error, errs ...error) bool {
+	for _, each := range errs {
+		if errors.Is(err, each) {
+			return true
+		}
+	}
+
+	return false
+}

core/errorx/check_test.go

@@ -0,0 +1,70 @@
+package errorx
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestIn(t *testing.T) {
+	err1 := errors.New("error 1")
+	err2 := errors.New("error 2")
+	err3 := errors.New("error 3")
+
+	tests := []struct {
+		name string
+		err  error
+		errs []error
+		want bool
+	}{
+		{
+			name: "Error matches one of the errors in the list",
+			err:  err1,
+			errs: []error{err1, err2},
+			want: true,
+		},
+		{
+			name: "Error does not match any errors in the list",
+			err:  err3,
+			errs: []error{err1, err2},
+			want: false,
+		},
+		{
+			name: "Empty error list",
+			err:  err1,
+			errs: []error{},
+			want: false,
+		},
+		{
+			name: "Nil error with non-nil list",
+			err:  nil,
+			errs: []error{err1, err2},
+			want: false,
+		},
+		{
+			name: "Non-nil error with nil in list",
+			err:  err1,
+			errs: []error{nil, err2},
+			want: false,
+		},
+		{
+			name: "Error matches nil error in the list",
+			err:  nil,
+			errs: []error{nil, err2},
+			want: true,
+		},
+		{
+			name: "Nil error with empty list",
+			err:  nil,
+			errs: []error{},
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := In(tt.err, tt.errs...); got != tt.want {
+				t.Errorf("In() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

core/filex/file.go

@@ -35,6 +35,7 @@ func firstLine(file *os.File) (string, error) {
 	for {
 		buf := make([]byte, bufSize)
 		n, err := file.ReadAt(buf, offset)
+
 		if err != nil && err != io.EOF {
 			return "", err
 		}
@@ -45,6 +46,10 @@ func firstLine(file *os.File) (string, error) {
 			}
 		}
 
+		if err == io.EOF {
+			return string(append(first, buf[:n]...)), nil
+		}
+
 		first = append(first, buf[:n]...)
 		offset += bufSize
 	}
@@ -57,30 +62,42 @@ func lastLine(filename string, file *os.File) (string, error) {
 	}
 
 	var last []byte
+	bufLen := int64(bufSize)
 	offset := info.Size()
-	for {
-		offset -= bufSize
-		if offset < 0 {
+
+	for offset > 0 {
+		if offset < bufLen {
+			bufLen = offset
 			offset = 0
+		} else {
+			offset -= bufLen
 		}
-		buf := make([]byte, bufSize)
+
+		buf := make([]byte, bufLen)
 		n, err := file.ReadAt(buf, offset)
 		if err != nil && err != io.EOF {
 			return "", err
 		}
 
+		if n == 0 {
+			break
+		}
+
 		if buf[n-1] == '\n' {
 			buf = buf[:n-1]
 			n--
 		} else {
 			buf = buf[:n]
 		}
-		for n--; n >= 0; n-- {
-			if buf[n] == '\n' {
-				return string(append(buf[n+1:], last...)), nil
+
+		for i := n - 1; i >= 0; i-- {
+			if buf[i] == '\n' {
+				return string(append(buf[i+1:], last...)), nil
 			}
 		}
 
 		last = append(buf, last...)
 	}
+
+	return string(last), nil
 }

core/filex/file_test.go

@@ -52,6 +52,7 @@ last line`
 second line
 last line
 `
+	emptyContent = ``
 )
 
 func TestFirstLine(t *testing.T) {
@@ -79,6 +80,26 @@ func TestFirstLineError(t *testing.T) {
 	assert.Error(t, err)
 }
 
+func TestFirstLineEmptyFile(t *testing.T) {
+	filename, err := fs.TempFilenameWithText(emptyContent)
+	assert.Nil(t, err)
+	defer os.Remove(filename)
+
+	val, err := FirstLine(filename)
+	assert.Nil(t, err)
+	assert.Equal(t, "", val)
+}
+
+func TestFirstLineWithoutNewline(t *testing.T) {
+	filename, err := fs.TempFilenameWithText(longLine)
+	assert.Nil(t, err)
+	defer os.Remove(filename)
+
+	val, err := FirstLine(filename)
+	assert.Nil(t, err)
+	assert.Equal(t, longLine, val)
+}
+
 func TestLastLine(t *testing.T) {
 	filename, err := fs.TempFilenameWithText(text)
 	assert.Nil(t, err)
@@ -99,6 +120,16 @@ func TestLastLineWithLastNewline(t *testing.T) {
 	assert.Equal(t, longLine, val)
 }
 
+func TestLastLineWithoutLastNewline(t *testing.T) {
+	filename, err := fs.TempFilenameWithText(longLine)
+	assert.Nil(t, err)
+	defer os.Remove(filename)
+
+	val, err := LastLine(filename)
+	assert.Nil(t, err)
+	assert.Equal(t, longLine, val)
+}
+
 func TestLastLineShort(t *testing.T) {
 	filename, err := fs.TempFilenameWithText(shortText)
 	assert.Nil(t, err)
@@ -123,3 +154,67 @@ func TestLastLineError(t *testing.T) {
 	_, err := LastLine("/tmp/does-not-exist")
 	assert.Error(t, err)
 }
+
+func TestLastLineEmptyFile(t *testing.T) {
+	filename, err := fs.TempFilenameWithText(emptyContent)
+	assert.Nil(t, err)
+	defer os.Remove(filename)
+
+	val, err := LastLine(filename)
+	assert.Nil(t, err)
+	assert.Equal(t, "", val)
+}
+
+func TestFirstLineExactlyBufSize(t *testing.T) {
+	content := make([]byte, bufSize)
+	for i := range content {
+		content[i] = 'a'
+	}
+	content[bufSize-1] = '\n' // Ensure there is a newline at the edge
+
+	filename, err := fs.TempFilenameWithText(string(content))
+	assert.Nil(t, err)
+	defer os.Remove(filename)
+
+	val, err := FirstLine(filename)
+	assert.Nil(t, err)
+	assert.Equal(t, string(content[:bufSize-1]), val)
+}
+
+func TestLastLineExactlyBufSize(t *testing.T) {
+	content := make([]byte, bufSize)
+	for i := range content {
+		content[i] = 'a'
+	}
+	content[bufSize-1] = '\n' // Ensure there is a newline at the edge
+
+	filename, err := fs.TempFilenameWithText(string(content))
+	assert.Nil(t, err)
+	defer os.Remove(filename)
+
+	val, err := LastLine(filename)
+	assert.Nil(t, err)
+	assert.Equal(t, string(content[:bufSize-1]), val)
+}
+
+func TestFirstLineLargeFile(t *testing.T) {
+	content := text + text + text + "\n" + "extra"
+	filename, err := fs.TempFilenameWithText(content)
+	assert.Nil(t, err)
+	defer os.Remove(filename)
+
+	val, err := FirstLine(filename)
+	assert.Nil(t, err)
+	assert.Equal(t, "first line", val)
+}
+
+func TestLastLineLargeFile(t *testing.T) {
+	content := text + text + text + "\n" + "extra"
+	filename, err := fs.TempFilenameWithText(content)
+	assert.Nil(t, err)
+	defer os.Remove(filename)
+
+	val, err := LastLine(filename)
+	assert.Nil(t, err)
+	assert.Equal(t, "extra", val)
+}

core/filex/progressscanner.go

@@ -5,7 +5,7 @@ import "gopkg.in/cheggaaa/pb.v1"
 type (
 	// A Scanner is used to read lines.
 	Scanner interface {
-		// Scan checks if has remaining to read.
+		// Scan checks if it has remaining to read.
 		Scan() bool
 		// Text returns next line.
 		Text() string

core/fs/files.go

@@ -1,4 +1,4 @@
-//go:build linux || darwin
+//go:build linux || darwin || freebsd
 
 package fs
 

core/fx/parallel.go

@@ -1,6 +1,9 @@
 package fx
 
-import "github.com/zeromicro/go-zero/core/threading"
+import (
+	"github.com/zeromicro/go-zero/core/errorx"
+	"github.com/zeromicro/go-zero/core/threading"
+)
 
 // Parallel runs fns parallelly and waits for done.
 func Parallel(fns ...func()) {
@@ -10,3 +13,20 @@ func Parallel(fns ...func()) {
 	}
 	group.Wait()
 }
+
+func ParallelErr(fns ...func() error) error {
+	var be errorx.BatchError
+
+	group := threading.NewRoutineGroup()
+	for _, fn := range fns {
+		f := fn
+		group.RunSafe(func() {
+			if err := f(); err != nil {
+				be.Add(err)
+			}
+		})
+	}
+	group.Wait()
+
+	return be.Err()
+}

core/fx/parallel_test.go

@@ -1,6 +1,7 @@
 package fx
 
 import (
+	"errors"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -22,3 +23,54 @@ func TestParallel(t *testing.T) {
 	})
 	assert.Equal(t, int32(6), count)
 }
+
+func TestParallelErr(t *testing.T) {
+	var count int32
+	err := ParallelErr(
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 1)
+			return errors.New("failed to exec #1")
+		},
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 2)
+			return errors.New("failed to exec #2")
+
+		},
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 3)
+			return nil
+		},
+	)
+
+	assert.Equal(t, int32(6), count)
+	assert.Error(t, err)
+	assert.ErrorContains(t, err, "failed to exec #1", "failed to exec #2")
+}
+
+func TestParallelErrErrorNil(t *testing.T) {
+	var count int32
+	err := ParallelErr(
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 1)
+			return nil
+		},
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 2)
+			return nil
+
+		},
+		func() error {
+			time.Sleep(time.Millisecond * 100)
+			atomic.AddInt32(&count, 3)
+			return nil
+		},
+	)
+
+	assert.Equal(t, int32(6), count)
+	assert.NoError(t, err)
+}

core/fx/retry.go

@@ -2,6 +2,7 @@ package fx
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	"github.com/zeromicro/go-zero/core/errorx"
@@ -14,9 +15,10 @@ type (
 	RetryOption func(*retryOptions)
 
 	retryOptions struct {
-		times    int
-		interval time.Duration
-		timeout  time.Duration
+		times        int
+		interval     time.Duration
+		timeout      time.Duration
+		ignoreErrors []error
 	}
 )
 
@@ -62,6 +64,11 @@ func retry(ctx context.Context, fn func(errChan chan error, retryCount int), opt
 		select {
 		case err := <-errChan:
 			if err != nil {
+				for _, ignoreErr := range options.ignoreErrors {
+					if errors.Is(err, ignoreErr) {
+						return nil
+					}
+				}
 				berr.Add(err)
 			} else {
 				return nil
@@ -84,19 +91,28 @@ func retry(ctx context.Context, fn func(errChan chan error, retryCount int), opt
 	return berr.Err()
 }
 
-// WithRetry customize a DoWithRetry call with given retry times.
-func WithRetry(times int) RetryOption {
+// WithIgnoreErrors Ignore the specified errors
+func WithIgnoreErrors(ignoreErrors []error) RetryOption {
 	return func(options *retryOptions) {
-		options.times = times
+		options.ignoreErrors = ignoreErrors
 	}
 }
 
+// WithInterval customizes a DoWithRetry call with given interval.
 func WithInterval(interval time.Duration) RetryOption {
 	return func(options *retryOptions) {
 		options.interval = interval
 	}
 }
 
+// WithRetry customizes a DoWithRetry call with given retry times.
+func WithRetry(times int) RetryOption {
+	return func(options *retryOptions) {
+		options.times = times
+	}
+}
+
+// WithTimeout customizes a DoWithRetry call with given timeout.
 func WithTimeout(timeout time.Duration) RetryOption {
 	return func(options *retryOptions) {
 		options.timeout = timeout

core/fx/retry_test.go

@@ -97,6 +97,24 @@ func TestRetryWithInterval(t *testing.T) {
 
 }
 
+func TestRetryWithWithIgnoreErrors(t *testing.T) {
+	ignoreErr1 := errors.New("ignore error1")
+	ignoreErr2 := errors.New("ignore error2")
+	ignoreErrs := []error{ignoreErr1, ignoreErr2}
+
+	assert.Nil(t, DoWithRetry(func() error {
+		return ignoreErr1
+	}, WithIgnoreErrors(ignoreErrs)))
+
+	assert.Nil(t, DoWithRetry(func() error {
+		return ignoreErr2
+	}, WithIgnoreErrors(ignoreErrs)))
+
+	assert.NotNil(t, DoWithRetry(func() error {
+		return errors.New("any")
+	}))
+}
+
 func TestRetryCtx(t *testing.T) {
 	t.Run("with timeout", func(t *testing.T) {
 		assert.NotNil(t, DoWithRetryCtx(context.Background(), func(ctx context.Context, retryCount int) error {

core/fx/stream.go

@@ -84,10 +84,10 @@ func Range(source <-chan any) Stream {
 	}
 }
 
-// AllMach returns whether all elements of this stream match the provided predicate.
+// AllMatch returns whether all elements of this stream match the provided predicate.
 // May not evaluate the predicate on all elements if not necessary for determining the result.
 // If the stream is empty then true is returned and the predicate is not evaluated.
-func (s Stream) AllMach(predicate func(item any) bool) bool {
+func (s Stream) AllMatch(predicate func(item any) bool) bool {
 	for item := range s.source {
 		if !predicate(item) {
 			// make sure the former goroutine not block, and current func returns fast.
@@ -99,10 +99,10 @@ func (s Stream) AllMach(predicate func(item any) bool) bool {
 	return true
 }
 
-// AnyMach returns whether any elements of this stream match the provided predicate.
+// AnyMatch returns whether any elements of this stream match the provided predicate.
 // May not evaluate the predicate on all elements if not necessary for determining the result.
 // If the stream is empty then false is returned and the predicate is not evaluated.
-func (s Stream) AnyMach(predicate func(item any) bool) bool {
+func (s Stream) AnyMatch(predicate func(item any) bool) bool {
 	for item := range s.source {
 		if predicate(item) {
 			// make sure the former goroutine not block, and current func returns fast.
@@ -352,7 +352,7 @@ func (s Stream) Parallel(fn ParallelFunc, opts ...Option) {
 	}, opts...).Done()
 }
 
-// Reduce is an utility method to let the caller deal with the underlying channel.
+// Reduce is a utility method to let the caller deal with the underlying channel.
 func (s Stream) Reduce(fn ReduceFunc) (any, error) {
 	return fn(s.source)
 }

core/fx/stream_test.go

@@ -398,16 +398,16 @@ func TestWalk(t *testing.T) {
 
 func TestStream_AnyMach(t *testing.T) {
 	runCheckedTest(t, func(t *testing.T) {
-		assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item any) bool {
+		assetEqual(t, false, Just(1, 2, 3).AnyMatch(func(item any) bool {
 			return item.(int) == 4
 		}))
-		assetEqual(t, false, Just(1, 2, 3).AnyMach(func(item any) bool {
+		assetEqual(t, false, Just(1, 2, 3).AnyMatch(func(item any) bool {
 			return item.(int) == 0
 		}))
-		assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item any) bool {
+		assetEqual(t, true, Just(1, 2, 3).AnyMatch(func(item any) bool {
 			return item.(int) == 2
 		}))
-		assetEqual(t, true, Just(1, 2, 3).AnyMach(func(item any) bool {
+		assetEqual(t, true, Just(1, 2, 3).AnyMatch(func(item any) bool {
 			return item.(int) == 2
 		}))
 	})
@@ -416,17 +416,17 @@ func TestStream_AnyMach(t *testing.T) {
 func TestStream_AllMach(t *testing.T) {
 	runCheckedTest(t, func(t *testing.T) {
 		assetEqual(
-			t, true, Just(1, 2, 3).AllMach(func(item any) bool {
+			t, true, Just(1, 2, 3).AllMatch(func(item any) bool {
 				return true
 			}),
 		)
 		assetEqual(
-			t, false, Just(1, 2, 3).AllMach(func(item any) bool {
+			t, false, Just(1, 2, 3).AllMatch(func(item any) bool {
 				return false
 			}),
 		)
 		assetEqual(
-			t, false, Just(1, 2, 3).AllMach(func(item any) bool {
+			t, false, Just(1, 2, 3).AllMatch(func(item any) bool {
 				return item.(int) == 1
 			}),
 		)

core/hash/consistenthash_test.go

@@ -86,21 +86,16 @@ func TestConsistentHashIncrementalTransfer(t *testing.T) {
 
 func TestConsistentHashTransferOnFailure(t *testing.T) {
 	index := 41
-	keys, newKeys := getKeysBeforeAndAfterFailure(t, "localhost:", index)
-	var transferred int
-	for k, v := range newKeys {
-		if v != keys[k] {
-			transferred++
-		}
-	}
-
-	ratio := float32(transferred) / float32(requestSize)
-	assert.True(t, ratio < 2.5/float32(keySize), fmt.Sprintf("%d: %f", index, ratio))
+	ratioNotExists := getTransferRatioOnFailure(t, index)
+	assert.True(t, ratioNotExists == 0, fmt.Sprintf("%d: %f", index, ratioNotExists))
+	index = 13
+	ratio := getTransferRatioOnFailure(t, index)
+	assert.True(t, ratio < 2.5/keySize, fmt.Sprintf("%d: %f", index, ratio))
 }
 
 func TestConsistentHashLeastTransferOnFailure(t *testing.T) {
 	prefix := "localhost:"
-	index := 41
+	index := 13
 	keys, newKeys := getKeysBeforeAndAfterFailure(t, prefix, index)
 	for k, v := range keys {
 		newV := newKeys[k]
@@ -164,6 +159,17 @@ func getKeysBeforeAndAfterFailure(t *testing.T, prefix string, index int) (map[i
 	return keys, newKeys
 }
 
+func getTransferRatioOnFailure(t *testing.T, index int) float32 {
+	keys, newKeys := getKeysBeforeAndAfterFailure(t, "localhost:", index)
+	var transferred int
+	for k, v := range newKeys {
+		if v != keys[k] {
+			transferred++
+		}
+	}
+	return float32(transferred) / float32(requestSize)
+}
+
 type mockNode struct {
 	addr string
 	id   int

core/hash/hash.go

@@ -2,7 +2,7 @@ package hash
 
 import (
 	"crypto/md5"
-	"fmt"
+	"encoding/hex"
 
 	"github.com/spaolacci/murmur3"
 )
@@ -20,6 +20,7 @@ func Md5(data []byte) []byte {
 }
 
 // Md5Hex returns the md5 hex string of data.
+// This function is optimized for better performance than fmt.Sprintf.
 func Md5Hex(data []byte) string {
-	return fmt.Sprintf("%x", Md5(data))
+	return hex.EncodeToString(Md5(data))
 }

core/iox/pipe.go

@@ -2,7 +2,7 @@ package iox
 
 import "os"
 
-// RedirectInOut redirects stdin to r, stdout to w, and callers need to call restore afterwards.
+// RedirectInOut redirects stdin to r, stdout to w, and callers need to call restore afterward.
 func RedirectInOut() (restore func(), err error) {
 	var r, w *os.File
 	r, w, err = os.Pipe()

core/iox/textfile.go

@@ -9,7 +9,7 @@ import (
 
 const bufSize = 32 * 1024
 
-// CountLines returns the number of lines in file.
+// CountLines returns the number of lines in the file.
 func CountLines(file string) (int, error) {
 	f, err := os.Open(file)
 	if err != nil {

core/iox/textlinescanner.go

@@ -2,11 +2,12 @@ package iox
 
 import (
 	"bufio"
+	"errors"
 	"io"
 	"strings"
 )
 
-// A TextLineScanner is a scanner that can scan lines from given reader.
+// A TextLineScanner is a scanner that can scan lines from the given reader.
 type TextLineScanner struct {
 	reader  *bufio.Reader
 	hasNext bool
@@ -14,7 +15,7 @@ type TextLineScanner struct {
 	err     error
 }
 
-// NewTextLineScanner returns a TextLineScanner with given reader.
+// NewTextLineScanner returns a TextLineScanner with the given reader.
 func NewTextLineScanner(reader io.Reader) *TextLineScanner {
 	return &TextLineScanner{
 		reader:  bufio.NewReader(reader),
@@ -30,7 +31,7 @@ func (scanner *TextLineScanner) Scan() bool {
 
 	line, err := scanner.reader.ReadString('\n')
 	scanner.line = strings.TrimRight(line, "\n")
-	if err == io.EOF {
+	if errors.Is(err, io.EOF) {
 		scanner.hasNext = false
 		return true
 	} else if err != nil {

core/jsonx/json.go

@@ -8,9 +8,25 @@ import (
 	"strings"
 )
 
-// Marshal marshals v into json bytes.
+// Marshal marshals v into json bytes, without escaping HTML and removes the trailing newline.
 func Marshal(v any) ([]byte, error) {
-	return json.Marshal(v)
+	// why not use json.Marshal?  https://github.com/golang/go/issues/28453
+	// it changes the behavior of json.Marshal, like & -> \u0026, < -> \u003c, > -> \u003e
+	// which is not what we want in API responses
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	enc.SetEscapeHTML(false)
+	if err := enc.Encode(v); err != nil {
+		return nil, err
+	}
+
+	bs := buf.Bytes()
+	// Remove trailing newline added by json.Encoder.Encode
+	if len(bs) > 0 && bs[len(bs)-1] == '\n' {
+		bs = bs[:len(bs)-1]
+	}
+
+	return bs, nil
 }
 
 // MarshalToString marshals v into a string.

core/jsonx/json_test.go

@@ -1,6 +1,7 @@
 package jsonx
 
 import (
+	"fmt"
 	"strings"
 	"testing"
 
@@ -101,3 +102,105 @@ func TestUnmarshalFromReaderError(t *testing.T) {
 	err := UnmarshalFromReader(strings.NewReader(s), &v)
 	assert.NotNil(t, err)
 }
+
+func Test_doMarshalJson(t *testing.T) {
+	type args struct {
+		v any
+	}
+
+	tests := []struct {
+		name    string
+		args    args
+		want    []byte
+		wantErr assert.ErrorAssertionFunc
+	}{
+		{
+			name:    "nil",
+			args:    args{nil},
+			want:    []byte("null"),
+			wantErr: assert.NoError,
+		},
+		{
+			name:    "string",
+			args:    args{"hello"},
+			want:    []byte(`"hello"`),
+			wantErr: assert.NoError,
+		},
+		{
+			name:    "int",
+			args:    args{42},
+			want:    []byte("42"),
+			wantErr: assert.NoError,
+		},
+		{
+			name:    "bool",
+			args:    args{true},
+			want:    []byte("true"),
+			wantErr: assert.NoError,
+		},
+		{
+			name: "struct",
+			args: args{
+				struct {
+					Name string `json:"name"`
+				}{Name: "test"},
+			},
+			want:    []byte(`{"name":"test"}`),
+			wantErr: assert.NoError,
+		},
+		{
+			name:    "slice",
+			args:    args{[]int{1, 2, 3}},
+			want:    []byte("[1,2,3]"),
+			wantErr: assert.NoError,
+		},
+		{
+			name:    "map",
+			args:    args{map[string]int{"a": 1, "b": 2}},
+			want:    []byte(`{"a":1,"b":2}`),
+			wantErr: assert.NoError,
+		},
+		{
+			name:    "unmarshalable type",
+			args:    args{complex(1, 2)},
+			want:    nil,
+			wantErr: assert.Error,
+		},
+		{
+			name:    "channel type",
+			args:    args{make(chan int)},
+			want:    nil,
+			wantErr: assert.Error,
+		},
+		{
+			name:    "url with query params",
+			args:    args{"https://example.com/api?name=test&age=25"},
+			want:    []byte(`"https://example.com/api?name=test&age=25"`),
+			wantErr: assert.NoError,
+		},
+		{
+			name:    "url with encoded query params",
+			args:    args{"https://example.com/api?data=hello%20world&special=%26%3D"},
+			want:    []byte(`"https://example.com/api?data=hello%20world&special=%26%3D"`),
+			wantErr: assert.NoError,
+		},
+		{
+			name:    "url with multiple query params",
+			args:    args{"http://localhost:8080/users?page=1&limit=10&sort=name&order=asc"},
+			want:    []byte(`"http://localhost:8080/users?page=1&limit=10&sort=name&order=asc"`),
+			wantErr: assert.NoError,
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := Marshal(tt.args.v)
+			if !tt.wantErr(t, err, fmt.Sprintf("Marshal(%v)", tt.args.v)) {
+				return
+			}
+
+			assert.Equalf(t, string(tt.want), string(got), "Marshal(%v)", tt.args.v)
+		})
+	}
+}

core/limit/periodlimit.go

@@ -2,6 +2,7 @@ package limit
 
 import (
 	"context"
+	_ "embed"
 	"errors"
 	"strconv"
 	"time"
@@ -28,20 +29,9 @@ var (
 	// ErrUnknownCode is an error that represents unknown status code.
 	ErrUnknownCode = errors.New("unknown status code")
 
-	// to be compatible with aliyun redis, we cannot use `local key = KEYS[1]` to reuse the key
-	periodScript = redis.NewScript(`local limit = tonumber(ARGV[1])
-local window = tonumber(ARGV[2])
-local current = redis.call("INCRBY", KEYS[1], 1)
-if current == 1 then
-    redis.call("expire", KEYS[1], window)
-end
-if current < limit then
-    return 1
-elseif current == limit then
-    return 2
-else
-    return 0
-end`)
+	//go:embed periodscript.lua
+	periodLuaScript string
+	periodScript    = redis.NewScript(periodLuaScript)
 )
 
 type (

core/limit/periodscript.lua

@@ -0,0 +1,14 @@
+-- to be compatible with aliyun redis, we cannot use `local key = KEYS[1]` to reuse the key
+local limit = tonumber(ARGV[1])
+local window = tonumber(ARGV[2])
+local current = redis.call("INCRBY", KEYS[1], 1)
+if current == 1 then
+    redis.call("expire", KEYS[1], window)
+end
+if current < limit then
+    return 1
+elseif current == limit then
+    return 2
+else
+    return 0
+end

core/limit/tokenlimit.go

@@ -2,6 +2,7 @@ package limit
 
 import (
 	"context"
+	_ "embed"
 	"errors"
 	"fmt"
 	"strconv"
@@ -9,6 +10,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/zeromicro/go-zero/core/errorx"
 	"github.com/zeromicro/go-zero/core/logx"
 	"github.com/zeromicro/go-zero/core/stores/redis"
 	xrate "golang.org/x/time/rate"
@@ -20,37 +22,11 @@ const (
 	pingInterval    = time.Millisecond * 100
 )
 
-// to be compatible with aliyun redis, we cannot use `local key = KEYS[1]` to reuse the key
-// KEYS[1] as tokens_key
-// KEYS[2] as timestamp_key
-var script = redis.NewScript(`local rate = tonumber(ARGV[1])
-local capacity = tonumber(ARGV[2])
-local now = tonumber(ARGV[3])
-local requested = tonumber(ARGV[4])
-local fill_time = capacity/rate
-local ttl = math.floor(fill_time*2)
-local last_tokens = tonumber(redis.call("get", KEYS[1]))
-if last_tokens == nil then
-    last_tokens = capacity
-end
-
-local last_refreshed = tonumber(redis.call("get", KEYS[2]))
-if last_refreshed == nil then
-    last_refreshed = 0
-end
-
-local delta = math.max(0, now-last_refreshed)
-local filled_tokens = math.min(capacity, last_tokens+(delta*rate))
-local allowed = filled_tokens >= requested
-local new_tokens = filled_tokens
-if allowed then
-    new_tokens = filled_tokens - requested
-end
-
-redis.call("setex", KEYS[1], ttl, new_tokens)
-redis.call("setex", KEYS[2], ttl, now)
-
-return allowed`)
+var (
+	//go:embed tokenscript.lua
+	tokenLuaScript string
+	tokenScript    = redis.NewScript(tokenLuaScript)
+)
 
 // A TokenLimiter controls how frequently events are allowed to happen with in one second.
 type TokenLimiter struct {
@@ -112,7 +88,7 @@ func (lim *TokenLimiter) reserveN(ctx context.Context, now time.Time, n int) boo
 	}
 
 	resp, err := lim.store.ScriptRunCtx(ctx,
-		script,
+		tokenScript,
 		[]string{
 			lim.tokenKey,
 			lim.timestampKey,
@@ -125,10 +101,10 @@ func (lim *TokenLimiter) reserveN(ctx context.Context, now time.Time, n int) boo
 		})
 	// redis allowed == false
 	// Lua boolean false -> r Nil bulk reply
-	if err == redis.Nil {
+	if errors.Is(err, redis.Nil) {
 		return false
 	}
-	if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+	if errorx.In(err, context.DeadlineExceeded, context.Canceled) {
 		logx.Errorf("fail to use rate limiter: %s", err)
 		return false
 	}

core/limit/tokenscript.lua

@@ -0,0 +1,31 @@
+-- to be compatible with aliyun redis, we cannot use `local key = KEYS[1]` to reuse the key
+-- KEYS[1] as tokens_key
+-- KEYS[2] as timestamp_key
+local rate = tonumber(ARGV[1])
+local capacity = tonumber(ARGV[2])
+local now = tonumber(ARGV[3])
+local requested = tonumber(ARGV[4])
+local fill_time = capacity/rate
+local ttl = math.floor(fill_time*2)
+local last_tokens = tonumber(redis.call("get", KEYS[1]))
+if last_tokens == nil then
+    last_tokens = capacity
+end
+
+local last_refreshed = tonumber(redis.call("get", KEYS[2]))
+if last_refreshed == nil then
+    last_refreshed = 0
+end
+
+local delta = math.max(0, now-last_refreshed)
+local filled_tokens = math.min(capacity, last_tokens+(delta*rate))
+local allowed = filled_tokens >= requested
+local new_tokens = filled_tokens
+if allowed then
+    new_tokens = filled_tokens - requested
+end
+
+redis.call("setex", KEYS[1], ttl, new_tokens)
+redis.call("setex", KEYS[2], ttl, now)
+
+return allowed

core/load/adaptiveshedder.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/zeromicro/go-zero/core/collection"
 	"github.com/zeromicro/go-zero/core/logx"
+	"github.com/zeromicro/go-zero/core/mathx"
 	"github.com/zeromicro/go-zero/core/stat"
 	"github.com/zeromicro/go-zero/core/syncx"
 	"github.com/zeromicro/go-zero/core/timex"
@@ -21,8 +22,11 @@ const (
 	defaultCpuThreshold = 900
 	defaultMinRt        = float64(time.Second / time.Millisecond)
 	// moving average hyperparameter beta for calculating requests on the fly
-	flyingBeta      = 0.9
-	coolOffDuration = time.Second
+	flyingBeta               = 0.9
+	coolOffDuration          = time.Second
+	cpuMax                   = 1000 // millicpu
+	millisecondsPerSecond    = 1000
+	overloadFactorLowerBound = 0.1
 )
 
 var (
@@ -66,14 +70,14 @@ type (
 
 	adaptiveShedder struct {
 		cpuThreshold    int64
-		windows         int64
+		windowScale     float64
 		flying          int64
 		avgFlying       float64
 		avgFlyingLock   syncx.SpinLock
 		overloadTime    *syncx.AtomicDuration
 		droppedRecently *syncx.AtomicBool
-		passCounter     *collection.RollingWindow
-		rtCounter       *collection.RollingWindow
+		passCounter     *collection.RollingWindow[int64, *collection.Bucket[int64]]
+		rtCounter       *collection.RollingWindow[int64, *collection.Bucket[int64]]
 	}
 )
 
@@ -103,15 +107,16 @@ func NewAdaptiveShedder(opts ...ShedderOption) Shedder {
 		opt(&options)
 	}
 	bucketDuration := options.window / time.Duration(options.buckets)
+	newBucket := func() *collection.Bucket[int64] {
+		return new(collection.Bucket[int64])
+	}
 	return &adaptiveShedder{
 		cpuThreshold:    options.cpuThreshold,
-		windows:         int64(time.Second / bucketDuration),
+		windowScale:     float64(time.Second) / float64(bucketDuration) / millisecondsPerSecond,
 		overloadTime:    syncx.NewAtomicDuration(),
 		droppedRecently: syncx.NewAtomicBool(),
-		passCounter: collection.NewRollingWindow(options.buckets, bucketDuration,
-			collection.IgnoreCurrentBucket()),
-		rtCounter: collection.NewRollingWindow(options.buckets, bucketDuration,
-			collection.IgnoreCurrentBucket()),
+		passCounter:     collection.NewRollingWindow[int64, *collection.Bucket[int64]](newBucket, options.buckets, bucketDuration, collection.IgnoreCurrentBucket[int64, *collection.Bucket[int64]]()),
+		rtCounter:       collection.NewRollingWindow[int64, *collection.Bucket[int64]](newBucket, options.buckets, bucketDuration, collection.IgnoreCurrentBucket[int64, *collection.Bucket[int64]]()),
 	}
 }
 
@@ -134,10 +139,10 @@ func (as *adaptiveShedder) Allow() (Promise, error) {
 func (as *adaptiveShedder) addFlying(delta int64) {
 	flying := atomic.AddInt64(&as.flying, delta)
 	// update avgFlying when the request is finished.
-	// this strategy makes avgFlying have a little bit lag against flying, and smoother.
+	// this strategy makes avgFlying have a little bit of lag against flying, and smoother.
 	// when the flying requests increase rapidly, avgFlying increase slower, accept more requests.
-	// when the flying requests drop rapidly, avgFlying drop slower, accept less requests.
-	// it makes the service to serve as more requests as possible.
+	// when the flying requests drop rapidly, avgFlying drop slower, accept fewer requests.
+	// it makes the service to serve as many requests as possible.
 	if delta < 0 {
 		as.avgFlyingLock.Lock()
 		as.avgFlying = as.avgFlying*flyingBeta + float64(flying)*(1-flyingBeta)
@@ -149,39 +154,42 @@ func (as *adaptiveShedder) highThru() bool {
 	as.avgFlyingLock.Lock()
 	avgFlying := as.avgFlying
 	as.avgFlyingLock.Unlock()
-	maxFlight := as.maxFlight()
-	return int64(avgFlying) > maxFlight && atomic.LoadInt64(&as.flying) > maxFlight
+	maxFlight := as.maxFlight() * as.overloadFactor()
+	return avgFlying > maxFlight && float64(atomic.LoadInt64(&as.flying)) > maxFlight
 }
 
-func (as *adaptiveShedder) maxFlight() int64 {
+func (as *adaptiveShedder) maxFlight() float64 {
 	// windows = buckets per second
 	// maxQPS = maxPASS * windows
 	// minRT = min average response time in milliseconds
-	// maxQPS * minRT / milliseconds_per_second
-	return int64(math.Max(1, float64(as.maxPass()*as.windows)*(as.minRt()/1e3)))
+	// allowedFlying = maxQPS * minRT / milliseconds_per_second
+	maxFlight := float64(as.maxPass()) * as.minRt() * as.windowScale
+	return mathx.AtLeast(maxFlight, 1)
 }
 
 func (as *adaptiveShedder) maxPass() int64 {
-	var result float64 = 1
+	var result int64 = 1
 
-	as.passCounter.Reduce(func(b *collection.Bucket) {
+	as.passCounter.Reduce(func(b *collection.Bucket[int64]) {
 		if b.Sum > result {
 			result = b.Sum
 		}
 	})
 
-	return int64(result)
+	return result
 }
 
 func (as *adaptiveShedder) minRt() float64 {
+	// if no requests in previous windows, return defaultMinRt,
+	// its a reasonable large value to avoid dropping requests.
 	result := defaultMinRt
 
-	as.rtCounter.Reduce(func(b *collection.Bucket) {
+	as.rtCounter.Reduce(func(b *collection.Bucket[int64]) {
 		if b.Count <= 0 {
 			return
 		}
 
-		avg := math.Round(b.Sum / float64(b.Count))
+		avg := math.Round(float64(b.Sum) / float64(b.Count))
 		if avg < result {
 			result = avg
 		}
@@ -190,6 +198,13 @@ func (as *adaptiveShedder) minRt() float64 {
 	return result
 }
 
+func (as *adaptiveShedder) overloadFactor() float64 {
+	// as.cpuThreshold must be less than cpuMax
+	factor := (cpuMax - float64(stat.CpuUsage())) / (cpuMax - float64(as.cpuThreshold))
+	// at least accept 10% of acceptable requests, even cpu is highly overloaded.
+	return mathx.Between(factor, overloadFactorLowerBound, 1)
+}
+
 func (as *adaptiveShedder) shouldDrop() bool {
 	if as.systemOverloaded() || as.stillHot() {
 		if as.highThru() {
@@ -236,14 +251,14 @@ func (as *adaptiveShedder) systemOverloaded() bool {
 	return true
 }
 
-// WithBuckets customizes the Shedder with given number of buckets.
+// WithBuckets customizes the Shedder with the given number of buckets.
 func WithBuckets(buckets int) ShedderOption {
 	return func(opts *shedderOptions) {
 		opts.buckets = buckets
 	}
 }
 
-// WithCpuThreshold customizes the Shedder with given cpu threshold.
+// WithCpuThreshold customizes the Shedder with the given cpu threshold.
 func WithCpuThreshold(threshold int64) ShedderOption {
 	return func(opts *shedderOptions) {
 		opts.cpuThreshold = threshold
@@ -269,6 +284,6 @@ func (p *promise) Fail() {
 func (p *promise) Pass() {
 	rt := float64(timex.Since(p.start)) / float64(time.Millisecond)
 	p.shedder.addFlying(-1)
-	p.shedder.rtCounter.Add(math.Ceil(rt))
+	p.shedder.rtCounter.Add(int64(math.Ceil(rt)))
 	p.shedder.passCounter.Add(1)
 }

core/load/adaptiveshedder_test.go

@@ -19,6 +19,7 @@ import (
 const (
 	buckets        = 10
 	bucketDuration = time.Millisecond * 50
+	windowFactor   = 0.01
 )
 
 func init() {
@@ -57,7 +58,7 @@ func TestAdaptiveShedder(t *testing.T) {
 func TestAdaptiveShedderMaxPass(t *testing.T) {
 	passCounter := newRollingWindow()
 	for i := 1; i <= 10; i++ {
-		passCounter.Add(float64(i * 100))
+		passCounter.Add(int64(i * 100))
 		time.Sleep(bucketDuration)
 	}
 	shedder := &adaptiveShedder{
@@ -82,7 +83,7 @@ func TestAdaptiveShedderMinRt(t *testing.T) {
 			time.Sleep(bucketDuration)
 		}
 		for j := i*10 + 1; j <= i*10+10; j++ {
-			rtCounter.Add(float64(j))
+			rtCounter.Add(int64(j))
 		}
 	}
 	shedder := &adaptiveShedder{
@@ -106,18 +107,18 @@ func TestAdaptiveShedderMaxFlight(t *testing.T) {
 		if i > 0 {
 			time.Sleep(bucketDuration)
 		}
-		passCounter.Add(float64((i + 1) * 100))
+		passCounter.Add(int64((i + 1) * 100))
 		for j := i*10 + 1; j <= i*10+10; j++ {
-			rtCounter.Add(float64(j))
+			rtCounter.Add(int64(j))
 		}
 	}
 	shedder := &adaptiveShedder{
 		passCounter:     passCounter,
 		rtCounter:       rtCounter,
-		windows:         buckets,
+		windowScale:     windowFactor,
 		droppedRecently: syncx.NewAtomicBool(),
 	}
-	assert.Equal(t, int64(54), shedder.maxFlight())
+	assert.Equal(t, float64(54), shedder.maxFlight())
 }
 
 func TestAdaptiveShedderShouldDrop(t *testing.T) {
@@ -128,15 +129,15 @@ func TestAdaptiveShedderShouldDrop(t *testing.T) {
 		if i > 0 {
 			time.Sleep(bucketDuration)
 		}
-		passCounter.Add(float64((i + 1) * 100))
+		passCounter.Add(int64((i + 1) * 100))
 		for j := i*10 + 1; j <= i*10+10; j++ {
-			rtCounter.Add(float64(j))
+			rtCounter.Add(int64(j))
 		}
 	}
 	shedder := &adaptiveShedder{
 		passCounter:     passCounter,
 		rtCounter:       rtCounter,
-		windows:         buckets,
+		windowScale:     windowFactor,
 		overloadTime:    syncx.NewAtomicDuration(),
 		droppedRecently: syncx.NewAtomicBool(),
 	}
@@ -149,7 +150,8 @@ func TestAdaptiveShedderShouldDrop(t *testing.T) {
 
 	// cpu >=  800, inflight > maxPass
 	shedder.avgFlying = 80
-	shedder.flying = 50
+	// because of the overloadFactor, so we need to make sure maxFlight is greater than flying
+	shedder.flying = int64(shedder.maxFlight()*shedder.overloadFactor()) - 5
 	assert.False(t, shedder.shouldDrop())
 
 	// cpu >=  800, inflight > maxPass
@@ -182,15 +184,15 @@ func TestAdaptiveShedderStillHot(t *testing.T) {
 		if i > 0 {
 			time.Sleep(bucketDuration)
 		}
-		passCounter.Add(float64((i + 1) * 100))
+		passCounter.Add(int64((i + 1) * 100))
 		for j := i*10 + 1; j <= i*10+10; j++ {
-			rtCounter.Add(float64(j))
+			rtCounter.Add(int64(j))
 		}
 	}
 	shedder := &adaptiveShedder{
 		passCounter:     passCounter,
 		rtCounter:       rtCounter,
-		windows:         buckets,
+		windowScale:     windowFactor,
 		overloadTime:    syncx.NewAtomicDuration(),
 		droppedRecently: syncx.ForAtomicBool(true),
 	}
@@ -239,6 +241,32 @@ func BenchmarkAdaptiveShedder_Allow(b *testing.B) {
 	b.Run("low load", bench)
 }
 
-func newRollingWindow() *collection.RollingWindow {
-	return collection.NewRollingWindow(buckets, bucketDuration, collection.IgnoreCurrentBucket())
+func BenchmarkMaxFlight(b *testing.B) {
+	passCounter := newRollingWindow()
+	rtCounter := newRollingWindow()
+	for i := 0; i < 10; i++ {
+		if i > 0 {
+			time.Sleep(bucketDuration)
+		}
+		passCounter.Add(int64((i + 1) * 100))
+		for j := i*10 + 1; j <= i*10+10; j++ {
+			rtCounter.Add(int64(j))
+		}
+	}
+	shedder := &adaptiveShedder{
+		passCounter:     passCounter,
+		rtCounter:       rtCounter,
+		windowScale:     windowFactor,
+		droppedRecently: syncx.NewAtomicBool(),
+	}
+
+	for i := 0; i < b.N; i++ {
+		_ = shedder.maxFlight()
+	}
+}
+
+func newRollingWindow() *collection.RollingWindow[int64, *collection.Bucket[int64]] {
+	return collection.NewRollingWindow[int64, *collection.Bucket[int64]](func() *collection.Bucket[int64] {
+		return new(collection.Bucket[int64])
+	}, buckets, bucketDuration, collection.IgnoreCurrentBucket[int64, *collection.Bucket[int64]]())
 }

core/load/sheddergroup.go

@@ -6,7 +6,7 @@ import (
 	"github.com/zeromicro/go-zero/core/syncx"
 )
 
-// A ShedderGroup is a manager to manage key based shedders.
+// A ShedderGroup is a manager to manage key-based shedders.
 type ShedderGroup struct {
 	options []ShedderOption
 	manager *syncx.ResourceManager

core/logc/logs.go

@@ -37,12 +37,19 @@ func Debugf(ctx context.Context, format string, v ...interface{}) {
 	getLogger(ctx).Debugf(format, v...)
 }
 
+// Debugfn writes fn result into access log.
+// This is useful when the function is expensive to compute,
+// and we want to log it only when necessary.
+func Debugfn(ctx context.Context, fn func() any) {
+	getLogger(ctx).Debugfn(fn)
+}
+
 // Debugv writes v into access log with json content.
 func Debugv(ctx context.Context, v interface{}) {
 	getLogger(ctx).Debugv(v)
 }
 
-// Debugw writes msg along with fields into access log.
+// Debugw writes msg along with fields into the access log.
 func Debugw(ctx context.Context, msg string, fields ...LogField) {
 	getLogger(ctx).Debugw(msg, fields...)
 }
@@ -57,13 +64,20 @@ func Errorf(ctx context.Context, format string, v ...any) {
 	getLogger(ctx).Errorf(fmt.Errorf(format, v...).Error())
 }
 
+// Errorfn writes fn result into error log.
+// This is useful when the function is expensive to compute,
+// and we want to log it only when necessary.
+func Errorfn(ctx context.Context, fn func() any) {
+	getLogger(ctx).Errorfn(fn)
+}
+
 // Errorv writes v into error log with json content.
 // No call stack attached, because not elegant to pack the messages.
 func Errorv(ctx context.Context, v any) {
 	getLogger(ctx).Errorv(v)
 }
 
-// Errorw writes msg along with fields into error log.
+// Errorw writes msg along with fields into the error log.
 func Errorw(ctx context.Context, msg string, fields ...LogField) {
 	getLogger(ctx).Errorw(msg, fields...)
 }
@@ -83,12 +97,19 @@ func Infof(ctx context.Context, format string, v ...any) {
 	getLogger(ctx).Infof(format, v...)
 }
 
+// Infofn writes fn result into access log.
+// This is useful when the function is expensive to compute,
+// and we want to log it only when necessary.
+func Infofn(ctx context.Context, fn func() any) {
+	getLogger(ctx).Infofn(fn)
+}
+
 // Infov writes v into access log with json content.
 func Infov(ctx context.Context, v any) {
 	getLogger(ctx).Infov(v)
 }
 
-// Infow writes msg along with fields into access log.
+// Infow writes msg along with fields into the access log.
 func Infow(ctx context.Context, msg string, fields ...LogField) {
 	getLogger(ctx).Infow(msg, fields...)
 }
@@ -108,10 +129,11 @@ func SetLevel(level uint32) {
 	logx.SetLevel(level)
 }
 
-// SetUp sets up the logx. If already set up, just return nil.
-// we allow SetUp to be called multiple times, because for example
+// SetUp sets up the logx.
+// If already set up, return nil.
+// We allow SetUp to be called multiple times, because, for example,
 // we need to allow different service frameworks to initialize logx respectively.
-// the same logic for SetUp
+// The same logic for SetUp
 func SetUp(c LogConf) error {
 	return logx.SetUp(c)
 }
@@ -126,6 +148,13 @@ func Slowf(ctx context.Context, format string, v ...any) {
 	getLogger(ctx).Slowf(format, v...)
 }
 
+// Slowfn writes fn result into slow log.
+// This is useful when the function is expensive to compute,
+// and we want to log it only when necessary.
+func Slowfn(ctx context.Context, fn func() any) {
+	getLogger(ctx).Slowfn(fn)
+}
+
 // Slowv writes v into slow log with json content.
 func Slowv(ctx context.Context, v any) {
 	getLogger(ctx).Slowv(v)

core/logc/logs_test.go

@@ -49,6 +49,15 @@ func TestErrorf(t *testing.T) {
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
+func TestErrorfn(t *testing.T) {
+	buf := logtest.NewCollector(t)
+	file, line := getFileLine()
+	Errorfn(context.Background(), func() any {
+		return fmt.Sprintf("foo %s", "bar")
+	})
+	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
+}
+
 func TestErrorv(t *testing.T) {
 	buf := logtest.NewCollector(t)
 	file, line := getFileLine()
@@ -77,6 +86,15 @@ func TestInfof(t *testing.T) {
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
+func TestInfofn(t *testing.T) {
+	buf := logtest.NewCollector(t)
+	file, line := getFileLine()
+	Infofn(context.Background(), func() any {
+		return fmt.Sprintf("foo %s", "bar")
+	})
+	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
+}
+
 func TestInfov(t *testing.T) {
 	buf := logtest.NewCollector(t)
 	file, line := getFileLine()
@@ -105,6 +123,15 @@ func TestDebugf(t *testing.T) {
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
 }
 
+func TestDebugfn(t *testing.T) {
+	buf := logtest.NewCollector(t)
+	file, line := getFileLine()
+	Debugfn(context.Background(), func() any {
+		return fmt.Sprintf("foo %s", "bar")
+	})
+	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)))
+}
+
 func TestDebugv(t *testing.T) {
 	buf := logtest.NewCollector(t)
 	file, line := getFileLine()
@@ -148,6 +175,15 @@ func TestSlowf(t *testing.T) {
 	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)), buf.String())
 }
 
+func TestSlowfn(t *testing.T) {
+	buf := logtest.NewCollector(t)
+	file, line := getFileLine()
+	Slowfn(context.Background(), func() any {
+		return fmt.Sprintf("foo %s", "bar")
+	})
+	assert.True(t, strings.Contains(buf.String(), fmt.Sprintf("%s:%d", file, line+1)), buf.String())
+}
+
 func TestSlowv(t *testing.T) {
 	buf := logtest.NewCollector(t)
 	file, line := getFileLine()

core/logx/config.go

@@ -1,45 +1,70 @@
 package logx
 
-// A LogConf is a logging config.
-type LogConf struct {
-	// ServiceName represents the service name.
-	ServiceName string `json:",optional"`
-	// Mode represents the logging mode, default is `console`.
-	// console: log to console.
-	// file: log to file.
-	// volume: used in k8s, prepend the hostname to the log file name.
-	Mode string `json:",default=console,options=[console,file,volume]"`
-	// Encoding represents the encoding type, default is `json`.
-	// json: json encoding.
-	// plain: plain text encoding, typically used in development.
-	Encoding string `json:",default=json,options=[json,plain]"`
-	// TimeFormat represents the time format, default is `2006-01-02T15:04:05.000Z07:00`.
-	TimeFormat string `json:",optional"`
-	// Path represents the log file path, default is `logs`.
-	Path string `json:",default=logs"`
-	// Level represents the log level, default is `info`.
-	Level string `json:",default=info,options=[debug,info,error,severe]"`
-	// MaxContentLength represents the max content bytes, default is no limit.
-	MaxContentLength uint32 `json:",optional"`
-	// Compress represents whether to compress the log file, default is `false`.
-	Compress bool `json:",optional"`
-	// Stat represents whether to log statistics, default is `true`.
-	Stat bool `json:",default=true"`
-	// KeepDays represents how many days the log files will be kept. Default to keep all files.
-	// Only take effect when Mode is `file` or `volume`, both work when Rotation is `daily` or `size`.
-	KeepDays int `json:",optional"`
-	// StackCooldownMillis represents the cooldown time for stack logging, default is 100ms.
-	StackCooldownMillis int `json:",default=100"`
-	// MaxBackups represents how many backup log files will be kept. 0 means all files will be kept forever.
-	// Only take effect when RotationRuleType is `size`.
-	// Even though `MaxBackups` sets 0, log files will still be removed
-	// if the `KeepDays` limitation is reached.
-	MaxBackups int `json:",default=0"`
-	// MaxSize represents how much space the writing log file takes up. 0 means no limit. The unit is `MB`.
-	// Only take effect when RotationRuleType is `size`
-	MaxSize int `json:",default=0"`
-	// Rotation represents the type of log rotation rule. Default is `daily`.
-	// daily: daily rotation.
-	// size: size limited rotation.
-	Rotation string `json:",default=daily,options=[daily,size]"`
-}
+type (
+	// A LogConf is a logging config.
+	LogConf struct {
+		// ServiceName represents the service name.
+		ServiceName string `json:",optional"`
+		// Mode represents the logging mode, default is `console`.
+		// console: log to console.
+		// file: log to file.
+		// volume: used in k8s, prepend the hostname to the log file name.
+		Mode string `json:",default=console,options=[console,file,volume]"`
+		// Encoding represents the encoding type, default is `json`.
+		// json: json encoding.
+		// plain: plain text encoding, typically used in development.
+		Encoding string `json:",default=json,options=[json,plain]"`
+		// TimeFormat represents the time format, default is `2006-01-02T15:04:05.000Z07:00`.
+		TimeFormat string `json:",optional"`
+		// Path represents the log file path, default is `logs`.
+		Path string `json:",default=logs"`
+		// Level represents the log level, default is `info`.
+		Level string `json:",default=info,options=[debug,info,error,severe]"`
+		// MaxContentLength represents the max content bytes, default is no limit.
+		MaxContentLength uint32 `json:",optional"`
+		// Compress represents whether to compress the log file, default is `false`.
+		Compress bool `json:",optional"`
+		// Stat represents whether to log statistics, default is `true`.
+		Stat bool `json:",default=true"`
+		// KeepDays represents how many days the log files will be kept. Default to keep all files.
+		// Only take effect when Mode is `file` or `volume`, both work when Rotation is `daily` or `size`.
+		KeepDays int `json:",optional"`
+		// StackCooldownMillis represents the cooldown time for stack logging, default is 100ms.
+		StackCooldownMillis int `json:",default=100"`
+		// MaxBackups represents how many backup log files will be kept. 0 means all files will be kept forever.
+		// Only take effect when RotationRuleType is `size`.
+		// Even though `MaxBackups` sets 0, log files will still be removed
+		// if the `KeepDays` limitation is reached.
+		MaxBackups int `json:",default=0"`
+		// MaxSize represents how much space the writing log file takes up. 0 means no limit. The unit is `MB`.
+		// Only take effect when RotationRuleType is `size`
+		MaxSize int `json:",default=0"`
+		// Rotation represents the type of log rotation rule. Default is `daily`.
+		// daily: daily rotation.
+		// size: size limited rotation.
+		Rotation string `json:",default=daily,options=[daily,size]"`
+		// FileTimeFormat represents the time format for file name, default is `2006-01-02T15:04:05.000Z07:00`.
+		FileTimeFormat string `json:",optional"`
+		// FieldKeys represents the field keys.
+		FieldKeys fieldKeyConf `json:",optional"`
+	}
+
+	fieldKeyConf struct {
+		// CallerKey represents the caller key.
+		CallerKey string `json:",default=caller"`
+		// ContentKey represents the content key.
+		ContentKey string `json:",default=content"`
+		// DurationKey represents the duration key.
+		DurationKey string `json:",default=duration"`
+		// LevelKey represents the level key.
+		LevelKey string `json:",default=level"`
+		// SpanKey represents the span key.
+		SpanKey string `json:",default=span"`
+		// TimestampKey represents the timestamp key.
+		TimestampKey string `json:",default=@timestamp"`
+		// TraceKey represents the trace key.
+		TraceKey string `json:",default=trace"`
+		// TruncatedKey represents the truncated key.
+		TruncatedKey string `json:",default=truncated"`
+	}
+)

core/logx/fields.go

@@ -7,12 +7,11 @@ import (
 )
 
 var (
-	fieldsContextKey contextKey
 	globalFields     atomic.Value
 	globalFieldsLock sync.Mutex
 )
 
-type contextKey struct{}
+type fieldsKey struct{}
 
 // AddGlobalFields adds global fields.
 func AddGlobalFields(fields ...LogField) {
@@ -29,16 +28,16 @@ func AddGlobalFields(fields ...LogField) {
 
 // ContextWithFields returns a new context with the given fields.
 func ContextWithFields(ctx context.Context, fields ...LogField) context.Context {
-	if val := ctx.Value(fieldsContextKey); val != nil {
+	if val := ctx.Value(fieldsKey{}); val != nil {
 		if arr, ok := val.([]LogField); ok {
 			allFields := make([]LogField, 0, len(arr)+len(fields))
 			allFields = append(allFields, arr...)
 			allFields = append(allFields, fields...)
-			return context.WithValue(ctx, fieldsContextKey, allFields)
+			return context.WithValue(ctx, fieldsKey{}, allFields)
 		}
 	}
 
-	return context.WithValue(ctx, fieldsContextKey, fields)
+	return context.WithValue(ctx, fieldsKey{}, fields)
 }
 
 // WithFields returns a new logger with the given fields.

core/logx/fields_test.go

@@ -34,7 +34,7 @@ func TestAddGlobalFields(t *testing.T) {
 
 func TestContextWithFields(t *testing.T) {
 	ctx := ContextWithFields(context.Background(), Field("a", 1), Field("b", 2))
-	vals := ctx.Value(fieldsContextKey)
+	vals := ctx.Value(fieldsKey{})
 	assert.NotNil(t, vals)
 	fields, ok := vals.([]LogField)
 	assert.True(t, ok)
@@ -43,7 +43,7 @@ func TestContextWithFields(t *testing.T) {
 
 func TestWithFields(t *testing.T) {
 	ctx := WithFields(context.Background(), Field("a", 1), Field("b", 2))
-	vals := ctx.Value(fieldsContextKey)
+	vals := ctx.Value(fieldsKey{})
 	assert.NotNil(t, vals)
 	fields, ok := vals.([]LogField)
 	assert.True(t, ok)
@@ -55,7 +55,7 @@ func TestWithFieldsAppend(t *testing.T) {
 	ctx := context.WithValue(context.Background(), dummyKey, "dummy")
 	ctx = ContextWithFields(ctx, Field("a", 1), Field("b", 2))
 	ctx = ContextWithFields(ctx, Field("c", 3), Field("d", 4))
-	vals := ctx.Value(fieldsContextKey)
+	vals := ctx.Value(fieldsKey{})
 	assert.NotNil(t, vals)
 	fields, ok := vals.([]LogField)
 	assert.True(t, ok)
@@ -80,8 +80,8 @@ func TestWithFieldsAppendCopy(t *testing.T) {
 	ctxa := ContextWithFields(ctx, af)
 	ctxb := ContextWithFields(ctx, bf)
 
-	assert.EqualValues(t, af, ctxa.Value(fieldsContextKey).([]LogField)[count])
-	assert.EqualValues(t, bf, ctxb.Value(fieldsContextKey).([]LogField)[count])
+	assert.EqualValues(t, af, ctxa.Value(fieldsKey{}).([]LogField)[count])
+	assert.EqualValues(t, bf, ctxb.Value(fieldsKey{}).([]LogField)[count])
 }
 
 func BenchmarkAtomicValue(b *testing.B) {

core/logx/lesslogger.go

@@ -1,6 +1,6 @@
 package logx
 
-// A LessLogger is a logger that control to log once during the given duration.
+// A LessLogger is a logger that controls to log once during the given duration.
 type LessLogger struct {
 	*limitedExecutor
 }

core/logx/logger.go

@@ -7,18 +7,22 @@ import (
 
 // A Logger represents a logger.
 type Logger interface {
-	// Debug logs a message at info level.
+	// Debug logs a message at debug level.
 	Debug(...any)
-	// Debugf logs a message at info level.
+	// Debugf logs a message at debug level.
 	Debugf(string, ...any)
-	// Debugv logs a message at info level.
+	// Debugfn logs a message at debug level.
+	Debugfn(func() any)
+	// Debugv logs a message at debug level.
 	Debugv(any)
-	// Debugw logs a message at info level.
+	// Debugw logs a message at debug level.
 	Debugw(string, ...LogField)
 	// Error logs a message at error level.
 	Error(...any)
 	// Errorf logs a message at error level.
 	Errorf(string, ...any)
+	// Errorfn logs a message at error level.
+	Errorfn(func() any)
 	// Errorv logs a message at error level.
 	Errorv(any)
 	// Errorw logs a message at error level.
@@ -27,6 +31,8 @@ type Logger interface {
 	Info(...any)
 	// Infof logs a message at info level.
 	Infof(string, ...any)
+	// Infofn logs a message at info level.
+	Infofn(func() any)
 	// Infov logs a message at info level.
 	Infov(any)
 	// Infow logs a message at info level.
@@ -35,6 +41,8 @@ type Logger interface {
 	Slow(...any)
 	// Slowf logs a message at slow level.
 	Slowf(string, ...any)
+	// Slowfn logs a message at slow level.
+	Slowfn(func() any)
 	// Slowv logs a message at slow level.
 	Slowv(any)
 	// Sloww logs a message at slow level.

core/logx/logs.go

@@ -6,10 +6,10 @@ import (
 	"log"
 	"os"
 	"path"
+	"reflect"
 	"runtime/debug"
 	"sync"
 	"sync/atomic"
-	"time"
 
 	"github.com/zeromicro/go-zero/core/sysx"
 )
@@ -17,14 +17,13 @@ import (
 const callerDepth = 4
 
 var (
-	timeFormat = "2006-01-02T15:04:05.000Z07:00"
-	logLevel   uint32
+	timeFormat        = "2006-01-02T15:04:05.000Z07:00"
 	encoding   uint32 = jsonEncodingType
 	// maxContentLength is used to truncate the log content, 0 for not truncating.
 	maxContentLength uint32
 	// use uint32 for atomic operations
-	disableLog  uint32
 	disableStat uint32
+	logLevel    uint32
 	options     logOptions
 	writer      = new(atomicWriter)
 	setupOnce   sync.Once
@@ -52,6 +51,26 @@ type (
 	}
 )
 
+// AddWriter adds a new writer.
+// If there is already a writer, the new writer will be added to the writer chain.
+// For example, to write logs to both file and console, if there is already a file writer,
+// ```go
+// logx.AddWriter(logx.NewWriter(os.Stdout))
+// ```
+func AddWriter(w Writer) {
+	ow := Reset()
+	if ow == nil {
+		SetWriter(w)
+	} else {
+		// no need to check if the existing writer is a comboWriter,
+		// because it is not common to add more than one writer.
+		// even more than one writer, the behavior is the same.
+		SetWriter(comboWriter{
+			writers: []Writer{ow, w},
+		})
+	}
+}
+
 // Alert alerts v in alert level, and the message is written to error log.
 func Alert(v string) {
 	getWriter().Alert(v)
@@ -80,6 +99,14 @@ func Debugf(format string, v ...any) {
 	}
 }
 
+// Debugfn writes function result into access log if debug level enabled.
+// This is useful when the function is expensive to call and debug level disabled.
+func Debugfn(fn func() any) {
+	if shallLog(DebugLevel) {
+		writeDebug(fn())
+	}
+}
+
 // Debugv writes v into access log with json content.
 func Debugv(v any) {
 	if shallLog(DebugLevel) {
@@ -87,7 +114,7 @@ func Debugv(v any) {
 	}
 }
 
-// Debugw writes msg along with fields into access log.
+// Debugw writes msg along with fields into the access log.
 func Debugw(msg string, fields ...LogField) {
 	if shallLog(DebugLevel) {
 		writeDebug(msg, fields...)
@@ -96,7 +123,7 @@ func Debugw(msg string, fields ...LogField) {
 
 // Disable disables the logging.
 func Disable() {
-	atomic.StoreUint32(&disableLog, 1)
+	atomic.StoreUint32(&logLevel, disableLevel)
 	writer.Store(nopWriter{})
 }
 
@@ -119,6 +146,13 @@ func Errorf(format string, v ...any) {
 	}
 }
 
+// Errorfn writes function result into error log.
+func Errorfn(fn func() any) {
+	if shallLog(ErrorLevel) {
+		writeError(fn())
+	}
+}
+
 // ErrorStack writes v along with call stack into error log.
 func ErrorStack(v ...any) {
 	if shallLog(ErrorLevel) {
@@ -143,7 +177,7 @@ func Errorv(v any) {
 	}
 }
 
-// Errorw writes msg along with fields into error log.
+// Errorw writes msg along with fields into the error log.
 func Errorw(msg string, fields ...LogField) {
 	if shallLog(ErrorLevel) {
 		writeError(msg, fields...)
@@ -152,39 +186,9 @@ func Errorw(msg string, fields ...LogField) {
 
 // Field returns a LogField for the given key and value.
 func Field(key string, value any) LogField {
-	switch val := value.(type) {
-	case error:
-		return LogField{Key: key, Value: val.Error()}
-	case []error:
-		var errs []string
-		for _, err := range val {
-			errs = append(errs, err.Error())
-		}
-		return LogField{Key: key, Value: errs}
-	case time.Duration:
-		return LogField{Key: key, Value: fmt.Sprint(val)}
-	case []time.Duration:
-		var durs []string
-		for _, dur := range val {
-			durs = append(durs, fmt.Sprint(dur))
-		}
-		return LogField{Key: key, Value: durs}
-	case []time.Time:
-		var times []string
-		for _, t := range val {
-			times = append(times, fmt.Sprint(t))
-		}
-		return LogField{Key: key, Value: times}
-	case fmt.Stringer:
-		return LogField{Key: key, Value: val.String()}
-	case []fmt.Stringer:
-		var strs []string
-		for _, str := range val {
-			strs = append(strs, str.String())
-		}
-		return LogField{Key: key, Value: strs}
-	default:
-		return LogField{Key: key, Value: val}
+	return LogField{
+		Key:   key,
+		Value: value,
 	}
 }
 
@@ -202,6 +206,14 @@ func Infof(format string, v ...any) {
 	}
 }
 
+// Infofn writes function result into access log.
+// This is useful when the function is expensive to call and info level disabled.
+func Infofn(fn func() any) {
+	if shallLog(InfoLevel) {
+		writeInfo(fn())
+	}
+}
+
 // Infov writes v into access log with json content.
 func Infov(v any) {
 	if shallLog(InfoLevel) {
@@ -209,7 +221,7 @@ func Infov(v any) {
 	}
 }
 
-// Infow writes msg along with fields into access log.
+// Infow writes msg along with fields into the access log.
 func Infow(msg string, fields ...LogField) {
 	if shallLog(InfoLevel) {
 		writeInfo(msg, fields...)
@@ -250,20 +262,22 @@ func SetLevel(level uint32) {
 
 // SetWriter sets the logging writer. It can be used to customize the logging.
 func SetWriter(w Writer) {
-	if atomic.LoadUint32(&disableLog) == 0 {
+	if atomic.LoadUint32(&logLevel) != disableLevel {
 		writer.Store(w)
 	}
 }
 
-// SetUp sets up the logx. If already set up, just return nil.
-// we allow SetUp to be called multiple times, because for example
+// SetUp sets up the logx.
+// If already set up, return nil.
+// We allow SetUp to be called multiple times, because, for example,
 // we need to allow different service frameworks to initialize logx respectively.
 func SetUp(c LogConf) (err error) {
-	// Just ignore the subsequent SetUp calls.
+	// Ignore the later SetUp calls.
 	// Because multiple services in one process might call SetUp respectively.
 	// Need to wait for the first caller to complete the execution.
 	setupOnce.Do(func() {
-		setupLogLevel(c)
+		setupLogLevel(c.Level)
+		setupFieldKeys(c.FieldKeys)
 
 		if !c.Stat {
 			DisableStat()
@@ -273,6 +287,10 @@ func SetUp(c LogConf) (err error) {
 			timeFormat = c.TimeFormat
 		}
 
+		if len(c.FileTimeFormat) > 0 {
+			fileTimeFormat = c.FileTimeFormat
+		}
+
 		atomic.StoreUint32(&maxContentLength, c.MaxContentLength)
 
 		switch c.Encoding {
@@ -323,6 +341,14 @@ func Slowf(format string, v ...any) {
 	}
 }
 
+// Slowfn writes function result into slow log.
+// This is useful when the function is expensive to call and slow level disabled.
+func Slowfn(fn func() any) {
+	if shallLog(ErrorLevel) {
+		writeSlow(fn())
+	}
+}
+
 // Slowv writes v into slow log with json content.
 func Slowv(v any) {
 	if shallLog(ErrorLevel) {
@@ -414,6 +440,32 @@ func createOutput(path string) (io.WriteCloser, error) {
 	return NewLogger(path, rule, options.gzipEnabled)
 }
 
+func encodeError(err error) (ret string) {
+	return encodeWithRecover(err, func() string {
+		return err.Error()
+	})
+}
+
+func encodeStringer(v fmt.Stringer) (ret string) {
+	return encodeWithRecover(v, func() string {
+		return v.String()
+	})
+}
+
+func encodeWithRecover(arg any, fn func() string) (ret string) {
+	defer func() {
+		if err := recover(); err != nil {
+			if v := reflect.ValueOf(arg); v.Kind() == reflect.Ptr && v.IsNil() {
+				ret = nilAngleString
+			} else {
+				ret = fmt.Sprintf("panic: %v", err)
+			}
+		}
+	}()
+
+	return fn()
+}
+
 func getWriter() Writer {
 	w := writer.Load()
 	if w == nil {
@@ -429,8 +481,35 @@ func handleOptions(opts []LogOption) {
 	}
 }
 
-func setupLogLevel(c LogConf) {
-	switch c.Level {
+func setupFieldKeys(c fieldKeyConf) {
+	if len(c.CallerKey) > 0 {
+		callerKey = c.CallerKey
+	}
+	if len(c.ContentKey) > 0 {
+		contentKey = c.ContentKey
+	}
+	if len(c.DurationKey) > 0 {
+		durationKey = c.DurationKey
+	}
+	if len(c.LevelKey) > 0 {
+		levelKey = c.LevelKey
+	}
+	if len(c.SpanKey) > 0 {
+		spanKey = c.SpanKey
+	}
+	if len(c.TimestampKey) > 0 {
+		timestampKey = c.TimestampKey
+	}
+	if len(c.TraceKey) > 0 {
+		traceKey = c.TraceKey
+	}
+	if len(c.TruncatedKey) > 0 {
+		truncatedKey = c.TruncatedKey
+	}
+}
+
+func setupLogLevel(level string) {
+	switch level {
 	case levelDebug:
 		SetLevel(DebugLevel)
 	case levelInfo:
@@ -478,15 +557,15 @@ func shallLogStat() bool {
 // If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
 // The caller should check shallLog before calling this function.
 func writeDebug(val any, fields ...LogField) {
-	getWriter().Debug(val, addCaller(fields...)...)
+	getWriter().Debug(val, mergeGlobalFields(addCaller(fields...))...)
 }
 
-// writeError writes v into error log.
+// writeError writes v into the error log.
 // Not checking shallLog here is for performance consideration.
 // If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
 // The caller should check shallLog before calling this function.
 func writeError(val any, fields ...LogField) {
-	getWriter().Error(val, addCaller(fields...)...)
+	getWriter().Error(val, mergeGlobalFields(addCaller(fields...))...)
 }
 
 // writeInfo writes v into info log.
@@ -494,7 +573,7 @@ func writeError(val any, fields ...LogField) {
 // If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
 // The caller should check shallLog before calling this function.
 func writeInfo(val any, fields ...LogField) {
-	getWriter().Info(val, addCaller(fields...)...)
+	getWriter().Info(val, mergeGlobalFields(addCaller(fields...))...)
 }
 
 // writeSevere writes v into severe log.
@@ -510,7 +589,7 @@ func writeSevere(msg string) {
 // If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
 // The caller should check shallLog before calling this function.
 func writeSlow(val any, fields ...LogField) {
-	getWriter().Slow(val, addCaller(fields...)...)
+	getWriter().Slow(val, mergeGlobalFields(addCaller(fields...))...)
 }
 
 // writeStack writes v into stack log.
@@ -521,10 +600,10 @@ func writeStack(msg string) {
 	getWriter().Stack(fmt.Sprintf("%s\n%s", msg, string(debug.Stack())))
 }
 
-// writeStat writes v into stat log.
+// writeStat writes v into the stat log.
 // Not checking shallLog here is for performance consideration.
 // If we check shallLog here, the fmt.Sprint might be called even if the log level is not enabled.
 // The caller should check shallLog before calling this function.
 func writeStat(msg string) {
-	getWriter().Stat(msg, addCaller()...)
+	getWriter().Stat(msg, mergeGlobalFields(addCaller())...)
 }

core/logx/logs_test.go

@@ -1,6 +1,7 @@
 package logx
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -16,6 +17,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/sdk/trace"
 )
 
 var (
@@ -248,6 +251,32 @@ func TestStructedLogDebugf(t *testing.T) {
 	})
 }
 
+func TestStructedLogDebugfn(t *testing.T) {
+	t.Run("debugfn with output", func(t *testing.T) {
+		w := new(mockWriter)
+		old := writer.Swap(w)
+		defer writer.Store(old)
+
+		doTestStructedLog(t, levelDebug, w, func(v ...any) {
+			Debugfn(func() any {
+				return fmt.Sprint(v...)
+			})
+		})
+	})
+
+	t.Run("debugfn without output", func(t *testing.T) {
+		w := new(mockWriter)
+		old := writer.Swap(w)
+		defer writer.Store(old)
+
+		doTestStructedLogEmpty(t, w, InfoLevel, func(v ...any) {
+			Debugfn(func() any {
+				return fmt.Sprint(v...)
+			})
+		})
+	})
+}
+
 func TestStructedLogDebugv(t *testing.T) {
 	w := new(mockWriter)
 	old := writer.Swap(w)
@@ -288,6 +317,32 @@ func TestStructedLogErrorf(t *testing.T) {
 	})
 }
 
+func TestStructedLogErrorfn(t *testing.T) {
+	t.Run("errorfn with output", func(t *testing.T) {
+		w := new(mockWriter)
+		old := writer.Swap(w)
+		defer writer.Store(old)
+
+		doTestStructedLog(t, levelError, w, func(v ...any) {
+			Errorfn(func() any {
+				return fmt.Sprint(v...)
+			})
+		})
+	})
+
+	t.Run("errorfn without output", func(t *testing.T) {
+		w := new(mockWriter)
+		old := writer.Swap(w)
+		defer writer.Store(old)
+
+		doTestStructedLogEmpty(t, w, SevereLevel, func(v ...any) {
+			Errorfn(func() any {
+				return fmt.Sprint(v...)
+			})
+		})
+	})
+}
+
 func TestStructedLogErrorv(t *testing.T) {
 	w := new(mockWriter)
 	old := writer.Swap(w)
@@ -328,6 +383,32 @@ func TestStructedLogInfof(t *testing.T) {
 	})
 }
 
+func TestStructedInfofn(t *testing.T) {
+	t.Run("infofn with output", func(t *testing.T) {
+		w := new(mockWriter)
+		old := writer.Swap(w)
+		defer writer.Store(old)
+
+		doTestStructedLog(t, levelInfo, w, func(v ...any) {
+			Infofn(func() any {
+				return fmt.Sprint(v...)
+			})
+		})
+	})
+
+	t.Run("infofn without output", func(t *testing.T) {
+		w := new(mockWriter)
+		old := writer.Swap(w)
+		defer writer.Store(old)
+
+		doTestStructedLogEmpty(t, w, ErrorLevel, func(v ...any) {
+			Infofn(func() any {
+				return fmt.Sprint(v...)
+			})
+		})
+	})
+}
+
 func TestStructedLogInfov(t *testing.T) {
 	w := new(mockWriter)
 	old := writer.Swap(w)
@@ -348,6 +429,27 @@ func TestStructedLogInfow(t *testing.T) {
 	})
 }
 
+func TestStructedLogFieldNil(t *testing.T) {
+	w := new(mockWriter)
+	old := writer.Swap(w)
+	defer writer.Store(old)
+
+	assert.NotPanics(t, func() {
+		var s *string
+		Infow("test", Field("bb", s))
+		var d *nilStringer
+		Infow("test", Field("bb", d))
+		var e *nilError
+		Errorw("test", Field("bb", e))
+	})
+	assert.NotPanics(t, func() {
+		var p panicStringer
+		Infow("test", Field("bb", p))
+		var ps innerPanicStringer
+		Infow("test", Field("bb", ps))
+	})
+}
+
 func TestStructedLogInfoConsoleAny(t *testing.T) {
 	w := new(mockWriter)
 	old := writer.Swap(w)
@@ -430,6 +532,17 @@ func TestStructedLogInfoConsoleText(t *testing.T) {
 	})
 }
 
+func TestInfofnWithErrorLevel(t *testing.T) {
+	called := false
+	SetLevel(ErrorLevel)
+	defer SetLevel(DebugLevel)
+	Infofn(func() any {
+		called = true
+		return "info log"
+	})
+	assert.False(t, called)
+}
+
 func TestStructedLogSlow(t *testing.T) {
 	w := new(mockWriter)
 	old := writer.Swap(w)
@@ -450,6 +563,32 @@ func TestStructedLogSlowf(t *testing.T) {
 	})
 }
 
+func TestStructedLogSlowfn(t *testing.T) {
+	t.Run("slowfn with output", func(t *testing.T) {
+		w := new(mockWriter)
+		old := writer.Swap(w)
+		defer writer.Store(old)
+
+		doTestStructedLog(t, levelSlow, w, func(v ...any) {
+			Slowfn(func() any {
+				return fmt.Sprint(v...)
+			})
+		})
+	})
+
+	t.Run("slowfn without output", func(t *testing.T) {
+		w := new(mockWriter)
+		old := writer.Swap(w)
+		defer writer.Store(old)
+
+		doTestStructedLogEmpty(t, w, SevereLevel, func(v ...any) {
+			Slowfn(func() any {
+				return fmt.Sprint(v...)
+			})
+		})
+	})
+}
+
 func TestStructedLogSlowv(t *testing.T) {
 	w := new(mockWriter)
 	old := writer.Swap(w)
@@ -570,7 +709,7 @@ func TestErrorfWithWrappedError(t *testing.T) {
 	old := writer.Swap(w)
 	defer writer.Store(old)
 
-	Errorf("hello %w", errors.New(message))
+	Errorf("hello %s", errors.New(message))
 	assert.True(t, strings.Contains(w.String(), "hello there"))
 }
 
@@ -640,15 +779,9 @@ func TestSetup(t *testing.T) {
 		MaxBackups:  3,
 		MaxSize:     1024 * 1024,
 	}))
-	setupLogLevel(LogConf{
-		Level: levelInfo,
-	})
-	setupLogLevel(LogConf{
-		Level: levelError,
-	})
-	setupLogLevel(LogConf{
-		Level: levelSevere,
-	})
+	setupLogLevel(levelInfo)
+	setupLogLevel(levelError)
+	setupLogLevel(levelSevere)
 	_, err := createOutput("")
 	assert.NotNil(t, err)
 	Disable()
@@ -658,6 +791,10 @@ func TestSetup(t *testing.T) {
 
 func TestDisable(t *testing.T) {
 	Disable()
+	defer func() {
+		SetLevel(InfoLevel)
+		atomic.StoreUint32(&encoding, jsonEncodingType)
+	}()
 
 	var opt logOptions
 	WithKeepDays(1)(&opt)
@@ -666,6 +803,7 @@ func TestDisable(t *testing.T) {
 	WithMaxSize(1024)(&opt)
 	assert.Nil(t, Close())
 	assert.Nil(t, Close())
+	assert.Equal(t, uint32(disableLevel), atomic.LoadUint32(&logLevel))
 }
 
 func TestDisableStat(t *testing.T) {
@@ -679,8 +817,19 @@ func TestDisableStat(t *testing.T) {
 	assert.Equal(t, 0, w.builder.Len())
 }
 
+func TestAddWriter(t *testing.T) {
+	const message = "hello there"
+	w := new(mockWriter)
+	AddWriter(w)
+	w1 := new(mockWriter)
+	AddWriter(w1)
+	Error(message)
+	assert.Contains(t, w.String(), message)
+	assert.Contains(t, w1.String(), message)
+}
+
 func TestSetWriter(t *testing.T) {
-	atomic.StoreUint32(&disableLog, 0)
+	atomic.StoreUint32(&logLevel, 0)
 	Reset()
 	SetWriter(nopWriter{})
 	assert.NotNil(t, writer.Load())
@@ -704,6 +853,95 @@ func TestWithKeepDays(t *testing.T) {
 	assert.Equal(t, 1, opt.keepDays)
 }
 
+func TestWithField_LogLevel(t *testing.T) {
+	tests := []struct {
+		name  string
+		level uint32
+		fn    func(string, ...LogField)
+		count int32
+	}{
+		{
+			name:  "debug/info",
+			level: DebugLevel,
+			fn:    Infow,
+			count: 1,
+		},
+		{
+			name:  "info/error",
+			level: InfoLevel,
+			fn:    Errorw,
+			count: 1,
+		},
+		{
+			name:  "info/info",
+			level: InfoLevel,
+			fn:    Infow,
+			count: 1,
+		},
+		{
+			name:  "info/severe",
+			level: InfoLevel,
+			fn:    Errorw,
+			count: 1,
+		},
+		{
+			name:  "error/info",
+			level: ErrorLevel,
+			fn:    Infow,
+			count: 0,
+		},
+		{
+			name:  "error/debug",
+			level: ErrorLevel,
+			fn:    Debugw,
+			count: 0,
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			olevel := atomic.LoadUint32(&logLevel)
+			SetLevel(tt.level)
+			defer SetLevel(olevel)
+
+			var val countingStringer
+			tt.fn("hello there", Field("foo", &val))
+			assert.Equal(t, tt.count, val.Count())
+		})
+	}
+}
+
+func TestWithField_LogLevelWithContext(t *testing.T) {
+	t.Run("context more than once with info/info", func(t *testing.T) {
+		olevel := atomic.LoadUint32(&logLevel)
+		SetLevel(InfoLevel)
+		defer SetLevel(olevel)
+
+		var val countingStringer
+		ctx := ContextWithFields(context.Background(), Field("foo", &val))
+		logger := WithContext(ctx)
+		logger.Info("hello there")
+		logger.Info("hello there")
+		logger.Info("hello there")
+		assert.True(t, val.Count() > 0)
+	})
+
+	t.Run("context more than once with error/info", func(t *testing.T) {
+		olevel := atomic.LoadUint32(&logLevel)
+		SetLevel(ErrorLevel)
+		defer SetLevel(olevel)
+
+		var val countingStringer
+		ctx := ContextWithFields(context.Background(), Field("foo", &val))
+		logger := WithContext(ctx)
+		logger.Info("hello there")
+		logger.Info("hello there")
+		logger.Info("hello there")
+		assert.Equal(t, int32(0), val.Count())
+	})
+}
+
 func BenchmarkCopyByteSliceAppend(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		var buf []byte
@@ -810,15 +1048,26 @@ func doTestStructedLogConsole(t *testing.T, w *mockWriter, write func(...any)) {
 	assert.True(t, strings.Contains(w.String(), message))
 }
 
+func doTestStructedLogEmpty(t *testing.T, w *mockWriter, level uint32, write func(...any)) {
+	olevel := atomic.LoadUint32(&logLevel)
+	SetLevel(level)
+	defer SetLevel(olevel)
+
+	const message = "hello there"
+	write(message)
+	assert.Empty(t, w.String())
+}
+
 func testSetLevelTwiceWithMode(t *testing.T, mode string, w *mockWriter) {
 	writer.Store(nil)
 	SetUp(LogConf{
-		Mode:       mode,
-		Level:      "debug",
-		Path:       "/dev/null",
-		Encoding:   plainEncoding,
-		Stat:       false,
-		TimeFormat: time.RFC3339,
+		Mode:           mode,
+		Level:          "debug",
+		Path:           "/dev/null",
+		Encoding:       plainEncoding,
+		Stat:           false,
+		TimeFormat:     time.RFC3339,
+		FileTimeFormat: time.DateTime,
 	})
 	SetUp(LogConf{
 		Mode:  mode,
@@ -858,3 +1107,112 @@ func validateFields(t *testing.T, content string, fields map[string]any) {
 		}
 	}
 }
+
+type nilError struct {
+	Name string
+}
+
+func (e *nilError) Error() string {
+	return e.Name
+}
+
+type nilStringer struct {
+	Name string
+}
+
+func (s *nilStringer) String() string {
+	return s.Name
+}
+
+type innerPanicStringer struct {
+	Inner *struct {
+		Name string
+	}
+}
+
+func (s innerPanicStringer) String() string {
+	return s.Inner.Name
+}
+
+type panicStringer struct {
+}
+
+func (s panicStringer) String() string {
+	panic("panic")
+}
+
+type countingStringer struct {
+	count int32
+}
+
+func (s *countingStringer) Count() int32 {
+	return atomic.LoadInt32(&s.count)
+}
+
+func (s *countingStringer) String() string {
+	atomic.AddInt32(&s.count, 1)
+	return "countingStringer"
+}
+
+func TestLogKey(t *testing.T) {
+	setupOnce = sync.Once{}
+	MustSetup(LogConf{
+		ServiceName: "any",
+		Mode:        "console",
+		Encoding:    "json",
+		TimeFormat:  timeFormat,
+		FieldKeys: fieldKeyConf{
+			CallerKey:    "_caller",
+			ContentKey:   "_content",
+			DurationKey:  "_duration",
+			LevelKey:     "_level",
+			SpanKey:      "_span",
+			TimestampKey: "_timestamp",
+			TraceKey:     "_trace",
+			TruncatedKey: "_truncated",
+		},
+	})
+
+	t.Cleanup(func() {
+		setupFieldKeys(fieldKeyConf{
+			CallerKey:    defaultCallerKey,
+			ContentKey:   defaultContentKey,
+			DurationKey:  defaultDurationKey,
+			LevelKey:     defaultLevelKey,
+			SpanKey:      defaultSpanKey,
+			TimestampKey: defaultTimestampKey,
+			TraceKey:     defaultTraceKey,
+			TruncatedKey: defaultTruncatedKey,
+		})
+	})
+
+	const message = "hello there"
+	w := new(mockWriter)
+	old := writer.Swap(w)
+	defer writer.Store(old)
+
+	otp := otel.GetTracerProvider()
+	tp := trace.NewTracerProvider(trace.WithSampler(trace.AlwaysSample()))
+	otel.SetTracerProvider(tp)
+	defer otel.SetTracerProvider(otp)
+
+	ctx, span := tp.Tracer("trace-id").Start(context.Background(), "span-id")
+	defer span.End()
+
+	WithContext(ctx).WithDuration(time.Second).Info(message)
+	now := time.Now()
+
+	var m map[string]string
+	if err := json.Unmarshal([]byte(w.String()), &m); err != nil {
+		t.Error(err)
+	}
+	assert.Equal(t, "info", m["_level"])
+	assert.Equal(t, message, m["_content"])
+	assert.Equal(t, "1000.0ms", m["_duration"])
+	assert.Regexp(t, `logx/logs_test.go:\d+`, m["_caller"])
+	assert.NotEmpty(t, m["_trace"])
+	assert.NotEmpty(t, m["_span"])
+	parsedTime, err := time.Parse(timeFormat, m["_timestamp"])
+	assert.True(t, err == nil)
+	assert.Equal(t, now.Minute(), parsedTime.Minute())
+}

core/logx/richlogger.go

@@ -52,6 +52,12 @@ func (l *richLogger) Debugf(format string, v ...any) {
 	}
 }
 
+func (l *richLogger) Debugfn(fn func() any) {
+	if shallLog(DebugLevel) {
+		l.debug(fn())
+	}
+}
+
 func (l *richLogger) Debugv(v any) {
 	if shallLog(DebugLevel) {
 		l.debug(v)
@@ -76,6 +82,12 @@ func (l *richLogger) Errorf(format string, v ...any) {
 	}
 }
 
+func (l *richLogger) Errorfn(fn func() any) {
+	if shallLog(ErrorLevel) {
+		l.err(fn())
+	}
+}
+
 func (l *richLogger) Errorv(v any) {
 	if shallLog(ErrorLevel) {
 		l.err(v)
@@ -100,6 +112,12 @@ func (l *richLogger) Infof(format string, v ...any) {
 	}
 }
 
+func (l *richLogger) Infofn(fn func() any) {
+	if shallLog(InfoLevel) {
+		l.info(fn())
+	}
+}
+
 func (l *richLogger) Infov(v any) {
 	if shallLog(InfoLevel) {
 		l.info(v)
@@ -124,6 +142,12 @@ func (l *richLogger) Slowf(format string, v ...any) {
 	}
 }
 
+func (l *richLogger) Slowfn(fn func() any) {
+	if shallLog(ErrorLevel) {
+		l.slow(fn())
+	}
+}
+
 func (l *richLogger) Slowv(v any) {
 	if shallLog(ErrorLevel) {
 		l.slow(v)
@@ -141,28 +165,50 @@ func (l *richLogger) WithCallerSkip(skip int) Logger {
 		return l
 	}
 
-	l.callerSkip = skip
-	return l
+	return &richLogger{
+		ctx:        l.ctx,
+		callerSkip: skip,
+		fields:     l.fields,
+	}
 }
 
 func (l *richLogger) WithContext(ctx context.Context) Logger {
-	l.ctx = ctx
-	return l
+	return &richLogger{
+		ctx:        ctx,
+		callerSkip: l.callerSkip,
+		fields:     l.fields,
+	}
 }
 
 func (l *richLogger) WithDuration(duration time.Duration) Logger {
-	l.fields = append(l.fields, Field(durationKey, timex.ReprOfDuration(duration)))
-	return l
+	fields := append(l.fields, Field(durationKey, timex.ReprOfDuration(duration)))
+
+	return &richLogger{
+		ctx:        l.ctx,
+		callerSkip: l.callerSkip,
+		fields:     fields,
+	}
 }
 
 func (l *richLogger) WithFields(fields ...LogField) Logger {
-	l.fields = append(l.fields, fields...)
-	return l
+	if len(fields) == 0 {
+		return l
+	}
+
+	f := append(l.fields, fields...)
+
+	return &richLogger{
+		ctx:        l.ctx,
+		callerSkip: l.callerSkip,
+		fields:     f,
+	}
 }
 
 func (l *richLogger) buildFields(fields ...LogField) []LogField {
 	fields = append(l.fields, fields...)
+	// caller field should always appear together with global fields
 	fields = append(fields, Field(callerKey, getCaller(callerDepth+l.callerSkip)))
+	fields = mergeGlobalFields(fields)
 
 	if l.ctx == nil {
 		return fields
@@ -178,7 +224,7 @@ func (l *richLogger) buildFields(fields ...LogField) []LogField {
 		fields = append(fields, Field(spanKey, spanID))
 	}
 
-	val := l.ctx.Value(fieldsContextKey)
+	val := l.ctx.Value(fieldsKey{})
 	if val != nil {
 		if arr, ok := val.([]LogField); ok {
 			fields = append(fields, arr...)

core/logx/richlogger_test.go

@@ -63,6 +63,11 @@ func TestTraceDebug(t *testing.T) {
 	l.WithDuration(time.Second).Debugf(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
+	l.WithDuration(time.Second).Debugfn(func() any {
+		return testlog
+	})
+	validate(t, w.String(), true, true)
+	w.Reset()
 	l.WithDuration(time.Second).Debugv(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
@@ -103,6 +108,11 @@ func TestTraceError(t *testing.T) {
 	l.WithDuration(time.Second).Errorf(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
+	l.WithDuration(time.Second).Errorfn(func() any {
+		return testlog
+	})
+	validate(t, w.String(), true, true)
+	w.Reset()
 	l.WithDuration(time.Second).Errorv(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
@@ -140,6 +150,11 @@ func TestTraceInfo(t *testing.T) {
 	l.WithDuration(time.Second).Infof(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
+	l.WithDuration(time.Second).Infofn(func() any {
+		return testlog
+	})
+	validate(t, w.String(), true, true)
+	w.Reset()
 	l.WithDuration(time.Second).Infov(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
@@ -213,6 +228,11 @@ func TestTraceSlow(t *testing.T) {
 	l.WithDuration(time.Second).Slowf(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
+	l.WithDuration(time.Second).Slowfn(func() any {
+		return testlog
+	})
+	validate(t, w.String(), true, true)
+	w.Reset()
 	l.WithDuration(time.Second).Slowv(testlog)
 	validate(t, w.String(), true, true)
 	w.Reset()
@@ -287,6 +307,54 @@ func TestLogWithCallerSkip(t *testing.T) {
 	assert.True(t, w.Contains(fmt.Sprintf("%s:%d", file, line+1)))
 }
 
+func TestLogWithCallerSkipCopy(t *testing.T) {
+	log1 := WithCallerSkip(2)
+	log2 := log1.WithCallerSkip(3)
+	log3 := log2.WithCallerSkip(-1)
+	assert.Equal(t, 2, log1.(*richLogger).callerSkip)
+	assert.Equal(t, 3, log2.(*richLogger).callerSkip)
+	assert.Equal(t, 3, log3.(*richLogger).callerSkip)
+}
+
+func TestLogWithContextCopy(t *testing.T) {
+	c1 := context.Background()
+	c2 := context.WithValue(context.Background(), "foo", "bar")
+	log1 := WithContext(c1)
+	log2 := log1.WithContext(c2)
+	assert.Equal(t, c1, log1.(*richLogger).ctx)
+	assert.Equal(t, c2, log2.(*richLogger).ctx)
+}
+
+func TestLogWithDurationCopy(t *testing.T) {
+	log1 := WithContext(context.Background())
+	log2 := log1.WithDuration(time.Second)
+	assert.Empty(t, log1.(*richLogger).fields)
+	assert.Equal(t, 1, len(log2.(*richLogger).fields))
+
+	var w mockWriter
+	old := writer.Swap(&w)
+	defer writer.Store(old)
+	log2.Info("hello")
+	assert.Contains(t, w.String(), `"duration":"1000.0ms"`)
+}
+
+func TestLogWithFieldsCopy(t *testing.T) {
+	log1 := WithContext(context.Background())
+	log2 := log1.WithFields(Field("foo", "bar"))
+	log3 := log1.WithFields()
+	assert.Empty(t, log1.(*richLogger).fields)
+	assert.Equal(t, 1, len(log2.(*richLogger).fields))
+	assert.Equal(t, log1, log3)
+	assert.Empty(t, log3.(*richLogger).fields)
+
+	var w mockWriter
+	old := writer.Swap(&w)
+	defer writer.Store(old)
+
+	log2.Info("hello")
+	assert.Contains(t, w.String(), `"foo":"bar"`)
+}
+
 func TestLoggerWithFields(t *testing.T) {
 	w := new(mockWriter)
 	old := writer.Swap(w)
@@ -355,3 +423,49 @@ type mockValue struct {
 	Foo     string `json:"foo"`
 	Content any    `json:"content"`
 }
+
+type testJson struct {
+	Name  string  `json:"name"`
+	Age   int     `json:"age"`
+	Score float64 `json:"score"`
+}
+
+func (t testJson) MarshalJSON() ([]byte, error) {
+	type testJsonImpl testJson
+	return json.Marshal(testJsonImpl(t))
+}
+
+func (t testJson) String() string {
+	return fmt.Sprintf("%s %d %f", t.Name, t.Age, t.Score)
+}
+
+func TestLogWithJson(t *testing.T) {
+	w := new(mockWriter)
+	old := writer.Swap(w)
+	writer.lock.RLock()
+	defer func() {
+		writer.lock.RUnlock()
+		writer.Store(old)
+	}()
+
+	l := WithContext(context.Background()).WithFields(Field("bar", testJson{
+		Name:  "foo",
+		Age:   1,
+		Score: 1.0,
+	}))
+	l.Info(testlog)
+
+	type mockValue2 struct {
+		mockValue
+		Bar testJson `json:"bar"`
+	}
+
+	var val mockValue2
+	err := json.Unmarshal([]byte(w.String()), &val)
+	assert.NoError(t, err)
+
+	assert.Equal(t, testlog, val.Content)
+	assert.Equal(t, "foo", val.Bar.Name)
+	assert.Equal(t, 1, val.Bar.Age)
+	assert.Equal(t, 1.0, val.Bar.Score)
+}

core/logx/rotatelogger.go

@@ -18,8 +18,6 @@ import (
 )
 
 const (
-	dateFormat      = "2006-01-02"
-	fileTimeFormat  = time.RFC3339
 	hoursPerDay     = 24
 	bufferSize      = 100
 	defaultDirMode  = 0o755
@@ -28,8 +26,12 @@ const (
 	megaBytes       = 1 << 20
 )
 
-// ErrLogFileClosed is an error that indicates the log file is already closed.
-var ErrLogFileClosed = errors.New("error: log file closed")
+var (
+	// ErrLogFileClosed is an error that indicates the log file is already closed.
+	ErrLogFileClosed = errors.New("error: log file closed")
+
+	fileTimeFormat = time.RFC3339
+)
 
 type (
 	// A RotateRule interface is used to define the log rotating rules.
@@ -113,7 +115,7 @@ func (r *DailyRotateRule) OutdatedFiles() []string {
 	}
 
 	var buf strings.Builder
-	boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay*r.days)).Format(dateFormat)
+	boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay*r.days)).Format(time.DateOnly)
 	buf.WriteString(r.filename)
 	buf.WriteString(r.delimiter)
 	buf.WriteString(boundary)
@@ -209,7 +211,7 @@ func (r *SizeLimitRotateRule) OutdatedFiles() []string {
 		}
 	}
 
-	var result []string
+	result := make([]string, 0, len(outdated))
 	for k := range outdated {
 		result = append(result, k)
 	}
@@ -319,7 +321,7 @@ func (l *RotateLogger) maybeCompressFile(file string) {
 	}()
 
 	if _, err := os.Stat(file); err != nil {
-		// file not exists or other error, ignore compression
+		// file doesn't exist or another error, ignore compression
 		return
 	}
 
@@ -422,7 +424,7 @@ func compressLogFile(file string) {
 }
 
 func getNowDate() string {
-	return time.Now().Format(dateFormat)
+	return time.Now().Format(time.DateOnly)
 }
 
 func getNowDateInRFC3339Format() string {

core/logx/rotatelogger_test.go

@@ -52,7 +52,7 @@ func TestDailyRotateRuleOutdatedFiles(t *testing.T) {
 	})
 
 	t.Run("temp files", func(t *testing.T) {
-		boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay) * 2).Format(time.DateOnly)
 		f1, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
 		assert.NoError(t, err)
 		_ = f1.Close()
@@ -73,7 +73,7 @@ func TestDailyRotateRuleOutdatedFiles(t *testing.T) {
 
 func TestDailyRotateRuleShallRotate(t *testing.T) {
 	var rule DailyRotateRule
-	rule.rotatedTime = time.Now().Add(time.Hour * 24).Format(dateFormat)
+	rule.rotatedTime = time.Now().Add(time.Hour * 24).Format(time.DateOnly)
 	assert.True(t, rule.ShallRotate(0))
 }
 
@@ -117,12 +117,12 @@ func TestSizeLimitRotateRuleOutdatedFiles(t *testing.T) {
 	})
 
 	t.Run("temp files", func(t *testing.T) {
-		boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay) * 2).Format(time.DateOnly)
 		f1, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
 		assert.NoError(t, err)
 		f2, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
 		assert.NoError(t, err)
-		boundary1 := time.Now().Add(time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		boundary1 := time.Now().Add(time.Hour * time.Duration(hoursPerDay) * 2).Format(time.DateOnly)
 		f3, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary1)
 		assert.NoError(t, err)
 		t.Cleanup(func() {
@@ -144,12 +144,12 @@ func TestSizeLimitRotateRuleOutdatedFiles(t *testing.T) {
 	})
 
 	t.Run("no backups", func(t *testing.T) {
-		boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		boundary := time.Now().Add(-time.Hour * time.Duration(hoursPerDay) * 2).Format(time.DateOnly)
 		f1, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
 		assert.NoError(t, err)
 		f2, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary)
 		assert.NoError(t, err)
-		boundary1 := time.Now().Add(time.Hour * time.Duration(hoursPerDay) * 2).Format(dateFormat)
+		boundary1 := time.Now().Add(time.Hour * time.Duration(hoursPerDay) * 2).Format(time.DateOnly)
 		f3, err := os.CreateTemp(os.TempDir(), "go-zero-test-"+boundary1)
 		assert.NoError(t, err)
 		t.Cleanup(func() {
@@ -319,7 +319,7 @@ func TestRotateLoggerWrite(t *testing.T) {
 	}
 	// the following write calls cannot be changed to Write, because of DATA RACE.
 	logger.write([]byte(`foo`))
-	rule.rotatedTime = time.Now().Add(-time.Hour * 24).Format(dateFormat)
+	rule.rotatedTime = time.Now().Add(-time.Hour * 24).Format(time.DateOnly)
 	logger.write([]byte(`bar`))
 	logger.Close()
 	logger.write([]byte(`baz`))
@@ -447,7 +447,7 @@ func TestRotateLoggerWithSizeLimitRotateRuleWrite(t *testing.T) {
 	}
 	// the following write calls cannot be changed to Write, because of DATA RACE.
 	logger.write([]byte(`foo`))
-	rule.rotatedTime = time.Now().Add(-time.Hour * 24).Format(dateFormat)
+	rule.rotatedTime = time.Now().Add(-time.Hour * 24).Format(time.DateOnly)
 	logger.write([]byte(`bar`))
 	logger.Close()
 	logger.write([]byte(`baz`))

core/logx/sensitive.go

@@ -0,0 +1,21 @@
+package logx
+
+// Sensitive is an interface that defines a method for masking sensitive information in logs.
+// It is typically implemented by types that contain sensitive data,
+// such as passwords or personal information.
+// Infov, Errorv, Debugv, and Slowv methods will call this method to mask sensitive data.
+// The values in LogField will also be masked if they implement the Sensitive interface.
+type Sensitive interface {
+	// MaskSensitive masks sensitive information in the log.
+	MaskSensitive() any
+}
+
+// maskSensitive returns the value returned by MaskSensitive method,
+// if the value implements Sensitive interface.
+func maskSensitive(v any) any {
+	if s, ok := v.(Sensitive); ok {
+		return s.MaskSensitive()
+	}
+
+	return v
+}

core/logx/sensitive_test.go

@@ -0,0 +1,50 @@
+package logx
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const maskedContent = "******"
+
+type User struct {
+	Name string
+	Pass string
+}
+
+func (u User) MaskSensitive() any {
+	return User{
+		Name: u.Name,
+		Pass: maskedContent,
+	}
+}
+
+type NonSensitiveUser struct {
+	Name string
+	Pass string
+}
+
+func TestMaskSensitive(t *testing.T) {
+	t.Run("sensitive", func(t *testing.T) {
+		user := User{
+			Name: "kevin",
+			Pass: "123",
+		}
+
+		mu := maskSensitive(user)
+		assert.Equal(t, user.Name, mu.(User).Name)
+		assert.Equal(t, maskedContent, mu.(User).Pass)
+	})
+
+	t.Run("non-sensitive", func(t *testing.T) {
+		user := NonSensitiveUser{
+			Name: "kevin",
+			Pass: "123",
+		}
+
+		mu := maskSensitive(user)
+		assert.Equal(t, user.Name, mu.(NonSensitiveUser).Name)
+		assert.Equal(t, user.Pass, mu.(NonSensitiveUser).Pass)
+	})
+}

core/logx/vars.go

@@ -15,6 +15,8 @@ const (
 	ErrorLevel
 	// SevereLevel only log severe messages
 	SevereLevel
+	// disableLevel doesn't log any messages
+	disableLevel = 0xff
 )
 
 const (
@@ -46,18 +48,19 @@ const (
 	levelDebug  = "debug"
 
 	backupFileDelimiter = "-"
+	nilAngleString      = "<nil>"
 	flags               = 0x0
 )
 
 const (
-	callerKey    = "caller"
-	contentKey   = "content"
-	durationKey  = "duration"
-	levelKey     = "level"
-	spanKey      = "span"
-	timestampKey = "@timestamp"
-	traceKey     = "trace"
-	truncatedKey = "truncated"
+	defaultCallerKey    = "caller"
+	defaultContentKey   = "content"
+	defaultDurationKey  = "duration"
+	defaultLevelKey     = "level"
+	defaultSpanKey      = "span"
+	defaultTimestampKey = "@timestamp"
+	defaultTraceKey     = "trace"
+	defaultTruncatedKey = "truncated"
 )
 
 var (
@@ -70,3 +73,14 @@ var (
 
 	truncatedField = Field(truncatedKey, true)
 )
+
+var (
+	callerKey    = defaultCallerKey
+	contentKey   = defaultContentKey
+	durationKey  = defaultDurationKey
+	levelKey     = defaultLevelKey
+	spanKey      = defaultSpanKey
+	timestampKey = defaultTimestampKey
+	traceKey     = defaultTraceKey
+	truncatedKey = defaultTruncatedKey
+)

core/logx/writer.go

@@ -10,21 +10,35 @@ import (
 	"runtime/debug"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	fatihcolor "github.com/fatih/color"
 	"github.com/zeromicro/go-zero/core/color"
+	"github.com/zeromicro/go-zero/core/errorx"
 )
 
 type (
+	// Writer is the interface for writing logs.
+	// It's designed to let users customize their own log writer,
+	// such as writing logs to a kafka, a database, or using third-party loggers.
 	Writer interface {
+		// Alert sends an alert message, if your writer implemented alerting functionality.
 		Alert(v any)
+		// Close closes the writer.
 		Close() error
+		// Debug logs a message at debug level.
 		Debug(v any, fields ...LogField)
+		// Error logs a message at error level.
 		Error(v any, fields ...LogField)
+		// Info logs a message at info level.
 		Info(v any, fields ...LogField)
+		// Severe logs a message at severe level.
 		Severe(v any)
+		// Slow logs a message at slow level.
 		Slow(v any, fields ...LogField)
+		// Stack logs a message at error level.
 		Stack(v any)
+		// Stat logs a message at stat level.
 		Stat(v any, fields ...LogField)
 	}
 
@@ -33,6 +47,10 @@ type (
 		lock   sync.RWMutex
 	}
 
+	comboWriter struct {
+		writers []Writer
+	}
+
 	concreteWriter struct {
 		infoLog   io.WriteCloser
 		errorLog  io.WriteCloser
@@ -88,6 +106,62 @@ func (w *atomicWriter) Swap(v Writer) Writer {
 	return old
 }
 
+func (c comboWriter) Alert(v any) {
+	for _, w := range c.writers {
+		w.Alert(v)
+	}
+}
+
+func (c comboWriter) Close() error {
+	var be errorx.BatchError
+	for _, w := range c.writers {
+		be.Add(w.Close())
+	}
+	return be.Err()
+}
+
+func (c comboWriter) Debug(v any, fields ...LogField) {
+	for _, w := range c.writers {
+		w.Debug(v, fields...)
+	}
+}
+
+func (c comboWriter) Error(v any, fields ...LogField) {
+	for _, w := range c.writers {
+		w.Error(v, fields...)
+	}
+}
+
+func (c comboWriter) Info(v any, fields ...LogField) {
+	for _, w := range c.writers {
+		w.Info(v, fields...)
+	}
+}
+
+func (c comboWriter) Severe(v any) {
+	for _, w := range c.writers {
+		w.Severe(v)
+	}
+}
+
+func (c comboWriter) Slow(v any, fields ...LogField) {
+	for _, w := range c.writers {
+		w.Slow(v, fields...)
+	}
+}
+
+func (c comboWriter) Stack(v any) {
+	for _, w := range c.writers {
+		w.Stack(v)
+	}
+}
+
+func (c comboWriter) Stat(v any, fields ...LogField) {
+	for _, w := range c.writers {
+		w.Stat(v, fields...)
+	}
+}
+
 func newConsoleWriter() Writer {
 	outLog := newLogWriter(log.New(fatihcolor.Output, "", flags))
 	errLog := newLogWriter(log.New(fatihcolor.Error, "", flags))
@@ -138,7 +212,6 @@ func newFileWriter(c LogConf) (Writer, error) {
 	statFile := path.Join(c.Path, statFilename)
 
 	handleOptions(opts)
-	setupLogLevel(c)
 
 	if infoLog, err = createOutput(accessFile); err != nil {
 		return nil, err
@@ -254,17 +327,30 @@ func (n nopWriter) Stack(_ any) {
 func (n nopWriter) Stat(_ any, _ ...LogField) {
 }
 
-func buildPlainFields(fields ...LogField) []string {
-	var items []string
-
-	for _, field := range fields {
-		items = append(items, fmt.Sprintf("%s=%+v", field.Key, field.Value))
+func buildPlainFields(fields logEntry) []string {
+	items := make([]string, 0, len(fields))
+	for k, v := range fields {
+		items = append(items, fmt.Sprintf("%s=%+v", k, v))
 	}
 
 	return items
 }
 
-func combineGlobalFields(fields []LogField) []LogField {
+func marshalJson(t interface{}) ([]byte, error) {
+	var buf bytes.Buffer
+	encoder := json.NewEncoder(&buf)
+	encoder.SetEscapeHTML(false)
+	err := encoder.Encode(t)
+	// go 1.5+ will append a newline to the end of the json string
+	// https://github.com/golang/go/issues/13520
+	if l := buf.Len(); l > 0 && buf.Bytes()[l-1] == '\n' {
+		buf.Truncate(l - 1)
+	}
+
+	return buf.Bytes(), err
+}
+
+func mergeGlobalFields(fields []LogField) []LogField {
 	globals := globalFields.Load()
 	if globals == nil {
 		return fields
@@ -279,25 +365,32 @@ func combineGlobalFields(fields []LogField) []LogField {
 }
 
 func output(writer io.Writer, level string, val any, fields ...LogField) {
-	// only truncate string content, don't know how to truncate the values of other types.
-	if v, ok := val.(string); ok {
+	switch v := val.(type) {
+	case string:
+		// only truncate string content, don't know how to truncate the values of other types.
 		maxLen := atomic.LoadUint32(&maxContentLength)
 		if maxLen > 0 && len(v) > int(maxLen) {
 			val = v[:maxLen]
 			fields = append(fields, truncatedField)
 		}
+	case Sensitive:
+		val = v.MaskSensitive()
 	}
 
-	fields = combineGlobalFields(fields)
+	// +3 for timestamp, level and content
+	entry := make(logEntry, len(fields)+3)
+	for _, field := range fields {
+		// mask sensitive data before processing types,
+		// in case field.Value is a sensitive type and also implemented fmt.Stringer.
+		mval := maskSensitive(field.Value)
+		entry[field.Key] = processFieldValue(mval)
+	}
 
 	switch atomic.LoadUint32(&encoding) {
 	case plainEncodingType:
-		writePlainAny(writer, level, val, buildPlainFields(fields...)...)
+		plainFields := buildPlainFields(entry)
+		writePlainAny(writer, level, val, plainFields...)
 	default:
-		entry := make(logEntry)
-		for _, field := range fields {
-			entry[field.Key] = field.Value
-		}
 		entry[timestampKey] = getTimestamp()
 		entry[levelKey] = level
 		entry[contentKey] = val
@@ -305,6 +398,45 @@ func output(writer io.Writer, level string, val any, fields ...LogField) {
 	}
 }
 
+func processFieldValue(value any) any {
+	switch val := value.(type) {
+	case error:
+		return encodeError(val)
+	case []error:
+		var errs []string
+		for _, err := range val {
+			errs = append(errs, encodeError(err))
+		}
+		return errs
+	case time.Duration:
+		return fmt.Sprint(val)
+	case []time.Duration:
+		var durs []string
+		for _, dur := range val {
+			durs = append(durs, fmt.Sprint(dur))
+		}
+		return durs
+	case []time.Time:
+		var times []string
+		for _, t := range val {
+			times = append(times, fmt.Sprint(t))
+		}
+		return times
+	case json.Marshaler:
+		return val
+	case fmt.Stringer:
+		return encodeStringer(val)
+	case []fmt.Stringer:
+		var strs []string
+		for _, str := range val {
+			strs = append(strs, encodeStringer(str))
+		}
+		return strs
+	default:
+		return val
+	}
+}
+
 func wrapLevelWithColor(level string) string {
 	var colour color.Color
 	switch level {
@@ -332,7 +464,7 @@ func wrapLevelWithColor(level string) string {
 }
 
 func writeJson(writer io.Writer, info any) {
-	if content, err := json.Marshal(info); err != nil {
+	if content, err := marshalJson(info); err != nil {
 		log.Printf("err: %s\n\n%s", err.Error(), debug.Stack())
 	} else if writer == nil {
 		log.Println(string(content))