Repository: cloudwego/hertz
Pull request #182 (2022-08-07 07:25:22)

feat: add service registration & service discovery & load balancing
#### What type of PR is this?
feat 

#### What this PR does / why we need it (English/Chinese):
- en: add service registration & service discovery & load balancing
- ch: 添加服务注册注册 & 服务发现 & 负载均衡

#### Which issue(s) this PR fixes:

#### example:
https://github.com/li-jin-gou/nacos-demo/tree/main/example

- server
```go
                addr := "127.0.0.1:8889"
		r := nacos_demo.NewNacosRegistry(cli)
		h := server.Default(
			server.WithHostPorts(addr),
			server.WithRegistry(r, &registry.Info{
				ServiceName: "hertz.test.demo",
				Addr:        utils.NewNetAddr("tcp", addr),
				Weight:      10,
				Tags:        nil,
			}))
		h.GET("/ping", func(c context.Context, ctx *app.RequestContext) {
			ctx.JSON(consts.StatusOK, utils.H{"ping": "pong2"})
		})
		h.Spin()
```

- client:
```go
	r := nacos_demo.NewNacosResolver(naocsCli)
	cli.Use(sd.Discovery(r))
	for i := 0; i < 10; i++ {
		status, body, err := cli.Get(context.Background(), nil, "http://hertz.test.demo/ping", config.WithSD(true))
		if err != nil {
			hlog.Fatal(err)
		}
		hlog.Infof("code=%d,body=%s", status, string(body))
	}
```

```
2022/07/26 13:52:47.310617 main.go:46: [Info] code =200, body ={"ping":"pong2"}
2022/07/26 13:52:47.311019 main.go:46: [Info] code = 200, body ={"ping":"pong1"}
2022/07/26 13:52:47.311186 main.go:46: [Info] code = 200, body ={"ping":"pong1"}
2022/07/26 13:52:47.311318 main.go:46: [Info] code = 200, body ={"ping":"pong1"}
2022/07/26 13:52:47.311445 main.go:46: [Info] code = 200, body ={"ping":"pong1"}
2022/07/26 13:52:47.311585 main.go:46: [Info] code = 200, body ={"ping":"pong2"}
2022/07/26 13:52:47.311728 main.go:46: [Info] code = 200, body ={"ping":"pong1"}
2022/07/26 13:52:47.311858 main.go:46: [Info] code = 200, body ={"ping":"pong1"}
2022/07/26 13:52:47.311977 main.go:46: [Info] code = 200, body ={"ping":"pong1"}
2022/07/26 13:52:47.312107 main.go:46: [Info] code = 200, body ={"ping":"pong1"}
```



Patch:
diff --git a/go.mod b/go.mod
index 305f60c..9cf0e61 100644
--- a/go.mod
+++ b/go.mod
@@ -7,6 +7,7 @@ require (
 	github.com/bytedance/gopkg v0.0.0-20220413063733-65bf48ffb3a7
 	github.com/bytedance/sonic v1.3.0
 	github.com/cloudwego/netpoll v0.2.4
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/sys v0.0.0-20220110181412-a018aaa089fe
 	google.golang.org/protobuf v1.27.1
 )
diff --git a/go.sum b/go.sum
index 57ad438..8444b8e 100644
--- a/go.sum
+++ b/go.sum
@@ -56,6 +56,7 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 golang.org/x/arch v0.0.0-20210923205945-b76863e36670 h1:18EFjUmQOcUvxNYSkA6jO9VAiXCnxFY6NyDX0bHDmkU=
 golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20220110181412-a018aaa089fe h1:W8vbETX/n8S6EmY0Pu4Ix7VvpsJUESTwl0oCK8MJOgk=
 golang.org/x/sys v0.0.0-20220110181412-a018aaa089fe/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
