Repository: apache/rocketmq-client-go
Pull request #167 (2019-08-22 07:56:36)

[ISSUE #164]add UnCompress function implementation
## What is the purpose of the change

#164 add UnCompress function implementation

## Brief changelog

add UnCompress function implementation

## Verifying this change

Verified

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, `it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR`.

- [x] Make sure there is a [Github issue](https://github.com/apache/rocketmq/issues) filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue. 
- [x] Format the pull request title like `[ISSUE #123] Fix UnknownException when host config not exist`. Each commit in the pull request should have a meaningful subject line and body.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when a cross-module dependency exists.
- [ ] If this contribution is large, please file an [Apache Individual Contributor License Agreement](http://www.apache.org/licenses/#clas).


codecov-io: # [Codecov](https://codecov.io/gh/apache/rocketmq-client-go/pull/167?src=pr&el=h1) Report
> Merging [#167](https://codecov.io/gh/apache/rocketmq-client-go/pull/167?src=pr&el=desc) into [native](https://codecov.io/gh/apache/rocketmq-client-go/commit/6303ecd0878adb88b6ffb78587584aa6dd51b480?src=pr&el=desc) will **increase** coverage by `0.07%`.
> The diff coverage is `50%`.

[![Impacted file tree graph](https://codecov.io/gh/apache/rocketmq-client-go/pull/167/graphs/tree.svg?width=650&token=CiV9ofPkG3&height=150&src=pr)](https://codecov.io/gh/apache/rocketmq-client-go/pull/167?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           native     #167      +/-   ##
==========================================
+ Coverage   24.66%   24.74%   +0.07%     
==========================================
  Files          42       42              
  Lines        4524     4531       +7     
==========================================
+ Hits         1116     1121       +5     
  Misses       3259     3259              
- Partials      149      151       +2
```


| [Impacted Files](https://codecov.io/gh/apache/rocketmq-client-go/pull/167?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [internal/utils/helper.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/167/diff?src=pr&el=tree#diff-aW50ZXJuYWwvdXRpbHMvaGVscGVyLmdv) | `45.45% <50%> (+45.45%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/apache/rocketmq-client-go/pull/167?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/rocketmq-client-go/pull/167?src=pr&el=footer). Last update [6303ecd...388874d](https://codecov.io/gh/apache/rocketmq-client-go/pull/167?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/consumer/consumer.go b/consumer/consumer.go
index 2c88e5b..9bed213 100644
--- a/consumer/consumer.go
+++ b/consumer/consumer.go
@@ -263,6 +263,8 @@ type defaultConsumer struct {
 	storage               OffsetStore
 	// chan for push consumer
 	prCh chan PullRequest
+
+	namesrv internal.Namesrvs
 }
 
 func (dc *defaultConsumer) start() error {
@@ -277,7 +279,7 @@ func (dc *defaultConsumer) start() error {
 	//dc.client = internal.GetOrNewRocketMQClient(dc.option.ClientOptions, nil)
 	if dc.model == Clustering {
 		dc.option.ChangeInstanceNameToPID()
-		dc.storage = NewRemoteOffsetStore(dc.consumerGroup, dc.client)
+		dc.storage = NewRemoteOffsetStore(dc.consumerGroup, dc.client, dc.namesrv)
 	} else {
 		dc.storage = NewLocalFileOffsetStore(dc.consumerGroup, dc.client.ClientID())
 	}
@@ -425,7 +427,7 @@ type lockBatchRequestBody struct {
 }
 
 func (dc *defaultConsumer) lock(mq *primitive.MessageQueue) bool {
-	brokerResult := internal.FindBrokerAddressInSubscribe(mq.BrokerName, internal.MasterId, true)
+	brokerResult := dc.namesrv.FindBrokerAddressInSubscribe(mq.BrokerName, internal.MasterId, true)
 
 	if brokerResult == nil {
 		return false
@@ -456,7 +458,7 @@ func (dc *defaultConsumer) lock(mq *primitive.MessageQueue) bool {
 }
 
 func (dc *defaultConsumer) unlock(mq *primitive.MessageQueue, oneway bool) {
-	brokerResult := internal.FindBrokerAddressInSubscribe(mq.BrokerName, internal.MasterId, true)
+	brokerResult := dc.namesrv.FindBrokerAddressInSubscribe(mq.BrokerName, internal.MasterId, true)
 
 	if brokerResult == nil {
 		return
@@ -478,7 +480,7 @@ func (dc *defaultConsumer) lockAll() {
 		if len(mqs) == 0 {
 			continue
 		}
-		brokerResult := internal.FindBrokerAddressInSubscribe(broker, internal.MasterId, true)
+		brokerResult := dc.namesrv.FindBrokerAddressInSubscribe(broker, internal.MasterId, true)
 		if brokerResult == nil {
 			continue
 		}
@@ -520,7 +522,7 @@ func (dc *defaultConsumer) unlockAll(oneway bool) {
 		if len(mqs) == 0 {
 			continue
 		}
-		brokerResult := internal.FindBrokerAddressInSubscribe(broker, internal.MasterId, true)
+		brokerResult := dc.namesrv.FindBrokerAddressInSubscribe(broker, internal.MasterId, true)
 		if brokerResult == nil {
 			continue
 		}
@@ -738,7 +740,7 @@ func (dc *defaultConsumer) computePullFromWhere(mq *primitive.MessageQueue) int6
 func (dc *defaultConsumer) pullInner(ctx context.Context, queue *primitive.MessageQueue, data *internal.SubscriptionData,
 	offset int64, numbers int, sysFlag int32, commitOffsetValue int64) (*primitive.PullResult, error) {
 
-	brokerResult := tryFindBroker(queue)
+	brokerResult := dc.tryFindBroker(queue)
 	if brokerResult == nil {
 		rlog.Warnf("no broker found for %s", queue.String())
 		return nil, ErrBrokerNotFound
@@ -819,10 +821,10 @@ func (dc *defaultConsumer) processPullResult(mq *primitive.MessageQueue, result
 }
 
 func (dc *defaultConsumer) findConsumerList(topic string) []string {
-	brokerAddr := internal.FindBrokerAddrByTopic(topic)
+	brokerAddr := dc.namesrv.FindBrokerAddrByTopic(topic)
 	if brokerAddr == "" {
-		internal.UpdateTopicRouteInfo(topic)
-		brokerAddr = internal.FindBrokerAddrByTopic(topic)
+		dc.namesrv.UpdateTopicRouteInfo(topic)
+		brokerAddr = dc.namesrv.FindBrokerAddrByTopic(topic)
 	}
 
 	if brokerAddr != "" {
@@ -852,10 +854,10 @@ func (dc *defaultConsumer) sendBack(msg *primitive.MessageExt, level int) error
 
 // QueryMaxOffset with specific queueId and topic
 func (dc *defaultConsumer) queryMaxOffset(mq *primitive.MessageQueue) (int64, error) {
-	brokerAddr := internal.FindBrokerAddrByName(mq.BrokerName)
+	brokerAddr := dc.namesrv.FindBrokerAddrByName(mq.BrokerName)
 	if brokerAddr == "" {
-		internal.UpdateTopicRouteInfo(mq.Topic)
-		brokerAddr = internal.FindBrokerAddrByName(mq.Topic)
+		dc.namesrv.UpdateTopicRouteInfo(mq.Topic)
+		brokerAddr = dc.namesrv.FindBrokerAddrByName(mq.Topic)
 	}
 	if brokerAddr == "" {
 		return -1, fmt.Errorf("the broker [%s] does not exist", mq.BrokerName)
@@ -881,10 +883,10 @@ func (dc *defaultConsumer) queryOffset(mq *primitive.MessageQueue) int64 {
 
 // SearchOffsetByTimestamp with specific queueId and topic
 func (dc *defaultConsumer) searchOffsetByTimestamp(mq *primitive.MessageQueue, timestamp int64) (int64, error) {
-	brokerAddr := internal.FindBrokerAddrByName(mq.BrokerName)
+	brokerAddr := dc.namesrv.FindBrokerAddrByName(mq.BrokerName)
 	if brokerAddr == "" {
-		internal.UpdateTopicRouteInfo(mq.Topic)
-		brokerAddr = internal.FindBrokerAddrByName(mq.Topic)
+		dc.namesrv.UpdateTopicRouteInfo(mq.Topic)
+		brokerAddr = dc.namesrv.FindBrokerAddrByName(mq.Topic)
 	}
 	if brokerAddr == "" {
 		return -1, fmt.Errorf("the broker [%s] does not exist", mq.BrokerName)
@@ -962,13 +964,13 @@ func clearCommitOffsetFlag(sysFlag int32) int32 {
 	return sysFlag & (^0x1 << 0)
 }
 
-func tryFindBroker(mq *primitive.MessageQueue) *internal.FindBrokerResult {
-	result := internal.FindBrokerAddressInSubscribe(mq.BrokerName, recalculatePullFromWhichNode(mq), false)
+func (dc *defaultConsumer) tryFindBroker(mq *primitive.MessageQueue) *internal.FindBrokerResult {
+	result := dc.namesrv.FindBrokerAddressInSubscribe(mq.BrokerName, recalculatePullFromWhichNode(mq), false)
 
 	if result == nil {
-		internal.UpdateTopicRouteInfo(mq.Topic)
+		dc.namesrv.UpdateTopicRouteInfo(mq.Topic)
 	}
-	return internal.FindBrokerAddressInSubscribe(mq.BrokerName, recalculatePullFromWhichNode(mq), false)
+	return dc.namesrv.FindBrokerAddressInSubscribe(mq.BrokerName, recalculatePullFromWhichNode(mq), false)
 }
 
 var (
diff --git a/consumer/interceptor.go b/consumer/interceptor.go
index 657633e..8641d14 100644
--- a/consumer/interceptor.go
+++ b/consumer/interceptor.go
@@ -38,7 +38,7 @@ func WithTrace(traceCfg primitive.TraceConfig) Option {
 }
 
 func newTraceInterceptor(traceCfg primitive.TraceConfig) primitive.Interceptor {
-	dispatcher := internal.NewTraceDispatcher(traceCfg.TraceTopic, traceCfg.Access)
+	dispatcher := internal.NewTraceDispatcher(traceCfg.TraceTopic, traceCfg.Access, traceCfg.NamesrvAddrs)
 	dispatcher.Start()
 
 	return func(ctx context.Context, req, reply interface{}, next primitive.Invoker) error {
diff --git a/consumer/offset_store.go b/consumer/offset_store.go
index 1eb246c..f799c1a 100644
--- a/consumer/offset_store.go
+++ b/consumer/offset_store.go
@@ -185,13 +185,15 @@ type remoteBrokerOffsetStore struct {
 	group       string
 	OffsetTable map[string]map[int]*queueOffset `json:"OffsetTable"`
 	client      internal.RMQClient
+	namesrv     internal.Namesrvs
 	mutex       sync.RWMutex
 }
 
-func NewRemoteOffsetStore(group string, client internal.RMQClient) OffsetStore {
+func NewRemoteOffsetStore(group string, client internal.RMQClient, namesrv internal.Namesrvs) OffsetStore {
 	return &remoteBrokerOffsetStore{
 		group:       group,
 		client:      client,
+		namesrv:     namesrv,
 		OffsetTable: make(map[string]map[int]*queueOffset),
 	}
 }
@@ -285,10 +287,10 @@ func (r *remoteBrokerOffsetStore) update(mq *primitive.MessageQueue, offset int6
 }
 
 func (r *remoteBrokerOffsetStore) fetchConsumeOffsetFromBroker(group string, mq *primitive.MessageQueue) (int64, error) {
-	broker := internal.FindBrokerAddrByName(mq.BrokerName)
+	broker := r.namesrv.FindBrokerAddrByName(mq.BrokerName)
 	if broker == "" {
-		internal.UpdateTopicRouteInfo(mq.Topic)
-		broker = internal.FindBrokerAddrByName(mq.BrokerName)
+		r.namesrv.UpdateTopicRouteInfo(mq.Topic)
+		broker = r.namesrv.FindBrokerAddrByName(mq.BrokerName)
 	}
 	if broker == "" {
 		return int64(-1), fmt.Errorf("broker: %s address not found", mq.BrokerName)
@@ -317,10 +319,10 @@ func (r *remoteBrokerOffsetStore) fetchConsumeOffsetFromBroker(group string, mq
 }
 
 func (r *remoteBrokerOffsetStore) updateConsumeOffsetToBroker(group, topic string, queue *queueOffset) error {
-	broker := internal.FindBrokerAddrByName(queue.Broker)
+	broker := r.namesrv.FindBrokerAddrByName(queue.Broker)
 	if broker == "" {
-		internal.UpdateTopicRouteInfo(topic)
-		broker = internal.FindBrokerAddrByName(queue.Broker)
+		r.namesrv.UpdateTopicRouteInfo(topic)
+		broker = r.namesrv.FindBrokerAddrByName(queue.Broker)
 	}
 	if broker == "" {
 		return fmt.Errorf("broker: %s address not found", queue.Broker)
diff --git a/consumer/option.go b/consumer/option.go
index f92c118..933919b 100644
--- a/consumer/option.go
+++ b/consumer/option.go
@@ -167,11 +167,9 @@ func WithGroupName(group string) Option {
 }
 
 // WithNameServer set NameServer address, only support one NameServer cluster in alpha2
-func WithNameServer(nameServers []string) Option {
+func WithNameServer(nameServers primitive.NamesrvAddr) Option {
 	return func(opts *consumerOptions) {
-		if len(nameServers) > 0 {
-			opts.NameServerAddrs = nameServers
-		}
+		opts.NameServerAddrs = nameServers
 	}
 }
 
diff --git a/consumer/pull_consumer.go b/consumer/pull_consumer.go
index dad27a8..7d47b8a 100644
--- a/consumer/pull_consumer.go
+++ b/consumer/pull_consumer.go
@@ -23,11 +23,12 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/pkg/errors"
+
 	"github.com/apache/rocketmq-client-go/internal"
 	"github.com/apache/rocketmq-client-go/internal/utils"
 	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
-	"github.com/pkg/errors"
 )
 
 type PullConsumer interface {
@@ -75,11 +76,10 @@ func NewPullConsumer(options ...Option) (*defaultPullConsumer, error) {
 		apply(&defaultOpts)
 	}
 
-	srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs...)
+	srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs)
 	if err != nil {
 		return nil, errors.Wrap(err, "new Namesrv failed.")
 	}
-	internal.RegisterNamsrv(srvs)
 
 	dc := &defaultConsumer{
 		consumerGroup: defaultOpts.GroupName,
@@ -88,6 +88,8 @@ func NewPullConsumer(options ...Option) (*defaultPullConsumer, error) {
 		prCh:          make(chan PullRequest, 4),
 		model:         defaultOpts.ConsumerModel,
 		option:        defaultOpts,
+
+		namesrv: srvs,
 	}
 
 	c := &defaultPullConsumer{
@@ -128,7 +130,7 @@ func (c *defaultPullConsumer) Pull(ctx context.Context, topic string, selector M
 }
 
 func (c *defaultPullConsumer) getNextQueueOf(topic string) *primitive.MessageQueue {
-	queues, err := internal.FetchSubscribeMessageQueues(topic)
+	queues, err := c.defaultConsumer.namesrv.FetchSubscribeMessageQueues(topic)
 	if err != nil && len(queues) > 0 {
 		rlog.Error(err.Error())
 		return nil
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index 6303ded..d5a5ffc 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -24,11 +24,12 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"github.com/apache/rocketmq-client-go/internal"
 	"github.com/apache/rocketmq-client-go/internal/remote"
 	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
-	"github.com/pkg/errors"
 )
 
 // In most scenarios, this is the mostly recommended usage to consume messages.
@@ -60,17 +61,19 @@ func NewPushConsumer(opts ...Option) (*pushConsumer, error) {
 	for _, apply := range opts {
 		apply(&defaultOpts)
 	}
-	srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs...)
+	srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs)
 	if err != nil {
 		return nil, errors.Wrap(err, "new Namesrv failed.")
 	}
 	if !defaultOpts.Credentials.IsEmpty() {
 		srvs.SetCredentials(defaultOpts.Credentials)
 	}
-	internal.RegisterNamsrv(srvs)
+	defaultOpts.Namesrv = srvs
+
 	if defaultOpts.Namespace != "" {
 		defaultOpts.GroupName = defaultOpts.Namespace + "%" + defaultOpts.GroupName
 	}
+
 	dc := &defaultConsumer{
 		client:         internal.GetOrNewRocketMQClient(defaultOpts.ClientOptions, nil),
 		consumerGroup:  defaultOpts.GroupName,
@@ -82,6 +85,7 @@ func NewPushConsumer(opts ...Option) (*pushConsumer, error) {
 		fromWhere:      defaultOpts.FromWhere,
 		allocate:       defaultOpts.Strategy,
 		option:         defaultOpts,
+		namesrv:        srvs,
 	}
 
 	p := &pushConsumer{
@@ -435,7 +439,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 		//	pullRequest.SubVersion = data.SubVersion
 		//}
 
-		brokerResult := tryFindBroker(request.mq)
+		brokerResult := pc.defaultConsumer.tryFindBroker(request.mq)
 		if brokerResult == nil {
 			rlog.Warnf("no broker found for %s", request.mq.String())
 			sleepTime = _PullDelayTimeWhenError
@@ -507,7 +511,7 @@ func (pc *pushConsumer) correctTagsOffset(pr *PullRequest) {
 func (pc *pushConsumer) sendMessageBack(brokerName string, msg *primitive.MessageExt, delayLevel int) bool {
 	var brokerAddr string
 	if len(brokerName) != 0 {
-		brokerAddr = internal.FindBrokerAddrByName(brokerName)
+		brokerAddr = pc.defaultConsumer.namesrv.FindBrokerAddrByName(brokerName)
 	} else {
 		brokerAddr = msg.StoreHost
 	}
diff --git a/errors.go b/errors.go
index 6b98475..8062d86 100644
--- a/errors.go
+++ b/errors.go
@@ -19,6 +19,7 @@ package rocketmq
 
 import (
 	"github.com/apache/rocketmq-client-go/rlog"
+
 	"github.com/pkg/errors"
 )
 
diff --git a/examples/consumer/trace/main.go b/examples/consumer/trace/main.go
index 9102175..23b09ab 100644
--- a/examples/consumer/trace/main.go
+++ b/examples/consumer/trace/main.go
@@ -31,7 +31,8 @@ import (
 func main() {
 	namesrvs := []string{"127.0.0.1:9876"}
 	traceCfg := primitive.TraceConfig{
-		Access: primitive.Local,
+		Access:       primitive.Local,
+		NamesrvAddrs: namesrvs,
 	}
 
 	c, _ := rocketmq.NewPushConsumer(
diff --git a/examples/producer/trace/main.go b/examples/producer/trace/main.go
index ac6983f..e8fd386 100644
--- a/examples/producer/trace/main.go
+++ b/examples/producer/trace/main.go
@@ -31,7 +31,8 @@ import (
 func main() {
 	namesrvs := []string{"127.0.0.1:9876"}
 	traceCfg := primitive.TraceConfig{
-		Access: primitive.Local,
+		Access:       primitive.Local,
+		NamesrvAddrs: namesrvs,
 	}
 
 	p, _ := rocketmq.NewProducer(
diff --git a/internal/client.go b/internal/client.go
index 11eca1e..1307ff2 100644
--- a/internal/client.go
+++ b/internal/client.go
@@ -96,7 +96,8 @@ func DefaultClientOptions() ClientOptions {
 
 type ClientOptions struct {
 	GroupName         string
-	NameServerAddrs   []string
+	NameServerAddrs   primitive.NamesrvAddr
+	Namesrv           *namesrvs
 	ClientIP          string
 	InstanceName      string
 	UnitMode          bool
@@ -162,6 +163,7 @@ type rmqClient struct {
 	remoteClient *remote.RemotingClient
 	hbMutex      sync.Mutex
 	close        bool
+	namesrvs     *namesrvs
 }
 
 var clientMap sync.Map
@@ -170,6 +172,7 @@ func GetOrNewRocketMQClient(option ClientOptions, callbackCh chan interface{}) *
 	client := &rmqClient{
 		option:       option,
 		remoteClient: remote.NewRemotingClient(),
+		namesrvs:     option.Namesrv,
 	}
 	actual, loaded := clientMap.LoadOrStore(client.ClientID(), client)
 	if !loaded {
@@ -244,7 +247,7 @@ func (c *rmqClient) Start() {
 		// TODO cleanOfflineBroker & sendHeartbeatToAllBrokerWithLock
 		go func() {
 			for !c.close {
-				cleanOfflineBroker()
+				c.namesrvs.cleanOfflineBroker()
 				c.SendHeartbeatToAllBrokerWithLock()
 				time.Sleep(_HeartbeatBrokerInterval)
 			}
@@ -349,7 +352,7 @@ func (c *rmqClient) SendHeartbeatToAllBrokerWithLock() {
 		rlog.Info("sending heartbeat, but no producer and no consumer")
 		return
 	}
-	brokerAddressesMap.Range(func(key, value interface{}) bool {
+	c.namesrvs.brokerAddressesMap.Range(func(key, value interface{}) bool {
 		brokerName := key.(string)
 		data := value.(*BrokerData)
 		for id, addr := range data.BrokerAddresses {
@@ -360,13 +363,13 @@ func (c *rmqClient) SendHeartbeatToAllBrokerWithLock() {
 				return true
 			}
 			if response.Code == ResSuccess {
-				v, exist := brokerVersionMap.Load(brokerName)
+				v, exist := c.namesrvs.brokerVersionMap.Load(brokerName)
 				var m map[string]int32
 				if exist {
 					m = v.(map[string]int32)
 				} else {
 					m = make(map[string]int32, 4)
-					brokerVersionMap.Store(brokerName, m)
+					c.namesrvs.brokerVersionMap.Store(brokerName, m)
 				}
 				m[brokerName] = int32(response.Version)
 				rlog.Debugf("send heart beat to broker[%s %d %s] success", brokerName, id, addr)
@@ -387,7 +390,7 @@ func (c *rmqClient) UpdateTopicRouteInfo() {
 		return true
 	})
 	for topic := range publishTopicSet {
-		c.UpdatePublishInfo(topic, UpdateTopicRouteInfo(topic))
+		c.UpdatePublishInfo(topic, c.namesrvs.UpdateTopicRouteInfo(topic))
 	}
 
 	subscribedTopicSet := make(map[string]bool, 0)
@@ -401,7 +404,7 @@ func (c *rmqClient) UpdateTopicRouteInfo() {
 	})
 
 	for topic := range subscribedTopicSet {
-		c.updateSubscribeInfo(topic, UpdateTopicRouteInfo(topic))
+		c.updateSubscribeInfo(topic, c.namesrvs.UpdateTopicRouteInfo(topic))
 	}
 }
 
@@ -567,7 +570,7 @@ func (c *rmqClient) UpdatePublishInfo(topic string, data *TopicRouteData) {
 	}
 	c.producerMap.Range(func(key, value interface{}) bool {
 		p := value.(InnerProducer)
-		publishInfo := routeData2PublishInfo(topic, data)
+		publishInfo := c.namesrvs.routeData2PublishInfo(topic, data)
 		publishInfo.HaveTopicRouterInfo = true
 		p.UpdateTopicPublishInfo(topic, publishInfo)
 		return true
diff --git a/internal/mock_namesrv.go b/internal/mock_namesrv.go
new file mode 100644
index 0000000..6333e48
--- /dev/null
+++ b/internal/mock_namesrv.go
@@ -0,0 +1,128 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: namesrv.go
+
+// Package internal is a generated GoMock package.
+package internal
+
+import (
+	primitive "github.com/apache/rocketmq-client-go/primitive"
+	gomock "github.com/golang/mock/gomock"
+	reflect "reflect"
+)
+
+// MockNamesrvs is a mock of Namesrvs interface
+type MockNamesrvs struct {
+	ctrl     *gomock.Controller
+	recorder *MockNamesrvsMockRecorder
+}
+
+// MockNamesrvsMockRecorder is the mock recorder for MockNamesrvs
+type MockNamesrvsMockRecorder struct {
+	mock *MockNamesrvs
+}
+
+// NewMockNamesrvs creates a new mock instance
+func NewMockNamesrvs(ctrl *gomock.Controller) *MockNamesrvs {
+	mock := &MockNamesrvs{ctrl: ctrl}
+	mock.recorder = &MockNamesrvsMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockNamesrvs) EXPECT() *MockNamesrvsMockRecorder {
+	return m.recorder
+}
+
+// AddBroker mocks base method
+func (m *MockNamesrvs) AddBroker(routeData *TopicRouteData) {
+	m.ctrl.Call(m, "AddBroker", routeData)
+}
+
+// AddBroker indicates an expected call of AddBroker
+func (mr *MockNamesrvsMockRecorder) AddBroker(routeData interface{}) *gomock.Call {
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddBroker", reflect.TypeOf((*MockNamesrvs)(nil).AddBroker), routeData)
+}
+
+// cleanOfflineBroker mocks base method
+func (m *MockNamesrvs) cleanOfflineBroker() {
+	m.ctrl.Call(m, "cleanOfflineBroker")
+}
+
+// cleanOfflineBroker indicates an expected call of cleanOfflineBroker
+func (mr *MockNamesrvsMockRecorder) cleanOfflineBroker() *gomock.Call {
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "cleanOfflineBroker", reflect.TypeOf((*MockNamesrvs)(nil).cleanOfflineBroker))
+}
+
+// UpdateTopicRouteInfo mocks base method
+func (m *MockNamesrvs) UpdateTopicRouteInfo(topic string) *TopicRouteData {
+	ret := m.ctrl.Call(m, "UpdateTopicRouteInfo", topic)
+	ret0, _ := ret[0].(*TopicRouteData)
+	return ret0
+}
+
+// UpdateTopicRouteInfo indicates an expected call of UpdateTopicRouteInfo
+func (mr *MockNamesrvsMockRecorder) UpdateTopicRouteInfo(topic interface{}) *gomock.Call {
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateTopicRouteInfo", reflect.TypeOf((*MockNamesrvs)(nil).UpdateTopicRouteInfo), topic)
+}
+
+// FetchPublishMessageQueues mocks base method
+func (m *MockNamesrvs) FetchPublishMessageQueues(topic string) ([]*primitive.MessageQueue, error) {
+	ret := m.ctrl.Call(m, "FetchPublishMessageQueues", topic)
+	ret0, _ := ret[0].([]*primitive.MessageQueue)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// FetchPublishMessageQueues indicates an expected call of FetchPublishMessageQueues
+func (mr *MockNamesrvsMockRecorder) FetchPublishMessageQueues(topic interface{}) *gomock.Call {
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FetchPublishMessageQueues", reflect.TypeOf((*MockNamesrvs)(nil).FetchPublishMessageQueues), topic)
+}
+
+// FindBrokerAddrByTopic mocks base method
+func (m *MockNamesrvs) FindBrokerAddrByTopic(topic string) string {
+	ret := m.ctrl.Call(m, "FindBrokerAddrByTopic", topic)
+	ret0, _ := ret[0].(string)
+	return ret0
+}
+
+// FindBrokerAddrByTopic indicates an expected call of FindBrokerAddrByTopic
+func (mr *MockNamesrvsMockRecorder) FindBrokerAddrByTopic(topic interface{}) *gomock.Call {
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FindBrokerAddrByTopic", reflect.TypeOf((*MockNamesrvs)(nil).FindBrokerAddrByTopic), topic)
+}
+
+// FindBrokerAddrByName mocks base method
+func (m *MockNamesrvs) FindBrokerAddrByName(brokerName string) string {
+	ret := m.ctrl.Call(m, "FindBrokerAddrByName", brokerName)
+	ret0, _ := ret[0].(string)
+	return ret0
+}
+
+// FindBrokerAddrByName indicates an expected call of FindBrokerAddrByName
+func (mr *MockNamesrvsMockRecorder) FindBrokerAddrByName(brokerName interface{}) *gomock.Call {
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FindBrokerAddrByName", reflect.TypeOf((*MockNamesrvs)(nil).FindBrokerAddrByName), brokerName)
+}
+
+// FindBrokerAddressInSubscribe mocks base method
+func (m *MockNamesrvs) FindBrokerAddressInSubscribe(brokerName string, brokerId int64, onlyThisBroker bool) *FindBrokerResult {
+	ret := m.ctrl.Call(m, "FindBrokerAddressInSubscribe", brokerName, brokerId, onlyThisBroker)
+	ret0, _ := ret[0].(*FindBrokerResult)
+	return ret0
+}
+
+// FindBrokerAddressInSubscribe indicates an expected call of FindBrokerAddressInSubscribe
+func (mr *MockNamesrvsMockRecorder) FindBrokerAddressInSubscribe(brokerName, brokerId, onlyThisBroker interface{}) *gomock.Call {
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FindBrokerAddressInSubscribe", reflect.TypeOf((*MockNamesrvs)(nil).FindBrokerAddressInSubscribe), brokerName, brokerId, onlyThisBroker)
+}
+
+// FetchSubscribeMessageQueues mocks base method
+func (m *MockNamesrvs) FetchSubscribeMessageQueues(topic string) ([]*primitive.MessageQueue, error) {
+	ret := m.ctrl.Call(m, "FetchSubscribeMessageQueues", topic)
+	ret0, _ := ret[0].([]*primitive.MessageQueue)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// FetchSubscribeMessageQueues indicates an expected call of FetchSubscribeMessageQueues
+func (mr *MockNamesrvsMockRecorder) FetchSubscribeMessageQueues(topic interface{}) *gomock.Call {
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FetchSubscribeMessageQueues", reflect.TypeOf((*MockNamesrvs)(nil).FetchSubscribeMessageQueues), topic)
+}
diff --git a/internal/namesrv.go b/internal/namesrv.go
index d8e1a40..3fc84dd 100644
--- a/internal/namesrv.go
+++ b/internal/namesrv.go
@@ -23,6 +23,7 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/apache/rocketmq-client-go/internal/remote"
 	"github.com/apache/rocketmq-client-go/primitive"
 )
 
@@ -34,8 +35,27 @@ var (
 	ErrIllegalIP    = errors.New("IP addr error")
 )
 
-// Namesrvs rocketmq namesrv instance.
-type Namesrvs struct {
+//go:generate mockgen -source namesrv.go -destination mock_namesrv.go -self_package github.com/apache/rocketmq-client-go/internal  --package internal Namesrvs
+type Namesrvs interface {
+	AddBroker(routeData *TopicRouteData)
+
+	cleanOfflineBroker()
+
+	UpdateTopicRouteInfo(topic string) *TopicRouteData
+
+	FetchPublishMessageQueues(topic string) ([]*primitive.MessageQueue, error)
+
+	FindBrokerAddrByTopic(topic string) string
+
+	FindBrokerAddrByName(brokerName string) string
+
+	FindBrokerAddressInSubscribe(brokerName string, brokerId int64, onlyThisBroker bool) *FindBrokerResult
+
+	FetchSubscribeMessageQueues(topic string) ([]*primitive.MessageQueue, error)
+}
+
+// namesrvs rocketmq namesrv instance.
+type namesrvs struct {
 	// namesrv addr list
 	srvs []string
 
@@ -45,36 +65,37 @@ type Namesrvs struct {
 	// index indicate the next position for getNamesrv
 	index int
 
-	// credentials for query topic route
-	credentials primitive.Credentials
-}
+	// brokerName -> *BrokerData
+	brokerAddressesMap sync.Map
 
-// NewNamesrv init Namesrv from namesrv addr string.
-func NewNamesrv(s ...string) (*Namesrvs, error) {
-	if len(s) == 0 {
-		return nil, ErrNoNameserver
-	}
+	// brokerName -> map[string]int32
+	brokerVersionMap sync.Map
 
-	ss := s
-	if len(ss) == 1 {
-		// compatible with multi server env string: "a;b;c"
-		ss = strings.Split(s[0], ";")
-	}
+	//subscribeInfoMap sync.Map
+	routeDataMap sync.Map
 
-	for _, srv := range ss {
-		if err := verifyIP(srv); err != nil {
-			return nil, err
-		}
-	}
+	lockNamesrv sync.Mutex
 
-	return &Namesrvs{
-		srvs: ss,
-		lock: new(sync.Mutex),
+	nameSrvClient *remote.RemotingClient
+}
+
+var _ Namesrvs = &namesrvs{}
+
+// NewNamesrv init Namesrv from namesrv addr string.
+func NewNamesrv(addr primitive.NamesrvAddr) (*namesrvs, error) {
+	if err := addr.Check(); err != nil {
+		return nil, err
+	}
+	nameSrvClient := remote.NewRemotingClient()
+	return &namesrvs{
+		srvs:          addr,
+		lock:          new(sync.Mutex),
+		nameSrvClient: nameSrvClient,
 	}, nil
 }
 
-// GetNamesrv return namesrv using round-robin strategy.
-func (s *Namesrvs) GetNamesrv() string {
+// getNamesrv return namesrv using round-robin strategy.
+func (s *namesrvs) getNamesrv() string {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -88,15 +109,15 @@ func (s *Namesrvs) GetNamesrv() string {
 	return strings.TrimLeft(addr, "http(s)://")
 }
 
-func (s *Namesrvs) Size() int {
+func (s *namesrvs) Size() int {
 	return len(s.srvs)
 }
 
-func (s *Namesrvs) String() string {
+func (s *namesrvs) String() string {
 	return strings.Join(s.srvs, ";")
 }
-func (s *Namesrvs) SetCredentials(credentials primitive.Credentials) {
-	s.credentials = credentials
+func (s *namesrvs) SetCredentials(credentials primitive.Credentials) {
+	s.nameSrvClient.RegisterInterceptor(remote.ACLInterceptor(credentials))
 }
 
 var (
diff --git a/internal/remote/remote_client.go b/internal/remote/remote_client.go
index f9b77d5..4159808 100644
--- a/internal/remote/remote_client.go
+++ b/internal/remote/remote_client.go
@@ -172,6 +172,9 @@ func (c *RemotingClient) receiveResponse(r net.Conn) {
 
 func (c *RemotingClient) createScanner(r io.Reader) *bufio.Scanner {
 	scanner := bufio.NewScanner(r)
+
+	// max batch size: 32, max message size: 4Mb
+	scanner.Buffer(make([]byte, 1024*1024), 128*1024*1024)
 	scanner.Split(func(data []byte, atEOF bool) (int, []byte, error) {
 		defer func() {
 			if err := recover(); err != nil {
diff --git a/internal/route.go b/internal/route.go
index 4f70eb0..8ed00b0 100644
--- a/internal/route.go
+++ b/internal/route.go
@@ -28,11 +28,12 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/tidwall/gjson"
+
 	"github.com/apache/rocketmq-client-go/internal/remote"
 	"github.com/apache/rocketmq-client-go/internal/utils"
 	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
-	"github.com/tidwall/gjson"
 )
 
 const (
@@ -46,39 +47,17 @@ const (
 
 var (
 	ErrTopicNotExist = errors.New("topic not exist")
-	nameSrvClient    = remote.NewRemotingClient()
-
-	nameSrvs *Namesrvs
-)
-
-func RegisterNamsrv(s *Namesrvs) {
-	if !s.credentials.IsEmpty() {
-		nameSrvClient.RegisterInterceptor(remote.ACLInterceptor(s.credentials))
-	}
-	nameSrvs = s
-}
-
-var (
-	// brokerName -> *BrokerData
-	brokerAddressesMap sync.Map
-
-	// brokerName -> map[string]int32
-	brokerVersionMap sync.Map
-
-	//subscribeInfoMap sync.Map
-	routeDataMap sync.Map
-	lockNamesrv  sync.Mutex
 )
 
-func cleanOfflineBroker() {
+func (s *namesrvs) cleanOfflineBroker() {
 	// TODO optimize
-	lockNamesrv.Lock()
-	brokerAddressesMap.Range(func(key, value interface{}) bool {
+	s.lockNamesrv.Lock()
+	s.brokerAddressesMap.Range(func(key, value interface{}) bool {
 		brokerName := key.(string)
 		bd := value.(*BrokerData)
 		for k, v := range bd.BrokerAddresses {
 			isBrokerAddrExistInTopicRoute := false
-			routeDataMap.Range(func(key, value interface{}) bool {
+			s.routeDataMap.Range(func(key, value interface{}) bool {
 				trd := value.(*TopicRouteData)
 				for idx := range trd.BrokerDataList {
 					for _, v1 := range trd.BrokerDataList[idx].BrokerAddresses {
@@ -96,12 +75,12 @@ func cleanOfflineBroker() {
 			}
 		}
 		if len(bd.BrokerAddresses) == 0 {
-			brokerAddressesMap.Delete(brokerName)
+			s.brokerAddressesMap.Delete(brokerName)
 			rlog.Infof("the broker [name=%s] name's host is offline, remove it", brokerName)
 		}
 		return true
 	})
-	lockNamesrv.Unlock()
+	s.lockNamesrv.Unlock()
 }
 
 // key is topic, value is TopicPublishInfo
@@ -126,12 +105,12 @@ func (info *TopicPublishInfo) fetchQueueIndex() int {
 	return int(qIndex) % length
 }
 
-func UpdateTopicRouteInfo(topic string) *TopicRouteData {
+func (s *namesrvs) UpdateTopicRouteInfo(topic string) *TopicRouteData {
 	// Todo process lock timeout
-	lockNamesrv.Lock()
-	defer lockNamesrv.Unlock()
+	s.lockNamesrv.Lock()
+	defer s.lockNamesrv.Unlock()
 
-	routeData, err := queryTopicRouteInfoFromServer(topic)
+	routeData, err := s.queryTopicRouteInfoFromServer(topic)
 	if err != nil {
 		rlog.Warnf("query topic route from server error: %s", err)
 		return nil
@@ -142,33 +121,32 @@ func UpdateTopicRouteInfo(topic string) *TopicRouteData {
 		return nil
 	}
 
-	oldRouteData, exist := routeDataMap.Load(topic)
+	oldRouteData, exist := s.routeDataMap.Load(topic)
 	changed := true
 	if exist {
-		changed = topicRouteDataIsChange(oldRouteData.(*TopicRouteData), routeData)
+		changed = s.topicRouteDataIsChange(oldRouteData.(*TopicRouteData), routeData)
 	}
 
 	if changed {
-		routeDataMap.Store(topic, routeData)
+		s.routeDataMap.Store(topic, routeData)
 		rlog.Infof("the topic [%s] route info changed, old %v ,new %s", topic,
 			oldRouteData, routeData.String())
 		for _, brokerData := range routeData.BrokerDataList {
-			brokerAddressesMap.Store(brokerData.BrokerName, brokerData)
+			s.brokerAddressesMap.Store(brokerData.BrokerName, brokerData)
 		}
 	}
 
 	return routeData.clone()
 }
 
-// just for test
-func AddBroker(routeData *TopicRouteData) {
+func (s *namesrvs) AddBroker(routeData *TopicRouteData) {
 	for _, brokerData := range routeData.BrokerDataList {
-		brokerAddressesMap.Store(brokerData.BrokerName, brokerData)
+		s.brokerAddressesMap.Store(brokerData.BrokerName, brokerData)
 	}
 }
 
-func FindBrokerAddrByTopic(topic string) string {
-	v, exist := routeDataMap.Load(topic)
+func (s *namesrvs) FindBrokerAddrByTopic(topic string) string {
+	v, exist := s.routeDataMap.Load(topic)
 	if !exist {
 		return ""
 	}
@@ -192,8 +170,8 @@ func FindBrokerAddrByTopic(topic string) string {
 	return addr
 }
 
-func FindBrokerAddrByName(brokerName string) string {
-	bd, exist := brokerAddressesMap.Load(brokerName)
+func (s *namesrvs) FindBrokerAddrByName(brokerName string) string {
+	bd, exist := s.brokerAddressesMap.Load(brokerName)
 
 	if !exist {
 		return ""
@@ -202,14 +180,14 @@ func FindBrokerAddrByName(brokerName string) string {
 	return bd.(*BrokerData).BrokerAddresses[MasterId]
 }
 
-func FindBrokerAddressInSubscribe(brokerName string, brokerId int64, onlyThisBroker bool) *FindBrokerResult {
+func (s *namesrvs) FindBrokerAddressInSubscribe(brokerName string, brokerId int64, onlyThisBroker bool) *FindBrokerResult {
 	var (
 		brokerAddr = ""
 		//slave      = false
 		//found      = false
 	)
 
-	v, exist := brokerAddressesMap.Load(brokerName)
+	v, exist := s.brokerAddressesMap.Load(brokerName)
 
 	if !exist {
 		return nil
@@ -236,15 +214,15 @@ func FindBrokerAddressInSubscribe(brokerName string, brokerId int64, onlyThisBro
 		result = &FindBrokerResult{
 			BrokerAddr:    brokerAddr,
 			Slave:         brokerId != 0,
-			BrokerVersion: findBrokerVersion(brokerName, brokerAddr),
+			BrokerVersion: s.findBrokerVersion(brokerName, brokerAddr),
 		}
 	}
 
 	return result
 }
 
-func FetchSubscribeMessageQueues(topic string) ([]*primitive.MessageQueue, error) {
-	routeData, err := queryTopicRouteInfoFromServer(topic)
+func (s *namesrvs) FetchSubscribeMessageQueues(topic string) ([]*primitive.MessageQueue, error) {
+	routeData, err := s.queryTopicRouteInfoFromServer(topic)
 
 	if err != nil {
 		return nil, err
@@ -262,8 +240,8 @@ func FetchSubscribeMessageQueues(topic string) ([]*primitive.MessageQueue, error
 	return mqs, nil
 }
 
-func FindMQByTopic(topic string) *primitive.MessageQueue {
-	mqs, err := FetchPublishMessageQueues(topic)
+func (s *namesrvs) FindMQByTopic(topic string) *primitive.MessageQueue {
+	mqs, err := s.FetchPublishMessageQueues(topic)
 	if err != nil {
 		return nil
 	}
@@ -272,21 +250,21 @@ func FindMQByTopic(topic string) *primitive.MessageQueue {
 	return mqs[i%len(mqs)]
 }
 
-func FetchPublishMessageQueues(topic string) ([]*primitive.MessageQueue, error) {
+func (s *namesrvs) FetchPublishMessageQueues(topic string) ([]*primitive.MessageQueue, error) {
 	var (
 		err       error
 		routeData *TopicRouteData
 	)
 
-	v, exist := routeDataMap.Load(topic)
+	v, exist := s.routeDataMap.Load(topic)
 	if !exist {
-		routeData, err = queryTopicRouteInfoFromServer(topic)
+		routeData, err = s.queryTopicRouteInfoFromServer(topic)
 		if err != nil {
 			rlog.Error("queryTopicRouteInfoFromServer failed. topic: ", topic)
 			return nil, err
 		}
-		routeDataMap.Store(topic, routeData)
-		AddBroker(routeData)
+		s.routeDataMap.Store(topic, routeData)
+		s.AddBroker(routeData)
 	} else {
 		routeData = v.(*TopicRouteData)
 	}
@@ -294,14 +272,14 @@ func FetchPublishMessageQueues(topic string) ([]*primitive.MessageQueue, error)
 	if err != nil {
 		return nil, err
 	}
-	publishinfo := routeData2PublishInfo(topic, routeData)
+	publishinfo := s.routeData2PublishInfo(topic, routeData)
 
 	return publishinfo.MqList, nil
 }
 
-func findBrokerVersion(brokerName, brokerAddr string) int32 {
+func (s *namesrvs) findBrokerVersion(brokerName, brokerAddr string) int32 {
 
-	versions, exist := brokerVersionMap.Load(brokerName)
+	versions, exist := s.brokerVersionMap.Load(brokerName)
 
 	if !exist {
 		return 0
@@ -315,7 +293,7 @@ func findBrokerVersion(brokerName, brokerAddr string) int32 {
 	return 0
 }
 
-func queryTopicRouteInfoFromServer(topic string) (*TopicRouteData, error) {
+func (s *namesrvs) queryTopicRouteInfoFromServer(topic string) (*TopicRouteData, error) {
 	request := &GetRouteInfoRequest{
 		Topic: topic,
 	}
@@ -324,16 +302,16 @@ func queryTopicRouteInfoFromServer(topic string) (*TopicRouteData, error) {
 		response *remote.RemotingCommand
 		err      error
 	)
-	for i := 0; i < nameSrvs.Size(); i++ {
+	for i := 0; i < s.Size(); i++ {
 		rc := remote.NewRemotingCommand(ReqGetRouteInfoByTopic, request, nil)
-		response, err = nameSrvClient.InvokeSync(getNameServerAddress(), rc, requestTimeout)
+		response, err = s.nameSrvClient.InvokeSync(s.getNameServerAddress(), rc, requestTimeout)
 
 		if err != nil {
 			continue
 		}
 	}
 	if err != nil {
-		rlog.Errorf("connect to namesrv: %v failed.", nameSrvs)
+		rlog.Errorf("connect to namesrv: %v failed.", s)
 		return nil, err
 	}
 
@@ -357,7 +335,7 @@ func queryTopicRouteInfoFromServer(topic string) (*TopicRouteData, error) {
 	}
 }
 
-func topicRouteDataIsChange(oldData *TopicRouteData, newData *TopicRouteData) bool {
+func (s *namesrvs) topicRouteDataIsChange(oldData *TopicRouteData, newData *TopicRouteData) bool {
 	if oldData == nil || newData == nil {
 		return true
 	}
@@ -380,7 +358,7 @@ func topicRouteDataIsChange(oldData *TopicRouteData, newData *TopicRouteData) bo
 	return !oldDataCloned.equals(newDataCloned)
 }
 
-func routeData2PublishInfo(topic string, data *TopicRouteData) *TopicPublishInfo {
+func (s *namesrvs) routeData2PublishInfo(topic string, data *TopicRouteData) *TopicPublishInfo {
 	publishInfo := &TopicPublishInfo{
 		RouteData:  data,
 		OrderTopic: false,
@@ -440,8 +418,8 @@ func routeData2PublishInfo(topic string, data *TopicRouteData) *TopicPublishInfo
 	return publishInfo
 }
 
-func getNameServerAddress() string {
-	return nameSrvs.GetNamesrv()
+func (s *namesrvs) getNameServerAddress() string {
+	return s.getNamesrv()
 }
 
 // TopicRouteData TopicRouteData
diff --git a/internal/trace.go b/internal/trace.go
index ebd3be5..ca60eea 100644
--- a/internal/trace.go
+++ b/internal/trace.go
@@ -27,6 +27,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"github.com/apache/rocketmq-client-go/internal/remote"
 	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
@@ -203,13 +205,13 @@ type traceDispatcher struct {
 	discardCount int64
 
 	// support deliver trace message to other cluster.
-	namesrvs []string
+	namesrvs *namesrvs
 	// round robin index
 	rrindex int32
 	cli     RMQClient
 }
 
-func NewTraceDispatcher(traceTopic string, access primitive.AccessChannel) *traceDispatcher {
+func NewTraceDispatcher(traceTopic string, access primitive.AccessChannel, nameServerAddrs []string) *traceDispatcher {
 	ctx := context.Background()
 	ctx, cancel := context.WithCancel(ctx)
 
@@ -222,8 +224,14 @@ func NewTraceDispatcher(traceTopic string, access primitive.AccessChannel) *trac
 		t = TraceTopicPrefix + traceTopic
 	}
 
+	srvs, err := NewNamesrv(nameServerAddrs)
+	if err != nil {
+		panic(errors.Wrap(err, "new Namesrv failed."))
+	}
+
 	cliOp := DefaultClientOptions()
 	cliOp.RetryTimes = 0
+	cliOp.Namesrv = srvs
 	cli := GetOrNewRocketMQClient(cliOp, nil)
 	return &traceDispatcher{
 		ctx:    ctx,
@@ -234,6 +242,7 @@ func NewTraceDispatcher(traceTopic string, access primitive.AccessChannel) *trac
 		input:      make(chan TraceContext, 1024),
 		batchCh:    make(chan []*TraceContext, 2048),
 		cli:        cli,
+		namesrvs:   srvs,
 	}
 }
 
@@ -396,7 +405,7 @@ func (td *traceDispatcher) sendTraceDataByMQ(keyset Keyset, regionID string, dat
 }
 
 func (td *traceDispatcher) findMq() (*primitive.MessageQueue, string) {
-	mqs, err := FetchPublishMessageQueues(td.traceTopic)
+	mqs, err := td.namesrvs.FetchPublishMessageQueues(td.traceTopic)
 	if err != nil {
 		rlog.Error("fetch publish message queues failed. err: %v", err)
 		return nil, ""
@@ -410,7 +419,7 @@ func (td *traceDispatcher) findMq() (*primitive.MessageQueue, string) {
 	mq := mqs[i]
 
 	brokerName := mq.BrokerName
-	addr := FindBrokerAddrByName(brokerName)
+	addr := td.namesrvs.FindBrokerAddrByName(brokerName)
 
 	return mq, addr
 }
diff --git a/internal/utils/helper.go b/internal/utils/helper.go
index 58ea791..7215d92 100644
--- a/internal/utils/helper.go
+++ b/internal/utils/helper.go
@@ -17,10 +17,25 @@ limitations under the License.
 
 package utils
 
+import (
+	"bytes"
+	"compress/zlib"
+	"io/ioutil"
+)
+
 func GetAddressByBytes(data []byte) string {
 	return "127.0.0.1"
 }
 
 func UnCompress(data []byte) []byte {
-	return data
+	rdata := bytes.NewReader(data)
+	r, err := zlib.NewReader(rdata)
+	if err != nil {
+		return data
+	}
+	retData, err := ioutil.ReadAll(r)
+	if err != nil {
+		return data
+	}
+	return retData
 }
diff --git a/primitive/base.go b/primitive/base.go
new file mode 100644
index 0000000..297e69b
--- /dev/null
+++ b/primitive/base.go
@@ -0,0 +1,75 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 primitive
+
+import (
+	"regexp"
+	"strings"
+)
+
+var (
+	ipRegex, _ = regexp.Compile(`^((25[0-5]|2[0-4]\d|((1\d{2})|([1-9]?\d)))\.){3}(25[0-5]|2[0-4]\d|((1\d{2})|([1-9]?\d)))`)
+)
+
+type NamesrvAddr []string
+
+func NewNamesrvAddr(s ...string) (NamesrvAddr, error) {
+	if len(s) == 0 {
+		return nil, ErrNoNameserver
+	}
+
+	ss := s
+	if len(ss) == 1 {
+		// compatible with multi server env string: "a;b;c"
+		ss = strings.Split(s[0], ";")
+	}
+
+	for _, srv := range ss {
+		if err := verifyIP(srv); err != nil {
+			return nil, err
+		}
+	}
+
+	addrs := make(NamesrvAddr, 0)
+	addrs = append(addrs, ss...)
+	return addrs, nil
+}
+
+func (addr NamesrvAddr) Check() error {
+	for _, srv := range addr {
+		if err := verifyIP(srv); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func verifyIP(ip string) error {
+	if strings.Contains(ip, ";") {
+		return ErrMultiIP
+	}
+	ips := ipRegex.FindAllString(ip, -1)
+	if len(ips) == 0 {
+		return ErrIllegalIP
+	}
+
+	if len(ips) > 1 {
+		return ErrMultiIP
+	}
+	return nil
+}
diff --git a/primitive/errors.go b/primitive/errors.go
new file mode 100644
index 0000000..af18999
--- /dev/null
+++ b/primitive/errors.go
@@ -0,0 +1,28 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 primitive
+
+import (
+	"errors"
+)
+
+var (
+	ErrNoNameserver = errors.New("nameServerAddrs can't be empty.")
+	ErrMultiIP      = errors.New("multiple IP addr does not support")
+	ErrIllegalIP    = errors.New("IP addr error")
+)
diff --git a/primitive/interceptor.go b/primitive/interceptor.go
index 6e82df8..92590a3 100644
--- a/primitive/interceptor.go
+++ b/primitive/interceptor.go
@@ -31,8 +31,9 @@ type Interceptor func(ctx context.Context, req, reply interface{}, next Invoker)
 
 // config for message trace.
 type TraceConfig struct {
-	TraceTopic string
-	Access     AccessChannel
+	TraceTopic   string
+	Access       AccessChannel
+	NamesrvAddrs []string
 }
 
 func ChainInterceptors(interceptors ...Interceptor) Interceptor {
diff --git a/producer/interceptor.go b/producer/interceptor.go
index d545f93..24cf80e 100644
--- a/producer/interceptor.go
+++ b/producer/interceptor.go
@@ -41,7 +41,7 @@ func WithTrace(traceCfg primitive.TraceConfig) Option {
 }
 
 func newTraceInterceptor(traceCfg primitive.TraceConfig) primitive.Interceptor {
-	dispatcher := internal.NewTraceDispatcher(traceCfg.TraceTopic, traceCfg.Access)
+	dispatcher := internal.NewTraceDispatcher(traceCfg.TraceTopic, traceCfg.Access, traceCfg.NamesrvAddrs)
 	dispatcher.Start()
 
 	return func(ctx context.Context, req, reply interface{}, next primitive.Invoker) error {
diff --git a/producer/option.go b/producer/option.go
index ac88120..c352b6e 100644
--- a/producer/option.go
+++ b/producer/option.go
@@ -53,11 +53,9 @@ func WithGroupName(group string) Option {
 }
 
 // WithNameServer set NameServer address, only support one NameServer cluster in alpha2
-func WithNameServer(nameServers []string) Option {
+func WithNameServer(nameServers primitive.NamesrvAddr) Option {
 	return func(opts *producerOptions) {
-		if len(nameServers) > 0 {
-			opts.NameServerAddrs = nameServers
-		}
+		opts.NameServerAddrs = nameServers
 	}
 }
 
diff --git a/producer/producer.go b/producer/producer.go
index 12ac962..0b0367f 100644
--- a/producer/producer.go
+++ b/producer/producer.go
@@ -55,14 +55,14 @@ func NewDefaultProducer(opts ...Option) (*defaultProducer, error) {
 	for _, apply := range opts {
 		apply(&defaultOpts)
 	}
-	srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs...)
+	srvs, err := internal.NewNamesrv(defaultOpts.NameServerAddrs)
 	if err != nil {
 		return nil, errors.Wrap(err, "new Namesrv failed.")
 	}
 	if !defaultOpts.Credentials.IsEmpty() {
 		srvs.SetCredentials(defaultOpts.Credentials)
 	}
-	internal.RegisterNamsrv(srvs)
+	defaultOpts.Namesrv = srvs
 
 	producer := &defaultProducer{
 		group:      defaultOpts.GroupName,
@@ -155,7 +155,7 @@ func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message,
 			continue
 		}
 
-		addr := internal.FindBrokerAddrByName(mq.BrokerName)
+		addr := p.options.Namesrv.FindBrokerAddrByName(mq.BrokerName)
 		if addr == "" {
 			return fmt.Errorf("topic=%s route info not found", mq.Topic)
 		}
@@ -200,7 +200,7 @@ func (p *defaultProducer) sendAsync(ctx context.Context, msg *primitive.Message,
 		return errors.Errorf("the topic=%s route info not found", msg.Topic)
 	}
 
-	addr := internal.FindBrokerAddrByName(mq.BrokerName)
+	addr := p.options.Namesrv.FindBrokerAddrByName(mq.BrokerName)
 	if addr == "" {
 		return errors.Errorf("topic=%s route info not found", mq.Topic)
 	}
@@ -246,7 +246,7 @@ func (p *defaultProducer) sendOneWay(ctx context.Context, msg *primitive.Message
 			continue
 		}
 
-		addr := internal.FindBrokerAddrByName(mq.BrokerName)
+		addr := p.options.Namesrv.FindBrokerAddrByName(mq.BrokerName)
 		if addr == "" {
 			return fmt.Errorf("topic=%s route info not found", mq.Topic)
 		}
@@ -299,7 +299,7 @@ func (p *defaultProducer) selectMessageQueue(msg *primitive.Message) *primitive.
 
 	v, exist := p.publishInfo.Load(topic)
 	if !exist {
-		p.client.UpdatePublishInfo(topic, internal.UpdateTopicRouteInfo(topic))
+		p.client.UpdatePublishInfo(topic, p.options.Namesrv.UpdateTopicRouteInfo(topic))
 		v, exist = p.publishInfo.Load(topic)
 	}
 
@@ -454,7 +454,7 @@ func (tp *transactionProducer) endTransaction(result primitive.SendResult, err e
 		msgID, _ = primitive.UnmarshalMsgID([]byte(result.MsgID))
 	}
 	// 估计没有反序列化回来
-	brokerAddr := internal.FindBrokerAddrByName(result.MessageQueue.BrokerName)
+	brokerAddr := tp.producer.options.Namesrv.FindBrokerAddrByName(result.MessageQueue.BrokerName)
 	requestHeader := &internal.EndTransactionRequestHeader{
 		TransactionId:        result.TransactionID,
 		CommitLogOffset:      msgID.Offset,


Test patch:
diff --git a/consumer/offset_store_test.go b/consumer/offset_store_test.go
index 6bfc90c..93fbb59 100644
--- a/consumer/offset_store_test.go
+++ b/consumer/offset_store_test.go
@@ -1,13 +1,14 @@
 package consumer
 
 import (
-	"github.com/agiledragon/gomonkey"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	. "github.com/smartystreets/goconvey/convey"
+
 	"github.com/apache/rocketmq-client-go/internal"
 	"github.com/apache/rocketmq-client-go/internal/remote"
 	"github.com/apache/rocketmq-client-go/primitive"
-	"github.com/golang/mock/gomock"
-	. "github.com/smartystreets/goconvey/convey"
-	"testing"
 )
 
 func TestNewLocalFileOffsetStore(t *testing.T) {
@@ -192,8 +193,10 @@ func TestRemoteBrokerOffsetStore(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		defer ctrl.Finish()
 
+		namesrv := internal.NewMockNamesrvs(ctrl)
+
 		rmqClient := internal.NewMockRMQClient(ctrl)
-		remoteStore := NewRemoteOffsetStore("testGroup", rmqClient)
+		remoteStore := NewRemoteOffsetStore("testGroup", rmqClient, namesrv)
 
 		type offsetCase struct {
 			queue          *primitive.MessageQueue
@@ -251,10 +254,7 @@ func TestRemoteBrokerOffsetStore(t *testing.T) {
 		Convey("test persist", func() {
 			queues := []*primitive.MessageQueue{mq}
 
-			patch := gomonkey.ApplyFunc(internal.FindBrokerAddrByName, func(_ string) string {
-				return "192.168.24.1:10911"
-			})
-			defer patch.Reset()
+			namesrv.EXPECT().FindBrokerAddrByName(gomock.Any()).Return("192.168.24.1:10911")
 
 			ret := &remote.RemotingCommand{
 				Code: internal.ResSuccess,
diff --git a/internal/namesrv_test.go b/internal/namesrv_test.go
index 145acff..30fceb9 100644
--- a/internal/namesrv_test.go
+++ b/internal/namesrv_test.go
@@ -25,48 +25,26 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestVerifyIP(t *testing.T) {
-	IPs := "127.0.0.1:9876"
-	err := verifyIP(IPs)
-	assert.Nil(t, err)
-
-	IPs = "12.24.123.243:10911"
-	err = verifyIP(IPs)
-	assert.Nil(t, err)
-
-	IPs = "xa2.0.0.1:9876"
-	err = verifyIP(IPs)
-	assert.Equal(t, "IP addr error", err.Error())
-
-	IPs = "333.0.0.1:9876"
-	err = verifyIP(IPs)
-	assert.Equal(t, "IP addr error", err.Error())
-
-	IPs = "127.0.0.1:9876;12.24.123.243:10911"
-	err = verifyIP(IPs)
-	assert.Equal(t, "multiple IP addr does not support", err.Error())
-}
-
 // TestSelector test roundrobin selector in namesrv
 func TestSelector(t *testing.T) {
 	srvs := []string{"127.0.0.1:9876", "127.0.0.1:9879", "12.24.123.243:10911", "12.24.123.243:10915"}
-	namesrv, err := NewNamesrv(srvs...)
+	namesrv, err := NewNamesrv(srvs)
 	assert.Nil(t, err)
 
-	assert.Equal(t, srvs[0], namesrv.GetNamesrv())
-	assert.Equal(t, srvs[1], namesrv.GetNamesrv())
-	assert.Equal(t, srvs[2], namesrv.GetNamesrv())
-	assert.Equal(t, srvs[3], namesrv.GetNamesrv())
-	assert.Equal(t, srvs[0], namesrv.GetNamesrv())
-	assert.Equal(t, srvs[1], namesrv.GetNamesrv())
-	assert.Equal(t, srvs[2], namesrv.GetNamesrv())
-	assert.Equal(t, srvs[3], namesrv.GetNamesrv())
-	assert.Equal(t, srvs[0], namesrv.GetNamesrv())
+	assert.Equal(t, srvs[0], namesrv.getNamesrv())
+	assert.Equal(t, srvs[1], namesrv.getNamesrv())
+	assert.Equal(t, srvs[2], namesrv.getNamesrv())
+	assert.Equal(t, srvs[3], namesrv.getNamesrv())
+	assert.Equal(t, srvs[0], namesrv.getNamesrv())
+	assert.Equal(t, srvs[1], namesrv.getNamesrv())
+	assert.Equal(t, srvs[2], namesrv.getNamesrv())
+	assert.Equal(t, srvs[3], namesrv.getNamesrv())
+	assert.Equal(t, srvs[0], namesrv.getNamesrv())
 }
 
 func TestGetNamesrv(t *testing.T) {
 	Convey("Test GetNamesrv round-robin strategy", t, func() {
-		ns := &Namesrvs{
+		ns := &namesrvs{
 			srvs: []string{"192.168.100.1",
 				"192.168.100.2",
 				"192.168.100.3",
@@ -77,10 +55,10 @@ func TestGetNamesrv(t *testing.T) {
 		}
 
 		index1 := ns.index
-		IP1 := ns.GetNamesrv()
+		IP1 := ns.getNamesrv()
 
 		index2 := ns.index
-		IP2 := ns.GetNamesrv()
+		IP2 := ns.getNamesrv()
 
 		So(index1+1, ShouldEqual, index2)
 		So(IP1, ShouldEqual, ns.srvs[index1])
diff --git a/internal/utils/helper_test.go b/internal/utils/helper_test.go
new file mode 100644
index 0000000..837c5ab
--- /dev/null
+++ b/internal/utils/helper_test.go
@@ -0,0 +1,37 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 (
+	"bytes"
+	"compress/zlib"
+	"testing"
+)
+
+func TestUnCompress(t *testing.T) {
+	var b bytes.Buffer
+	var oriStr string = "hello, go"
+	zr := zlib.NewWriter(&b)
+	zr.Write([]byte(oriStr))
+	zr.Close()
+
+	retBytes := UnCompress(b.Bytes())
+	if string(retBytes) != oriStr {
+		t.Errorf("UnCompress was incorrect, got %s, want: %s .", retBytes, []byte(oriStr))
+	}
+}
diff --git a/primitive/base_test.go b/primitive/base_test.go
new file mode 100644
index 0000000..09d55c3
--- /dev/null
+++ b/primitive/base_test.go
@@ -0,0 +1,46 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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 primitive
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestVerifyIP(t *testing.T) {
+	IPs := "127.0.0.1:9876"
+	err := verifyIP(IPs)
+	assert.Nil(t, err)
+
+	IPs = "12.24.123.243:10911"
+	err = verifyIP(IPs)
+	assert.Nil(t, err)
+
+	IPs = "xa2.0.0.1:9876"
+	err = verifyIP(IPs)
+	assert.Equal(t, "IP addr error", err.Error())
+
+	IPs = "333.0.0.1:9876"
+	err = verifyIP(IPs)
+	assert.Equal(t, "IP addr error", err.Error())
+
+	IPs = "127.0.0.1:9876;12.24.123.243:10911"
+	err = verifyIP(IPs)
+	assert.Equal(t, "multiple IP addr does not support", err.Error())
+}
diff --git a/producer/producer_test.go b/producer/producer_test.go
index c1c8046..849978a 100644
--- a/producer/producer_test.go
+++ b/producer/producer_test.go
@@ -22,11 +22,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
 	"github.com/apache/rocketmq-client-go/internal"
 	"github.com/apache/rocketmq-client-go/internal/remote"
 	"github.com/apache/rocketmq-client-go/primitive"
-	"github.com/golang/mock/gomock"
-	"github.com/stretchr/testify/assert"
 )
 
 const (
@@ -82,7 +83,7 @@ func mockB4Send(p *defaultProducer) {
 			},
 		},
 	})
-	internal.AddBroker(&internal.TopicRouteData{
+	p.options.Namesrv.AddBroker(&internal.TopicRouteData{
 		BrokerDataList: []*internal.BrokerData{
 			{
 				Cluster:    "cluster",