diff --git a/pkg/app/client/discovery/discovery.go b/pkg/app/client/discovery/discovery.go
new file mode 100644
index 0000000..0757a1b
--- /dev/null
+++ b/pkg/app/client/discovery/discovery.go
@@ -0,0 +1,112 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package discovery
+
+import (
+	"context"
+	"net"
+
+	"github.com/cloudwego/hertz/pkg/app/server/registry"
+	"github.com/cloudwego/hertz/pkg/common/utils"
+)
+
+type TargetInfo struct {
+	Host string
+	Tags map[string]string
+}
+
+type Resolver interface {
+	// Target should return a description for the given target that is suitable for being a key for cache.
+	Target(ctx context.Context, target *TargetInfo) string
+
+	// Resolve returns a list of instances for the given description of a target.
+	Resolve(ctx context.Context, desc string) (Result, error)
+
+	// Name returns the name of the resolver.
+	Name() string
+}
+
+// SynthesizedResolver synthesizes a Resolver using a resolve function.
+type SynthesizedResolver struct {
+	TargetFunc  func(ctx context.Context, target *TargetInfo) string
+	ResolveFunc func(ctx context.Context, key string) (Result, error)
+	NameFunc    func() string
+}
+
+func (sr SynthesizedResolver) Target(ctx context.Context, target *TargetInfo) string {
+	if sr.TargetFunc == nil {
+		return ""
+	}
+	return sr.TargetFunc(ctx, target)
+}
+
+func (sr SynthesizedResolver) Resolve(ctx context.Context, key string) (Result, error) {
+	return sr.ResolveFunc(ctx, key)
+}
+
+// Name implements the Resolver interface
+func (sr SynthesizedResolver) Name() string {
+	if sr.NameFunc == nil {
+		return ""
+	}
+	return sr.NameFunc()
+}
+
+// Instance contains information of an instance from the target service.
+type Instance interface {
+	Address() net.Addr
+	Weight() int
+	Tag(key string) (value string, exist bool)
+}
+
+type instance struct {
+	addr   net.Addr
+	weight int
+	tags   map[string]string
+}
+
+func (i *instance) Address() net.Addr {
+	return i.addr
+}
+
+func (i *instance) Weight() int {
+	if i.weight > 0 {
+		return i.weight
+	}
+	return registry.DefaultWeight
+}
+
+func (i *instance) Tag(key string) (value string, exist bool) {
+	value, exist = i.tags[key]
+	return
+}
+
+// NewInstance creates an Instance using the given network, address and tags
+func NewInstance(network, address string, weight int, tags map[string]string) Instance {
+	return &instance{
+		addr:   utils.NewNetAddr(network, address),
+		weight: weight,
+		tags:   tags,
+	}
+}
+
+// Result contains the result of service discovery process.
+// the instance list can/should be cached and CacheKey can be used to map the instance list in cache.
+type Result struct {
+	CacheKey  string
+	Instances []Instance
+}
diff --git a/pkg/app/client/loadbalance/lbcache.go b/pkg/app/client/loadbalance/lbcache.go
new file mode 100644
index 0000000..c249ebe
--- /dev/null
+++ b/pkg/app/client/loadbalance/lbcache.go
@@ -0,0 +1,166 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package loadbalance
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/app/client/discovery"
+	"github.com/cloudwego/hertz/pkg/common/errors"
+	"github.com/cloudwego/hertz/pkg/common/hlog"
+	"github.com/cloudwego/hertz/pkg/protocol"
+	"golang.org/x/sync/singleflight"
+)
+
+type cacheResult struct {
+	res         atomic.Value // newest and previous discovery result
+	expire      int32        // 0 = normal, 1 = expire and collect next ticker
+	serviceName string       // service psm
+}
+
+var (
+	balancerFactories    sync.Map // key: resolver name + load-balancer name
+	balancerFactoriesSfg singleflight.Group
+)
+
+func cacheKey(resolver, balancer string, opts Options) string {
+	return fmt.Sprintf("%s|%s|{%s %s}", resolver, balancer, opts.RefreshInterval, opts.ExpireInterval)
+}
+
+type BalancerFactory struct {
+	opts     Options
+	cache    sync.Map // key -> LoadBalancer
+	resolver discovery.Resolver
+	balancer Loadbalancer
+	sfg      singleflight.Group
+}
+
+type Config struct {
+	Resolver discovery.Resolver
+	Balancer Loadbalancer
+	LbOpts   Options
+}
+
+// NewBalancerFactory get or create a balancer with given target.
+// If it has the same key(resolver.Target(target)), we will cache and reuse the Balance.
+func NewBalancerFactory(config Config) *BalancerFactory {
+	config.LbOpts.Check()
+	uniqueKey := cacheKey(config.Resolver.Name(), config.Balancer.Name(), config.LbOpts)
+	val, ok := balancerFactories.Load(uniqueKey)
+	if ok {
+		return val.(*BalancerFactory)
+	}
+	val, _, _ = balancerFactoriesSfg.Do(uniqueKey, func() (interface{}, error) {
+		b := &BalancerFactory{
+			opts:     config.LbOpts,
+			resolver: config.Resolver,
+			balancer: config.Balancer,
+		}
+		go b.watcher()
+		go b.refresh()
+		balancerFactories.Store(uniqueKey, b)
+		return b, nil
+	})
+	return val.(*BalancerFactory)
+}
+
+// watch expired balancer
+func (b *BalancerFactory) watcher() {
+	for range time.Tick(b.opts.ExpireInterval) {
+		b.cache.Range(func(key, value interface{}) bool {
+			cache := value.(*cacheResult)
+			if atomic.CompareAndSwapInt32(&cache.expire, 0, 1) {
+				// 1. set expire flag
+				// 2. wait next ticker for collect, maybe the balancer is used again
+				// (avoid being immediate delete the balancer which had been created recently)
+			} else {
+				b.cache.Delete(key)
+				b.balancer.Delete(key.(string))
+			}
+			return true
+		})
+	}
+}
+
+// cache key with resolver name prefix avoid conflict for balancer
+func renameResultCacheKey(res *discovery.Result, resolverName string) {
+	res.CacheKey = resolverName + ":" + res.CacheKey
+}
+
+// refresh is used to update service discovery information periodically.
+func (b *BalancerFactory) refresh() {
+	for range time.Tick(b.opts.RefreshInterval) {
+		b.cache.Range(func(key, value interface{}) bool {
+			res, err := b.resolver.Resolve(context.Background(), key.(string))
+			if err != nil {
+				hlog.Warnf("Hertz: resolver refresh failed, key=%s error=%s", key, err.Error())
+				return true
+			}
+			renameResultCacheKey(&res, b.resolver.Name())
+			cache := value.(*cacheResult)
+			cache.res.Store(res)
+			atomic.StoreInt32(&cache.expire, 0)
+			b.balancer.Rebalance(res)
+			return true
+		})
+	}
+}
+
+func (b *BalancerFactory) GetInstance(ctx context.Context, req *protocol.Request) (discovery.Instance, error) {
+	cacheRes, err := b.getCacheResult(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	atomic.StoreInt32(&cacheRes.expire, 0)
+	ins := b.balancer.Pick(cacheRes.res.Load().(discovery.Result))
+	if ins == nil {
+		hlog.Errorf("HERTZ: null instance. serviceName: %s, options: %v", string(req.Host()), req.Options())
+		return nil, errors.NewPublic("instance not found")
+	}
+	return ins, nil
+}
+
+func (b *BalancerFactory) getCacheResult(ctx context.Context, req *protocol.Request) (*cacheResult, error) {
+	target := b.resolver.Target(ctx, &discovery.TargetInfo{Host: string(req.Host()), Tags: req.Options().Tags()})
+	cr, existed := b.cache.Load(target)
+	if existed {
+		return cr.(*cacheResult), nil
+	}
+	cr, err, _ := b.sfg.Do(target, func() (interface{}, error) {
+		cache := &cacheResult{
+			serviceName: string(req.Host()),
+		}
+		res, err := b.resolver.Resolve(ctx, target)
+		if err != nil {
+			return cache, err
+		}
+		renameResultCacheKey(&res, b.resolver.Name())
+		cache.res.Store(res)
+		atomic.StoreInt32(&cache.expire, 0)
+		b.balancer.Rebalance(res)
+		b.cache.Store(target, cache)
+		return cache, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return cr.(*cacheResult), nil
+}
diff --git a/pkg/app/client/loadbalance/loadbalance.go b/pkg/app/client/loadbalance/loadbalance.go
new file mode 100644
index 0000000..84e6b7c
--- /dev/null
+++ b/pkg/app/client/loadbalance/loadbalance.go
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package loadbalance
+
+import (
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/app/client/discovery"
+)
+
+// Loadbalancer picks instance for the given service discovery result.
+type Loadbalancer interface {
+	// Pick is used to select an instance according to discovery result
+	Pick(discovery.Result) discovery.Instance
+
+	// Rebalance is used to refresh the cache of load balance's information
+	Rebalance(discovery.Result)
+
+	// Delete is used to delete the cache of load balance's information when it is expired
+	Delete(string)
+
+	// Name returns the name of the Loadbalancer.
+	Name() string
+}
+
+const (
+	DefaultRefreshInterval = 5 * time.Second
+	DefaultExpireInterval  = 15 * time.Second
+)
+
+var DefaultLbOpts = Options{
+	RefreshInterval: DefaultRefreshInterval,
+	ExpireInterval:  DefaultExpireInterval,
+}
+
+// Options for LoadBalance option
+type Options struct {
+	// refresh discovery result timely
+	RefreshInterval time.Duration
+
+	// Balancer expire check interval
+	// we need remove idle Balancers for resource saving
+	ExpireInterval time.Duration
+}
+
+// Check checks option's param
+func (v *Options) Check() {
+	if v.RefreshInterval <= 0 {
+		v.RefreshInterval = DefaultRefreshInterval
+	}
+	if v.ExpireInterval <= 0 {
+		v.ExpireInterval = DefaultExpireInterval
+	}
+}
diff --git a/pkg/app/client/loadbalance/weight_random.go b/pkg/app/client/loadbalance/weight_random.go
new file mode 100644
index 0000000..1fa0f57
--- /dev/null
+++ b/pkg/app/client/loadbalance/weight_random.go
@@ -0,0 +1,109 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package loadbalance
+
+import (
+	"sync"
+
+	"github.com/bytedance/gopkg/lang/fastrand"
+	"github.com/cloudwego/hertz/pkg/app/client/discovery"
+	"github.com/cloudwego/hertz/pkg/common/hlog"
+	"golang.org/x/sync/singleflight"
+)
+
+type weightedBalancer struct {
+	cachedWeightInfo sync.Map
+	sfg              singleflight.Group
+}
+
+type weightInfo struct {
+	instances []discovery.Instance
+	entries   []int
+	weightSum int
+}
+
+// NewWeightedBalancer creates a loadbalancer using weighted-random algorithm.
+func NewWeightedBalancer() Loadbalancer {
+	lb := &weightedBalancer{}
+	return lb
+}
+
+func (wb *weightedBalancer) calcWeightInfo(e discovery.Result) *weightInfo {
+	w := &weightInfo{
+		instances: make([]discovery.Instance, len(e.Instances)),
+		weightSum: 0,
+		entries:   make([]int, len(e.Instances)),
+	}
+
+	var cnt int
+
+	for idx := range e.Instances {
+		weight := e.Instances[idx].Weight()
+		if weight > 0 {
+			w.instances[cnt] = e.Instances[idx]
+			w.entries[cnt] = weight
+			w.weightSum += weight
+			cnt++
+		} else {
+			hlog.Warnf("HERTZ: Invalid weight=%d on instance address=%s", weight, e.Instances[idx].Address())
+		}
+	}
+
+	w.instances = w.instances[:cnt]
+
+	return w
+}
+
+// Pick implements the Loadbalancer interface.
+func (wb *weightedBalancer) Pick(e discovery.Result) discovery.Instance {
+	wi, ok := wb.cachedWeightInfo.Load(e.CacheKey)
+	if !ok {
+		wi, _, _ = wb.sfg.Do(e.CacheKey, func() (interface{}, error) {
+			return wb.calcWeightInfo(e), nil
+		})
+		wb.cachedWeightInfo.Store(e.CacheKey, wi)
+	}
+
+	w := wi.(*weightInfo)
+	if w.weightSum <= 0 {
+		return nil
+	}
+
+	weight := fastrand.Intn(w.weightSum)
+	for i := 0; i < len(w.instances); i++ {
+		weight -= w.entries[i]
+		if weight < 0 {
+			return w.instances[i]
+		}
+	}
+
+	return nil
+}
+
+// Rebalance implements the Loadbalancer interface.
+func (wb *weightedBalancer) Rebalance(e discovery.Result) {
+	wb.cachedWeightInfo.Store(e.CacheKey, wb.calcWeightInfo(e))
+}
+
+// Delete implements the Loadbalancer interface.
+func (wb *weightedBalancer) Delete(cacheKey string) {
+	wb.cachedWeightInfo.Delete(cacheKey)
+}
+
+func (wb *weightedBalancer) Name() string {
+	return "weight_random"
+}
diff --git a/pkg/app/middlewares/client/sd/discovery.go b/pkg/app/middlewares/client/sd/discovery.go
new file mode 100644
index 0000000..038eb92
--- /dev/null
+++ b/pkg/app/middlewares/client/sd/discovery.go
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sd
+
+import (
+	"context"
+
+	"github.com/cloudwego/hertz/pkg/app/client"
+	"github.com/cloudwego/hertz/pkg/app/client/discovery"
+	"github.com/cloudwego/hertz/pkg/app/client/loadbalance"
+	"github.com/cloudwego/hertz/pkg/protocol"
+)
+
+// Discovery will construct a middleware with BalancerFactory.
+func Discovery(resolver discovery.Resolver, opts ...ServiceDiscoveryOption) func(next client.Endpoint) client.Endpoint {
+	options := &ServiceDiscoveryOptions{
+		Balancer: loadbalance.NewWeightedBalancer(),
+		LbOpts:   loadbalance.DefaultLbOpts,
+		Resolver: resolver,
+	}
+	options.Apply(opts)
+
+	lbConfig := loadbalance.Config{
+		Resolver: options.Resolver,
+		Balancer: options.Balancer,
+		LbOpts:   options.LbOpts,
+	}
+
+	f := loadbalance.NewBalancerFactory(lbConfig)
+	return func(next client.Endpoint) client.Endpoint {
+		return func(ctx context.Context, req *protocol.Request, resp *protocol.Response) (err error) {
+			if req.Options() != nil && req.Options().IsSD() {
+				ins, err := f.GetInstance(ctx, req)
+				if err != nil {
+					return err
+				}
+				req.SetHost(ins.Address().String())
+			}
+			return next(ctx, req, resp)
+		}
+	}
+}
diff --git a/pkg/app/middlewares/client/sd/options.go b/pkg/app/middlewares/client/sd/options.go
new file mode 100644
index 0000000..b41d22c
--- /dev/null
+++ b/pkg/app/middlewares/client/sd/options.go
@@ -0,0 +1,96 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sd
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"strings"
+
+	"github.com/cloudwego/hertz/pkg/app/client/discovery"
+	"github.com/cloudwego/hertz/pkg/app/client/loadbalance"
+	"github.com/cloudwego/hertz/pkg/app/server/registry"
+)
+
+// ServiceDiscoveryOptions service discovery option for client
+type ServiceDiscoveryOptions struct {
+	// Resolver is used to client discovery
+	Resolver discovery.Resolver
+
+	// Balancer is used to client load balance
+	Balancer loadbalance.Loadbalancer
+
+	// LbOpts LoadBalance option
+	LbOpts loadbalance.Options
+}
+
+func (o *ServiceDiscoveryOptions) Apply(opts []ServiceDiscoveryOption) {
+	for _, op := range opts {
+		op.F(o)
+	}
+}
+
+type ServiceDiscoveryOption struct {
+	F func(o *ServiceDiscoveryOptions)
+}
+
+// WithCustomizedAddrs specifies the target instance addresses when doing service discovery.
+// It overwrites the results from the Resolver
+func WithCustomizedAddrs(addrs ...string) ServiceDiscoveryOption {
+	return ServiceDiscoveryOption{
+		F: func(o *ServiceDiscoveryOptions) {
+			var ins []discovery.Instance
+			for _, addr := range addrs {
+				if _, err := net.ResolveTCPAddr("tcp", addr); err == nil {
+					ins = append(ins, discovery.NewInstance("tcp", addr, registry.DefaultWeight, nil))
+					continue
+				}
+				if _, err := net.ResolveUnixAddr("unix", addr); err == nil {
+					ins = append(ins, discovery.NewInstance("unix", addr, registry.DefaultWeight, nil))
+					continue
+				}
+				panic(fmt.Errorf("WithCustomizedAddrs: invalid '%s'", addr))
+			}
+			if len(ins) == 0 {
+				panic("WithCustomizedAddrs() requires at least one argument")
+			}
+
+			targets := strings.Join(addrs, ",")
+			o.Resolver = &discovery.SynthesizedResolver{
+				ResolveFunc: func(ctx context.Context, key string) (discovery.Result, error) {
+					return discovery.Result{
+						CacheKey:  "fixed",
+						Instances: ins,
+					}, nil
+				},
+				NameFunc: func() string { return targets },
+				TargetFunc: func(ctx context.Context, target *discovery.TargetInfo) string {
+					return targets
+				},
+			}
+		},
+	}
+}
+
+// WithLoadBalanceOptions  sets Loadbalancer and loadbalance options for hertz client
+func WithLoadBalanceOptions(lb loadbalance.Loadbalancer, options loadbalance.Options) ServiceDiscoveryOption {
+	return ServiceDiscoveryOption{F: func(o *ServiceDiscoveryOptions) {
+		o.LbOpts = options
+		o.Balancer = lb
+	}}
+}
diff --git a/pkg/app/server/hertz.go b/pkg/app/server/hertz.go
index d6c4f02..96854a7 100644
--- a/pkg/app/server/hertz.go
+++ b/pkg/app/server/hertz.go
@@ -56,6 +56,8 @@ func Default(opts ...config.Option) *Hertz {
 // Spin runs the server until catching os.Signal or error returned by h.Run().
 func (h *Hertz) Spin() {
 	errCh := make(chan error)
+	h.initOnRunHooks(errCh)
+	h.initOnShutdownHooks()
 	go func() {
 		errCh <- h.Run()
 	}()
@@ -114,3 +116,30 @@ func waitSignal(errCh chan error) error {
 
 	return nil
 }
+
+func (h *Hertz) initOnRunHooks(errChan chan error) {
+	// add register func to runHooks
+	opt := h.GetOptions()
+	h.OnRun = append(h.OnRun, func(ctx context.Context) error {
+		go func() {
+			// delay register 1s
+			time.Sleep(1 * time.Second)
+			if err := opt.Registry.Register(opt.RegistryInfo); err != nil {
+				hlog.Errorf("HERTZ: Register error=%v", err)
+				// pass err to errChan
+				errChan <- err
+			}
+		}()
+		return nil
+	})
+}
+
+func (h *Hertz) initOnShutdownHooks() {
+	opt := h.GetOptions()
+	// add deregister func to shutdownHooks
+	h.OnShutdown = append(h.OnShutdown, func(ctx context.Context) {
+		if err := opt.Registry.Deregister(opt.RegistryInfo); err != nil {
+			hlog.Errorf("HERTZ: Deregister error=%v", err)
+		}
+	})
+}
diff --git a/pkg/app/server/option.go b/pkg/app/server/option.go
index 94ab8fa..c67ddcc 100644
--- a/pkg/app/server/option.go
+++ b/pkg/app/server/option.go
@@ -21,6 +21,7 @@ import (
 	"net"
 	"time"
 
+	"github.com/cloudwego/hertz/pkg/app/server/registry"
 	"github.com/cloudwego/hertz/pkg/common/config"
 	"github.com/cloudwego/hertz/pkg/common/tracer"
 	"github.com/cloudwego/hertz/pkg/common/tracer/stats"
@@ -268,3 +269,11 @@ func WithTraceLevel(level stats.Level) config.Option {
 		o.TraceLevel = level
 	}}
 }
+
+// WithRegistry sets the registry and registry's info
+func WithRegistry(r registry.Registry, info *registry.Info) config.Option {
+	return config.Option{F: func(o *config.Options) {
+		o.Registry = r
+		o.RegistryInfo = info
+	}}
+}
diff --git a/pkg/app/server/registry/registry.go b/pkg/app/server/registry/registry.go
new file mode 100644
index 0000000..8bf2991
--- /dev/null
+++ b/pkg/app/server/registry/registry.go
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package registry
+
+import "net"
+
+const (
+	DefaultWeight = 10
+)
+
+// Registry is extension interface of service registry.
+type Registry interface {
+	Register(info *Info) error
+	Deregister(info *Info) error
+}
+
+// Info is used for registry.
+// The fields are just suggested, which is used depends on design.
+type Info struct {
+	// ServiceName will be set in hertz by default
+	ServiceName string
+	// Addr will be set in hertz by default
+	Addr net.Addr
+	// Weight will be set in hertz by default
+	Weight int
+	// extend other infos with Tags.
+	Tags map[string]string
+}
+
+// NoopRegistry is an empty implement of Registry
+var NoopRegistry Registry = &noopRegistry{}
+
+// NoopRegistry
+type noopRegistry struct{}
+
+func (e noopRegistry) Register(*Info) error {
+	return nil
+}
+
+func (e noopRegistry) Deregister(*Info) error {
+	return nil
+}
diff --git a/pkg/common/config/option.go b/pkg/common/config/option.go
index d1c1233..868b476 100644
--- a/pkg/common/config/option.go
+++ b/pkg/common/config/option.go
@@ -20,6 +20,8 @@ import (
 	"crypto/tls"
 	"net"
 	"time"
+
+	"github.com/cloudwego/hertz/pkg/app/server/registry"
 )
 
 // Option is the only struct that can be used to set Options.
@@ -64,6 +66,11 @@ type Options struct {
 	Tracers                      []interface{}
 	TraceLevel                   interface{}
 	ListenConfig                 *net.ListenConfig
+
+	// Registry is used for service registry.
+	Registry registry.Registry
+	// RegistryInfo is base info used for service registry.
+	RegistryInfo *registry.Info
 }
 
 func (o *Options) Apply(opts []Option) {
@@ -183,6 +190,8 @@ func NewOptions(opts []Option) *Options {
 
 		// trace level, default LevelDetailed
 		TraceLevel: new(interface{}),
+
+		Registry: registry.NoopRegistry,
 	}
 	options.Apply(opts)
 	return options
diff --git a/pkg/common/config/request_option.go b/pkg/common/config/request_option.go
index 4af8f01..8d4bc5a 100644
--- a/pkg/common/config/request_option.go
+++ b/pkg/common/config/request_option.go
@@ -65,6 +65,10 @@ func (o *RequestOptions) Tag(k string) string {
 	return o.tags[k]
 }
 
+func (o *RequestOptions) Tags() map[string]string {
+	return o.tags
+}
+
 func (o *RequestOptions) IsSD() bool {
 	return o.isSD
 }
@@ -81,7 +85,7 @@ func (o *RequestOptions) CopyTo(dst *RequestOptions) {
 	dst.isSD = o.isSD
 }
 
-// Pre define some RequestOption here
+// SetPreDefinedOpts Pre define some RequestOption here
 func SetPreDefinedOpts(opts ...RequestOption) {
 	preDefinedOpts = nil
 	preDefinedOpts = append(preDefinedOpts, opts...)
diff --git a/pkg/common/utils/netaddr.go b/pkg/common/utils/netaddr.go
new file mode 100644
index 0000000..0c0d3dd
--- /dev/null
+++ b/pkg/common/utils/netaddr.go
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2021 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package utils
+
+import "net"
+
+var _ net.Addr = &NetAddr{}
+
+// NetAddr implements the net.Addr interface.
+type NetAddr struct {
+	network string
+	address string
+}
+
+// NewNetAddr creates a new NetAddr object with the network and address provided.
+func NewNetAddr(network, address string) net.Addr {
+	return &NetAddr{network, address}
+}
+
+// Network implements the net.Addr interface.
+func (na *NetAddr) Network() string {
+	return na.network
+}
+
+// String implements the net.Addr interface.
+func (na *NetAddr) String() string {
+	return na.address
+}
diff --git a/pkg/network/standard/transport.go b/pkg/network/standard/transport.go
index 666ff4d..2739830 100644
--- a/pkg/network/standard/transport.go
+++ b/pkg/network/standard/transport.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"crypto/tls"
 	"net"
+	"sync"
 	"time"
 
 	"github.com/cloudwego/hertz/pkg/common/config"
@@ -44,16 +45,18 @@ type transport struct {
 	ln               net.Listener
 	tls              *tls.Config
 	listenConfig     *net.ListenConfig
+	lock             sync.Mutex
 }
 
 func (t *transport) serve() (err error) {
 	network.UnlinkUdsFile(t.network, t.addr) //nolint:errcheck
-
+	t.lock.Lock()
 	if t.listenConfig != nil {
 		t.ln, err = t.listenConfig.Listen(context.Background(), t.network, t.addr)
 	} else {
 		t.ln, err = net.Listen(t.network, t.addr)
 	}
+	t.lock.Unlock()
 	if err != nil {
 		return err
 	}
@@ -88,9 +91,11 @@ func (t *transport) Shutdown(ctx context.Context) error {
 	defer func() {
 		network.UnlinkUdsFile(t.network, t.addr) //nolint:errcheck
 	}()
+	t.lock.Lock()
 	if t.ln != nil {
 		_ = t.ln.Close()
 	}
+	t.lock.Unlock()
 	<-ctx.Done()
 	return nil
 }
diff --git a/pkg/route/engine.go b/pkg/route/engine.go
index 91151d5..d6c12fb 100644
--- a/pkg/route/engine.go
+++ b/pkg/route/engine.go
@@ -94,6 +94,8 @@ type hijackConn struct {
 
 type CtxCallback func(ctx context.Context)
 
+type CtxErrCallback func(ctx context.Context) error
+
 // RouteInfo represents a request route's specification which contains method and path and its handler.
 type RouteInfo struct {
 	Method      string
@@ -177,7 +179,10 @@ type Engine struct {
 	// Indicates the engine status (Init/Running/Shutdown/Closed).
 	status uint32
 
-	// Hook functions get triggered sequentially before engine start to shutdown
+	// Hook functions get triggered sequentially when engine start
+	OnRun []CtxErrCallback
+
+	// Hook functions get triggered simultaneously when engine shutdown
 	OnShutdown []CtxCallback
 }
 
@@ -252,7 +257,7 @@ func (engine *Engine) NewContext() *app.RequestContext {
 //	one request (in hand or next incoming), idleTimeout or ExitWaitTime
 // 4. Exit
 func (engine *Engine) Shutdown(ctx context.Context) (err error) {
-	if engine.status != statusRunning {
+	if atomic.LoadUint32(&engine.status) != statusRunning {
 		return errStatusNotRunning
 	}
 	if !atomic.CompareAndSwapUint32(&engine.status, statusRunning, statusShutdown) {
@@ -283,6 +288,14 @@ func (engine *Engine) Run() (err error) {
 	}
 	defer atomic.StoreUint32(&engine.status, statusClosed)
 
+	// trigger hooks if any
+	ctx := context.Background()
+	for i := range engine.OnRun {
+		if err = engine.OnRun[i](ctx); err != nil {
+			return err
+		}
+	}
+
 	return engine.listenAndServe()
 }
 


Test patch:
diff --git a/pkg/app/client/loadbalance/lbcache_test.go b/pkg/app/client/loadbalance/lbcache_test.go
new file mode 100644
index 0000000..af7c6fe
--- /dev/null
+++ b/pkg/app/client/loadbalance/lbcache_test.go
@@ -0,0 +1,177 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package loadbalance
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/cloudwego/hertz/pkg/app/client/discovery"
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+	"github.com/cloudwego/hertz/pkg/protocol"
+)
+
+func TestBuilder(t *testing.T) {
+	ins := discovery.NewInstance("tcp", "127.0.0.1:8888", 10, map[string]string{"a": "b"})
+	r := &discovery.SynthesizedResolver{
+		ResolveFunc: func(ctx context.Context, key string) (discovery.Result, error) {
+			return discovery.Result{CacheKey: key, Instances: []discovery.Instance{ins}}, nil
+		},
+		TargetFunc: func(ctx context.Context, target *discovery.TargetInfo) string {
+			return "mockRoute"
+		},
+		NameFunc: func() string { return t.Name() },
+	}
+	lb := mockLoadbalancer{
+		rebalanceFunc: nil,
+		deleteFunc:    nil,
+		pickFunc: func(res discovery.Result) discovery.Instance {
+			assert.Assert(t, res.CacheKey == t.Name()+":mockRoute", res.CacheKey)
+			assert.Assert(t, len(res.Instances) == 1)
+			assert.Assert(t, len(res.Instances) == 1)
+			assert.Assert(t, res.Instances[0].Address().String() == "127.0.0.1:8888")
+			return res.Instances[0]
+		},
+		nameFunc: func() string { return "Synthesized" },
+	}
+	NewBalancerFactory(Config{
+		Balancer: lb,
+		LbOpts:   DefaultLbOpts,
+		Resolver: r,
+	})
+	b, ok := balancerFactories.Load(cacheKey(t.Name(), "Synthesized", DefaultLbOpts))
+	assert.Assert(t, ok)
+	assert.Assert(t, b != nil)
+	req := &protocol.Request{}
+	req.SetHost("hertz.api.test")
+	ins1, err := b.(*BalancerFactory).GetInstance(context.TODO(), req)
+	assert.Assert(t, err == nil)
+	assert.Assert(t, ins1.Address().String() == "127.0.0.1:8888")
+	assert.Assert(t, ins1.Weight() == 10)
+	value, exists := ins1.Tag("a")
+	assert.Assert(t, value == "b")
+	assert.Assert(t, exists == true)
+}
+
+func TestBalancerCache(t *testing.T) {
+	count := 10
+	inss := []discovery.Instance{}
+	for i := 0; i < count; i++ {
+		inss = append(inss, discovery.NewInstance("tcp", fmt.Sprint(i), 10, nil))
+	}
+	r := &discovery.SynthesizedResolver{
+		TargetFunc: func(ctx context.Context, target *discovery.TargetInfo) string {
+			return target.Host
+		},
+		ResolveFunc: func(ctx context.Context, key string) (discovery.Result, error) {
+			return discovery.Result{CacheKey: "svc", Instances: inss}, nil
+		},
+		NameFunc: func() string { return t.Name() },
+	}
+	lb := NewWeightedBalancer()
+	for i := 0; i < count; i++ {
+		blf := NewBalancerFactory(Config{
+			Balancer: lb,
+			LbOpts:   Options{},
+			Resolver: r,
+		})
+		req := &protocol.Request{}
+		req.SetHost("svc")
+		for a := 0; a < count; a++ {
+			addr, err := blf.GetInstance(context.TODO(), req)
+			assert.Assert(t, err == nil, err)
+			t.Logf("count: %d addr: %s\n", i, addr.Address().String())
+		}
+	}
+}
+
+func TestBalancerRefresh(t *testing.T) {
+	var ins atomic.Value
+	ins.Store(discovery.NewInstance("tcp", "127.0.0.1:8888", 10, nil))
+	r := &discovery.SynthesizedResolver{
+		TargetFunc: func(ctx context.Context, target *discovery.TargetInfo) string {
+			return target.Host
+		},
+		ResolveFunc: func(ctx context.Context, key string) (discovery.Result, error) {
+			return discovery.Result{CacheKey: "svc1", Instances: []discovery.Instance{ins.Load().(discovery.Instance)}}, nil
+		},
+		NameFunc: func() string { return t.Name() },
+	}
+	blf := NewBalancerFactory(Config{
+		Balancer: NewWeightedBalancer(),
+		LbOpts:   DefaultLbOpts,
+		Resolver: r,
+	})
+	req := &protocol.Request{}
+	req.SetHost("svc1")
+	addr, err := blf.GetInstance(context.Background(), req)
+	assert.Assert(t, err == nil, err)
+	assert.Assert(t, addr.Address().String() == "127.0.0.1:8888")
+	ins.Store(discovery.NewInstance("tcp", "127.0.0.1:8889", 10, nil))
+	addr, err = blf.GetInstance(context.Background(), req)
+	assert.Assert(t, err == nil, err)
+	assert.Assert(t, addr.Address().String() == "127.0.0.1:8888")
+	time.Sleep(6 * time.Second)
+	addr, err = blf.GetInstance(context.Background(), req)
+	assert.Assert(t, err == nil, err)
+	assert.Assert(t, addr.Address().String() == "127.0.0.1:8889")
+}
+
+func TestCacheKey(t *testing.T) {
+	uniqueKey := cacheKey("hello", "world", Options{RefreshInterval: 15 * time.Second, ExpireInterval: 5 * time.Minute})
+	assert.Assert(t, uniqueKey == "hello|world|{15s 5m0s}")
+}
+
+type mockLoadbalancer struct {
+	rebalanceFunc func(ch discovery.Result)
+	deleteFunc    func(key string)
+	pickFunc      func(discovery.Result) discovery.Instance
+	nameFunc      func() string
+}
+
+// Rebalance implements the Loadbalancer interface.
+func (m mockLoadbalancer) Rebalance(ch discovery.Result) {
+	if m.rebalanceFunc != nil {
+		m.rebalanceFunc(ch)
+	}
+}
+
+// Delete implements the Loadbalancer interface.
+func (m mockLoadbalancer) Delete(ch string) {
+	if m.deleteFunc != nil {
+		m.deleteFunc(ch)
+	}
+}
+
+// Name implements the Loadbalancer interface.
+func (m mockLoadbalancer) Name() string {
+	if m.nameFunc != nil {
+		return m.nameFunc()
+	}
+	return ""
+}
+
+// Pick implements the Loadbalancer interface.
+func (m mockLoadbalancer) Pick(d discovery.Result) discovery.Instance {
+	if m.pickFunc != nil {
+		return m.pickFunc(d)
+	}
+	return nil
+}
diff --git a/pkg/app/client/loadbalance/weight_random_test.go b/pkg/app/client/loadbalance/weight_random_test.go
new file mode 100644
index 0000000..1cab4cb
--- /dev/null
+++ b/pkg/app/client/loadbalance/weight_random_test.go
@@ -0,0 +1,111 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package loadbalance
+
+import (
+	"math"
+	"testing"
+
+	"github.com/cloudwego/hertz/pkg/app/client/discovery"
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+)
+
+func TestWeightedBalancer(t *testing.T) {
+	balancer := NewWeightedBalancer()
+	// nil
+	ins := balancer.Pick(discovery.Result{})
+	assert.DeepEqual(t, ins, nil)
+
+	// empty instance
+	e := discovery.Result{
+		Instances: make([]discovery.Instance, 0),
+		CacheKey:  "a",
+	}
+	balancer.Rebalance(e)
+	ins = balancer.Pick(e)
+	assert.DeepEqual(t, ins, nil)
+
+	// one instance
+	insList := []discovery.Instance{
+		discovery.NewInstance("tcp", "127.0.0.1:8888", 20, nil),
+	}
+	e = discovery.Result{
+		Instances: insList,
+		CacheKey:  "b",
+	}
+	balancer.Rebalance(e)
+	for i := 0; i < 100; i++ {
+		ins = balancer.Pick(e)
+		assert.DeepEqual(t, ins.Weight(), 20)
+	}
+
+	// multi instances, weightSum > 0
+	insList = []discovery.Instance{
+		discovery.NewInstance("tcp", "127.0.0.1:8881", 10, nil),
+		discovery.NewInstance("tcp", "127.0.0.1:8882", 20, nil),
+		discovery.NewInstance("tcp", "127.0.0.1:8883", 50, nil),
+		discovery.NewInstance("tcp", "127.0.0.1:8884", 100, nil),
+		discovery.NewInstance("tcp", "127.0.0.1:8885", 200, nil),
+		discovery.NewInstance("tcp", "127.0.0.1:8886", 500, nil),
+	}
+
+	var weightSum int
+	for _, ins := range insList {
+		weight := ins.Weight()
+		weightSum += weight
+	}
+
+	n := 10000000
+	pickedStat := map[int]int{}
+	e = discovery.Result{
+		Instances: insList,
+		CacheKey:  "c",
+	}
+	balancer.Rebalance(e)
+	for i := 0; i < n; i++ {
+		ins = balancer.Pick(e)
+		weight := ins.Weight()
+		if pickedCnt, ok := pickedStat[weight]; ok {
+			pickedStat[weight] = pickedCnt + 1
+		} else {
+			pickedStat[weight] = 1
+		}
+	}
+
+	for _, ins := range insList {
+		weight := ins.Weight()
+		expect := float64(weight) / float64(weightSum) * float64(n)
+		actual := float64(pickedStat[weight])
+		delta := math.Abs(expect - actual)
+		assert.DeepEqual(t, true, delta/expect < 0.01)
+	}
+
+	// have instances that weight < 0
+	insList = []discovery.Instance{
+		discovery.NewInstance("tcp", "127.0.0.1:8881", 10, nil),
+		discovery.NewInstance("tcp", "127.0.0.1:8882", -10, nil),
+	}
+	e = discovery.Result{
+		Instances: insList,
+		CacheKey:  "d",
+	}
+	balancer.Rebalance(e)
+	for i := 0; i < 1000; i++ {
+		ins = balancer.Pick(e)
+		assert.DeepEqual(t, 10, ins.Weight())
+	}
+}
diff --git a/pkg/app/middlewares/client/sd/discovery_test.go b/pkg/app/middlewares/client/sd/discovery_test.go
new file mode 100644
index 0000000..a1c8c8d
--- /dev/null
+++ b/pkg/app/middlewares/client/sd/discovery_test.go
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sd
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cloudwego/hertz/pkg/app/client/discovery"
+	"github.com/cloudwego/hertz/pkg/common/config"
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+	"github.com/cloudwego/hertz/pkg/protocol"
+)
+
+func TestDiscovery(t *testing.T) {
+	inss := []discovery.Instance{
+		discovery.NewInstance("tcp", "127.0.0.1:8888", 10, nil),
+		discovery.NewInstance("tcp", "127.0.0.1:8889", 10, nil),
+	}
+	r := &discovery.SynthesizedResolver{
+		TargetFunc: func(ctx context.Context, target *discovery.TargetInfo) string {
+			return target.Host
+		},
+		ResolveFunc: func(ctx context.Context, key string) (discovery.Result, error) {
+			return discovery.Result{CacheKey: "svc1", Instances: inss}, nil
+		},
+		NameFunc: func() string { return t.Name() },
+	}
+
+	mw := Discovery(r)
+	checkMdw := func(ctx context.Context, req *protocol.Request, resp *protocol.Response) (err error) {
+		t.Log(string(req.Host()))
+		assert.Assert(t, string(req.Host()) == "127.0.0.1:8888" || string(req.Host()) == "127.0.0.1:8889")
+		return nil
+	}
+	for i := 0; i < 10; i++ {
+		req := &protocol.Request{}
+		resp := &protocol.Response{}
+		req.Options().Apply([]config.RequestOption{config.WithSD(true)})
+		req.SetRequestURI("http://service_name")
+		_ = mw(checkMdw)(context.Background(), req, resp)
+	}
+}
diff --git a/pkg/app/middlewares/client/sd/options_test.go b/pkg/app/middlewares/client/sd/options_test.go
new file mode 100644
index 0000000..44a7356
--- /dev/null
+++ b/pkg/app/middlewares/client/sd/options_test.go
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sd
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cloudwego/hertz/pkg/common/test/assert"
+)
+
+func TestWithCustomizedAddrs(t *testing.T) {
+	var options []ServiceDiscoveryOption
+	options = append(options, WithCustomizedAddrs("127.0.0.1:8080"))
+	opts := &ServiceDiscoveryOptions{}
+	opts.Apply(options)
+	assert.Assert(t, opts.Resolver.Name() == "127.0.0.1:8080")
+	res, err := opts.Resolver.Resolve(context.Background(), "")
+	assert.Assert(t, err == nil)
+	assert.Assert(t, res.Instances[0].Address().String() == "127.0.0.1:8080")
+}
diff --git a/pkg/app/server/hertz_test.go b/pkg/app/server/hertz_test.go
index b5ff4b5..0d27901 100644
--- a/pkg/app/server/hertz_test.go
+++ b/pkg/app/server/hertz_test.go
@@ -37,6 +37,8 @@ import (
 
 	"github.com/cloudwego/hertz/pkg/app"
 	c "github.com/cloudwego/hertz/pkg/app/client"
+	"github.com/cloudwego/hertz/pkg/app/server/registry"
+	"github.com/cloudwego/hertz/pkg/common/config"
 	errs "github.com/cloudwego/hertz/pkg/common/errors"
 	"github.com/cloudwego/hertz/pkg/common/test/assert"
 	"github.com/cloudwego/hertz/pkg/common/test/mock"
@@ -611,3 +613,126 @@ func TestReusePorts(t *testing.T) {
 		assert.DeepEqual(t, "{\"ping\":\"pong\"}", string(body))
 	}
 }
+
+func TestServiceRegisterFailed(t *testing.T) {
+	mockRegErr := errors.New("mock register error")
+	var rCount int32
+	var drCount int32
+	mockRegistry := MockRegistry{
+		RegisterFunc: func(info *registry.Info) error {
+			atomic.AddInt32(&rCount, 1)
+			return mockRegErr
+		},
+		DeregisterFunc: func(info *registry.Info) error {
+			atomic.AddInt32(&drCount, 1)
+			return nil
+		},
+	}
+	var opts []config.Option
+	opts = append(opts, WithRegistry(mockRegistry, nil))
+	opts = append(opts, WithHostPorts("127.0.0.1:9222"))
+	srv := New(opts...)
+	srv.Spin()
+	time.Sleep(2 * time.Second)
+	assert.Assert(t, atomic.LoadInt32(&rCount) == 1)
+}
+
+func TestServiceDeregisterFailed(t *testing.T) {
+	mockDeregErr := errors.New("mock deregister error")
+	var rCount int32
+	var drCount int32
+	mockRegistry := MockRegistry{
+		RegisterFunc: func(info *registry.Info) error {
+			atomic.AddInt32(&rCount, 1)
+			return nil
+		},
+		DeregisterFunc: func(info *registry.Info) error {
+			atomic.AddInt32(&drCount, 1)
+			return mockDeregErr
+		},
+	}
+	var opts []config.Option
+	opts = append(opts, WithRegistry(mockRegistry, nil))
+	opts = append(opts, WithHostPorts("127.0.0.1:9223"))
+	srv := New(opts...)
+	go srv.Spin()
+	time.Sleep(1 * time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	_ = srv.Shutdown(ctx)
+	time.Sleep(1 * time.Second)
+	assert.Assert(t, atomic.LoadInt32(&rCount) == 1)
+	assert.Assert(t, atomic.LoadInt32(&drCount) == 1)
+}
+
+func TestServiceRegistryInfo(t *testing.T) {
+	registryInfo := &registry.Info{
+		Weight:      100,
+		Tags:        map[string]string{"aa": "bb"},
+		ServiceName: "hertz.api.test",
+	}
+	checkInfo := func(info *registry.Info) {
+		assert.Assert(t, info.Weight == registryInfo.Weight)
+		assert.Assert(t, info.ServiceName == "hertz.api.test")
+		assert.Assert(t, len(info.Tags) == len(registryInfo.Tags), info.Tags)
+		assert.Assert(t, info.Tags["aa"] == registryInfo.Tags["aa"], info.Tags)
+	}
+	var rCount int32
+	var drCount int32
+	mockRegistry := MockRegistry{
+		RegisterFunc: func(info *registry.Info) error {
+			checkInfo(info)
+			atomic.AddInt32(&rCount, 1)
+			return nil
+		},
+		DeregisterFunc: func(info *registry.Info) error {
+			checkInfo(info)
+			atomic.AddInt32(&drCount, 1)
+			return nil
+		},
+	}
+	var opts []config.Option
+	opts = append(opts, WithRegistry(mockRegistry, registryInfo))
+	opts = append(opts, WithHostPorts("127.0.0.1:9225"))
+	srv := New(opts...)
+	go srv.Spin()
+	time.Sleep(2 * time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 0)
+	defer cancel()
+	_ = srv.Shutdown(ctx)
+	time.Sleep(2 * time.Second)
+	assert.Assert(t, atomic.LoadInt32(&rCount) == 1)
+	assert.Assert(t, atomic.LoadInt32(&drCount) == 1)
+}
+
+func TestServiceRegistryNoInitInfo(t *testing.T) {
+	checkInfo := func(info *registry.Info) {
+		assert.Assert(t, info == nil)
+	}
+	var rCount int32
+	var drCount int32
+	mockRegistry := MockRegistry{
+		RegisterFunc: func(info *registry.Info) error {
+			checkInfo(info)
+			atomic.AddInt32(&rCount, 1)
+			return nil
+		},
+		DeregisterFunc: func(info *registry.Info) error {
+			checkInfo(info)
+			atomic.AddInt32(&drCount, 1)
+			return nil
+		},
+	}
+	var opts []config.Option
+	opts = append(opts, WithRegistry(mockRegistry, nil))
+	opts = append(opts, WithHostPorts("127.0.0.1:9227"))
+	srv := New(opts...)
+	go srv.Spin()
+	time.Sleep(2 * time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 0)
+	defer cancel()
+	_ = srv.Shutdown(ctx)
+	time.Sleep(2 * time.Second)
+	assert.Assert(t, atomic.LoadInt32(&rCount) == 1)
+	assert.Assert(t, atomic.LoadInt32(&drCount) == 1)
+}
diff --git a/pkg/app/server/mocks_test.go b/pkg/app/server/mocks_test.go
new file mode 100644
index 0000000..b146913
--- /dev/null
+++ b/pkg/app/server/mocks_test.go
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package server
+
+import (
+	"github.com/cloudwego/hertz/pkg/app/server/registry"
+)
+
+var _ registry.Registry = &MockRegistry{}
+
+// MockRegistry is the mock implementation of registry.Registry interface.
+type MockRegistry struct {
+	RegisterFunc   func(info *registry.Info) error
+	DeregisterFunc func(info *registry.Info) error
+}
+
+// Register is the mock implementation of registry.Registry interface.
+func (m MockRegistry) Register(info *registry.Info) error {
+	if m.RegisterFunc != nil {
+		return m.RegisterFunc(info)
+	}
+	return nil
+}
+
+// Deregister is the mock implementation of registry.Registry interface.
+func (m MockRegistry) Deregister(info *registry.Info) error {
+	if m.DeregisterFunc != nil {
+		return m.DeregisterFunc(info)
+	}
+	return nil
+}
diff --git a/pkg/app/server/option_test.go b/pkg/app/server/option_test.go
index 4972698..4f3a885 100644
--- a/pkg/app/server/option_test.go
+++ b/pkg/app/server/option_test.go
@@ -20,12 +20,19 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cloudwego/hertz/pkg/app/server/registry"
 	"github.com/cloudwego/hertz/pkg/common/config"
 	"github.com/cloudwego/hertz/pkg/common/test/assert"
 	"github.com/cloudwego/hertz/pkg/common/tracer/stats"
+	"github.com/cloudwego/hertz/pkg/common/utils"
 )
 
 func TestOptions(t *testing.T) {
+	info := &registry.Info{
+		ServiceName: "hertz.test.api",
+		Weight:      10,
+		Addr:        utils.NewNetAddr("tcp", ":8888"),
+	}
 	opt := config.NewOptions([]config.Option{
 		WithReadTimeout(time.Second),
 		WithIdleTimeout(time.Second),
@@ -48,6 +55,7 @@ func TestOptions(t *testing.T) {
 		WithReadBufferSize(100),
 		WithALPN(true),
 		WithTraceLevel(stats.LevelDisabled),
+		WithRegistry(nil, info),
 	})
 	assert.DeepEqual(t, opt.ReadTimeout, time.Second)
 	assert.DeepEqual(t, opt.IdleTimeout, time.Second)
@@ -69,6 +77,8 @@ func TestOptions(t *testing.T) {
 	assert.DeepEqual(t, opt.ReadBufferSize, 100)
 	assert.DeepEqual(t, opt.ALPN, true)
 	assert.DeepEqual(t, opt.TraceLevel, stats.LevelDisabled)
+	assert.DeepEqual(t, opt.RegistryInfo, info)
+	assert.DeepEqual(t, opt.Registry, nil)
 }
 
 func TestDefaultOptions(t *testing.T) {
@@ -92,4 +102,6 @@ func TestDefaultOptions(t *testing.T) {
 	assert.DeepEqual(t, opt.H2C, false)
 	assert.DeepEqual(t, opt.ReadBufferSize, 4096)
 	assert.DeepEqual(t, opt.ALPN, false)
+	assert.DeepEqual(t, opt.Registry, registry.NoopRegistry)
+	assert.Assert(t, opt.RegistryInfo == nil)
 }