diff --git a/consumer/consumer.go b/consumer/consumer.go
index 13cef4e..070cd19 100644
--- a/consumer/consumer.go
+++ b/consumer/consumer.go
@@ -236,12 +236,13 @@
 	 *
 	 * See <a href="http://rocketmq.apache.org/docs/core-concept/">here</a> for further discussion.
 	 */
-	consumerGroup  string
-	model          MessageModel
-	allocate       func(string, string, []*primitive.MessageQueue, []string) []*primitive.MessageQueue
-	unitMode       bool
-	consumeOrderly bool
-	fromWhere      ConsumeFromWhere
+	consumerGroup          string
+	model                  MessageModel
+	allocate               func(string, string, []*primitive.MessageQueue, []string) []*primitive.MessageQueue
+	unitMode               bool
+	consumeOrderly         bool
+	fromWhere              ConsumeFromWhere
+	consumerStartTimestamp int64
 
 	cType     ConsumeType
 	client    internal.RMQClient
@@ -250,7 +251,7 @@
 	pause     bool
 	once      sync.Once
 	option    consumerOptions
-	// key: int, hash(*primitive.MessageQueue)
+	// key: primitive.MessageQueue
 	// value: *processQueue
 	processQueueTable sync.Map
 
@@ -287,7 +288,7 @@
 	dc.client.UpdateTopicRouteInfo()
 	dc.client.Start()
 	dc.state = internal.StateRunning
-
+	dc.consumerStartTimestamp = time.Now().UnixNano() / int64(time.Millisecond)
 	return nil
 }
 
diff --git a/consumer/process_queue.go b/consumer/process_queue.go
index 13b2d25..4f4dc0c 100644
--- a/consumer/process_queue.go
+++ b/consumer/process_queue.go
@@ -28,6 +28,7 @@
 	gods_util "github.com/emirpasic/gods/utils"
 	uatomic "go.uber.org/atomic"
 
+	"github.com/apache/rocketmq-client-go/internal"
 	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
 )
@@ -277,6 +278,28 @@
 	return -1
 }
 
+func (pq *processQueue) MinOrderlyCache() int64 {
+	if pq.consumingMsgOrderlyTreeMap.Empty() {
+		return -1
+	}
+	k, _ := pq.consumingMsgOrderlyTreeMap.Min()
+	if k != nil {
+		return k.(int64)
+	}
+	return -1
+}
+
+func (pq *processQueue) MaxOrderlyCache() int64 {
+	if pq.consumingMsgOrderlyTreeMap.Empty() {
+		return -1
+	}
+	k, _ := pq.consumingMsgOrderlyTreeMap.Max()
+	if k != nil {
+		return k.(int64)
+	}
+	return -1
+}
+
 func (pq *processQueue) clear() {
 	pq.mutex.Lock()
 	pq.msgCache.Clear()
@@ -302,3 +325,31 @@
 	pq.consumingMsgOrderlyTreeMap.Clear()
 	return offset + 1
 }
+
+func (pq *processQueue) currentInfo() internal.ProcessQueueInfo {
+	pq.mutex.RLock()
+	defer pq.mutex.RUnlock()
+	info := internal.ProcessQueueInfo{
+		Locked:               pq.locked.Load(),
+		TryUnlockTimes:       pq.tryUnlockTimes,
+		LastLockTimestamp:    pq.lastLockTime.UnixNano() / 10e6,
+		Dropped:              pq.dropped.Load(),
+		LastPullTimestamp:    pq.lastPullTime.UnixNano() / 10e6,
+		LastConsumeTimestamp: pq.lastConsumeTime.UnixNano() / 10e6,
+	}
+
+	if !pq.msgCache.Empty() {
+		info.CachedMsgMinOffset = pq.Min()
+		info.CachedMsgMaxOffset = pq.Max()
+		info.CachedMsgCount = pq.msgCache.Size()
+		info.CachedMsgSizeInMiB = pq.cachedMsgSize / int64(1024*1024)
+	}
+
+	if !pq.consumingMsgOrderlyTreeMap.Empty() {
+		info.TransactionMsgMinOffset = pq.MinOrderlyCache()
+		info.TransactionMsgMaxOffset = pq.MaxOrderlyCache()
+		info.TransactionMsgCount = pq.consumingMsgOrderlyTreeMap.Size()
+	}
+
+	return info
+}
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index 3c5b79a..98b0033 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -273,6 +273,45 @@
 	return pc.unitMode
 }
 
+func (pc *pushConsumer) GetConsumerRunningInfo() *internal.ConsumerRunningInfo {
+	info := internal.NewConsumerRunningInfo()
+
+	pc.subscriptionDataTable.Range(func(key, value interface{}) bool {
+		topic := key.(string)
+		info.SubscriptionData[value.(*internal.SubscriptionData)] = true
+		status := internal.ConsumeStatus{
+			PullRT:            getPullRT(topic, pc.consumerGroup).avgpt,
+			PullTPS:           getPullTPS(topic, pc.consumerGroup).tps,
+			ConsumeRT:         getConsumeRT(topic, pc.consumerGroup).avgpt,
+			ConsumeOKTPS:      getConsumeOKTPS(topic, pc.consumerGroup).tps,
+			ConsumeFailedTPS:  getConsumeFailedTPS(topic, pc.consumerGroup).tps,
+			ConsumeFailedMsgs: topicAndGroupConsumeFailedTPS.getStatsDataInHour(topic + "@" + pc.consumerGroup).sum,
+		}
+		info.StatusTable[topic] = status
+		return true
+	})
+
+	pc.processQueueTable.Range(func(key, value interface{}) bool {
+		mq := key.(primitive.MessageQueue)
+		pq := value.(*processQueue)
+		pInfo := pq.currentInfo()
+		pInfo.CommitOffset = pc.storage.read(&mq, _ReadMemoryThenStore)
+		info.MQTable[mq] = pInfo
+		return true
+	})
+
+	nsAddr := ""
+	for _, value := range pc.namesrv.AddrList() {
+		nsAddr += fmt.Sprintf("%s;", value)
+	}
+	info.Properties[internal.PropNameServerAddr] = nsAddr
+	info.Properties[internal.PropConsumeType] = string(pc.cType)
+	info.Properties[internal.PropConsumeOrderly] = strconv.FormatBool(pc.consumeOrderly)
+	info.Properties[internal.PropThreadPoolCoreSize] = "-1"
+	info.Properties[internal.PropConsumerStartTimestamp] = strconv.FormatInt(pc.consumerStartTimestamp, 10)
+	return info
+}
+
 func (pc *pushConsumer) messageQueueChanged(topic string, mqAll, mqDivided []*primitive.MessageQueue) {
 	v, exit := pc.subscriptionDataTable.Load(topic)
 	if !exit {
diff --git a/internal/client.go b/internal/client.go
index 3336944..d9c5f6a 100644
--- a/internal/client.go
+++ b/internal/client.go
@@ -18,7 +18,6 @@
 package internal
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -36,6 +35,7 @@
 )
 
 const (
+	clientVersion        = "v2.0.0-alpha3"
 	defaultTraceRegionID = "DefaultRegion"
 
 	// tracing message switch
@@ -83,6 +83,7 @@
 	SubscriptionDataList() []*SubscriptionData
 	Rebalance()
 	IsUnitMode() bool
+	GetConsumerRunningInfo() *ConsumerRunningInfo
 }
 
 func DefaultClientOptions() ClientOptions {
@@ -220,8 +221,30 @@
 
 		client.remoteClient.RegisterRequestFunc(ReqGetConsumerRunningInfo, func(req *remote.RemotingCommand, addr net.Addr) *remote.RemotingCommand {
 			rlog.Info("receive get consumer running info request...", nil)
+			header := new(GetConsumerRunningInfoHeader)
+			header.Decode(req.ExtFields)
+			val, exist := clientMap.Load(header.clientID)
 			res := remote.NewRemotingCommand(ResError, nil, nil)
-			res.Remark = "the go client has not supported consumer running info"
+			if !exist {
+				res.Remark = fmt.Sprintf("Can't find specified client instance of: %s", header.clientID)
+			} else {
+				cli, ok := val.(*rmqClient)
+				var runningInfo *ConsumerRunningInfo
+				if ok {
+					runningInfo = cli.getConsumerRunningInfo(header.consumerGroup)
+				}
+				if runningInfo != nil {
+					res.Code = ResSuccess
+					data, err := runningInfo.Encode()
+					if err != nil {
+						res.Remark = fmt.Sprintf("json marshal error: %s", err.Error())
+					} else {
+						res.Body = data
+					}
+				} else {
+					res.Remark = "there is unexpected error when get running info, please check log"
+				}
+			}
 			return res
 		})
 	}
@@ -659,6 +682,18 @@
 	return result
 }
 
+func (c *rmqClient) getConsumerRunningInfo(group string) *ConsumerRunningInfo {
+	consumer, exist := c.consumerMap.Load(group)
+	if !exist {
+		return nil
+	}
+	info := consumer.(InnerConsumer).GetConsumerRunningInfo()
+	if info != nil {
+		info.Properties[PropClientVersion] = clientVersion
+	}
+	return info
+}
+
 func routeData2SubscribeInfo(topic string, data *TopicRouteData) []*primitive.MessageQueue {
 	list := make([]*primitive.MessageQueue, 0)
 	for idx := range data.QueueDataList {
@@ -676,16 +711,6 @@
 	return list
 }
 
-func encodeMessages(message []*primitive.Message) []byte {
-	var buffer bytes.Buffer
-	index := 0
-	for index < len(message) {
-		buffer.Write(message[index].Body)
-		index++
-	}
-	return buffer.Bytes()
-}
-
 func brokerVIPChannel(brokerAddr string) string {
 	if !_VIPChannelEnable {
 		return brokerAddr
diff --git a/internal/mock_namesrv.go b/internal/mock_namesrv.go
index 630d4cc..19a2181 100644
--- a/internal/mock_namesrv.go
+++ b/internal/mock_namesrv.go
@@ -52,26 +52,31 @@
 
 // AddBroker mocks base method
 func (m *MockNamesrvs) AddBroker(routeData *TopicRouteData) {
+	m.ctrl.T.Helper()
 	m.ctrl.Call(m, "AddBroker", routeData)
 }
 
 // AddBroker indicates an expected call of AddBroker
 func (mr *MockNamesrvsMockRecorder) AddBroker(routeData interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddBroker", reflect.TypeOf((*MockNamesrvs)(nil).AddBroker), routeData)
 }
 
 // cleanOfflineBroker mocks base method
 func (m *MockNamesrvs) cleanOfflineBroker() {
+	m.ctrl.T.Helper()
 	m.ctrl.Call(m, "cleanOfflineBroker")
 }
 
 // cleanOfflineBroker indicates an expected call of cleanOfflineBroker
 func (mr *MockNamesrvsMockRecorder) cleanOfflineBroker() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "cleanOfflineBroker", reflect.TypeOf((*MockNamesrvs)(nil).cleanOfflineBroker))
 }
 
 // UpdateTopicRouteInfo mocks base method
 func (m *MockNamesrvs) UpdateTopicRouteInfo(topic string) *TopicRouteData {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "UpdateTopicRouteInfo", topic)
 	ret0, _ := ret[0].(*TopicRouteData)
 	return ret0
@@ -79,11 +84,13 @@
 
 // UpdateTopicRouteInfo indicates an expected call of UpdateTopicRouteInfo
 func (mr *MockNamesrvsMockRecorder) UpdateTopicRouteInfo(topic interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	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) {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "FetchPublishMessageQueues", topic)
 	ret0, _ := ret[0].([]*primitive.MessageQueue)
 	ret1, _ := ret[1].(error)
@@ -92,11 +99,13 @@
 
 // FetchPublishMessageQueues indicates an expected call of FetchPublishMessageQueues
 func (mr *MockNamesrvsMockRecorder) FetchPublishMessageQueues(topic interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	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 {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "FindBrokerAddrByTopic", topic)
 	ret0, _ := ret[0].(string)
 	return ret0
@@ -104,11 +113,13 @@
 
 // FindBrokerAddrByTopic indicates an expected call of FindBrokerAddrByTopic
 func (mr *MockNamesrvsMockRecorder) FindBrokerAddrByTopic(topic interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	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 {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "FindBrokerAddrByName", brokerName)
 	ret0, _ := ret[0].(string)
 	return ret0
@@ -116,11 +127,13 @@
 
 // FindBrokerAddrByName indicates an expected call of FindBrokerAddrByName
 func (mr *MockNamesrvsMockRecorder) FindBrokerAddrByName(brokerName interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	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 {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "FindBrokerAddressInSubscribe", brokerName, brokerId, onlyThisBroker)
 	ret0, _ := ret[0].(*FindBrokerResult)
 	return ret0
@@ -128,11 +141,13 @@
 
 // FindBrokerAddressInSubscribe indicates an expected call of FindBrokerAddressInSubscribe
 func (mr *MockNamesrvsMockRecorder) FindBrokerAddressInSubscribe(brokerName, brokerId, onlyThisBroker interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	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) {
+	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "FetchSubscribeMessageQueues", topic)
 	ret0, _ := ret[0].([]*primitive.MessageQueue)
 	ret1, _ := ret[1].(error)
@@ -141,5 +156,20 @@
 
 // FetchSubscribeMessageQueues indicates an expected call of FetchSubscribeMessageQueues
 func (mr *MockNamesrvsMockRecorder) FetchSubscribeMessageQueues(topic interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FetchSubscribeMessageQueues", reflect.TypeOf((*MockNamesrvs)(nil).FetchSubscribeMessageQueues), topic)
 }
+
+// AddrList mocks base method
+func (m *MockNamesrvs) AddrList() []string {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "AddrList")
+	ret0, _ := ret[0].([]string)
+	return ret0
+}
+
+// AddrList indicates an expected call of AddrList
+func (mr *MockNamesrvsMockRecorder) AddrList() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddrList", reflect.TypeOf((*MockNamesrvs)(nil).AddrList))
+}
diff --git a/internal/model.go b/internal/model.go
index 8c95458..ea1fdac 100644
--- a/internal/model.go
+++ b/internal/model.go
@@ -18,9 +18,14 @@
 package internal
 
 import (
+	"bytes"
 	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
 
 	"github.com/apache/rocketmq-client-go/internal/utils"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
 )
 
@@ -100,3 +105,158 @@
 	rlog.Debug("heartbeat: "+string(d), nil)
 	return d
 }
+
+const (
+	PropNameServerAddr         = "PROP_NAMESERVER_ADDR"
+	PropThreadPoolCoreSize     = "PROP_THREADPOOL_CORE_SIZE"
+	PropConsumeOrderly         = "PROP_CONSUMEORDERLY"
+	PropConsumeType            = "PROP_CONSUME_TYPE"
+	PropClientVersion          = "PROP_CLIENT_VERSION"
+	PropConsumerStartTimestamp = "PROP_CONSUMER_START_TIMESTAMP"
+)
+
+type ProcessQueueInfo struct {
+	CommitOffset            int64 `json:"commitOffset"`
+	CachedMsgMinOffset      int64 `json:"cachedMsgMinOffset"`
+	CachedMsgMaxOffset      int64 `json:"cachedMsgMaxOffset"`
+	CachedMsgCount          int   `json:"cachedMsgCount"`
+	CachedMsgSizeInMiB      int64 `json:"cachedMsgSizeInMiB"`
+	TransactionMsgMinOffset int64 `json:"transactionMsgMinOffset"`
+	TransactionMsgMaxOffset int64 `json:"transactionMsgMaxOffset"`
+	TransactionMsgCount     int   `json:"transactionMsgCount"`
+	Locked                  bool  `json:"locked"`
+	TryUnlockTimes          int64 `json:"tryUnlockTimes"`
+	LastLockTimestamp       int64 `json:"lastLockTimestamp"`
+	Dropped                 bool  `json:"dropped"`
+	LastPullTimestamp       int64 `json:"lastPullTimestamp"`
+	LastConsumeTimestamp    int64 `json:"lastConsumeTimestamp"`
+}
+
+type ConsumeStatus struct {
+	PullRT            float64 `json:"pullRT"`
+	PullTPS           float64 `json:"pullTPS"`
+	ConsumeRT         float64 `json:"consumeRT"`
+	ConsumeOKTPS      float64 `json:"consumeOKTPS"`
+	ConsumeFailedTPS  float64 `json:"consumeFailedTPS"`
+	ConsumeFailedMsgs int64   `json:"consumeFailedMsgs"`
+}
+
+type ConsumerRunningInfo struct {
+	Properties       map[string]string
+	SubscriptionData map[*SubscriptionData]bool
+	MQTable          map[primitive.MessageQueue]ProcessQueueInfo
+	StatusTable      map[string]ConsumeStatus
+}
+
+func (info ConsumerRunningInfo) Encode() ([]byte, error) {
+	data, err := json.Marshal(info.Properties)
+	if err != nil {
+		return nil, err
+	}
+	jsonData := fmt.Sprintf("{\"%s\":%s", "properties", string(data))
+
+	data, err = json.Marshal(info.StatusTable)
+	if err != nil {
+		return nil, err
+	}
+	jsonData = fmt.Sprintf("%s,\"%s\":%s", jsonData, "statusTable", string(data))
+
+	subs := make([]*SubscriptionData, len(info.SubscriptionData))
+	idx := 0
+	for k := range info.SubscriptionData {
+		subs[idx] = k
+		idx++
+	}
+
+	// make sure test case table
+	sort.Slice(subs, func(i, j int) bool {
+		sub1 := subs[i]
+		sub2 := subs[j]
+		if sub1.ClassFilterMode != sub2.ClassFilterMode {
+			return sub1.ClassFilterMode == false
+		}
+		com := strings.Compare(sub1.Topic, sub1.Topic)
+		if com != 0 {
+			return com > 0
+		}
+
+		com = strings.Compare(sub1.SubString, sub1.SubString)
+		if com != 0 {
+			return com > 0
+		}
+
+		if sub1.SubVersion != sub2.SubVersion {
+			return sub1.SubVersion > sub2.SubVersion
+		}
+
+		com = strings.Compare(sub1.ExpType, sub1.ExpType)
+		if com != 0 {
+			return com > 0
+		}
+
+		v1, _ := sub1.Tags.MarshalJSON()
+		v2, _ := sub2.Tags.MarshalJSON()
+		com = bytes.Compare(v1, v2)
+		if com != 0 {
+			return com > 0
+		}
+
+		v1, _ = sub1.Codes.MarshalJSON()
+		v2, _ = sub2.Codes.MarshalJSON()
+		com = bytes.Compare(v1, v2)
+		if com != 0 {
+			return com > 0
+		}
+		return true
+	})
+
+	data, err = json.Marshal(subs)
+	if err != nil {
+		return nil, err
+	}
+	jsonData = fmt.Sprintf("%s,\"%s\":%s", jsonData, "subscriptionSet", string(data))
+
+	tableJson := ""
+	keys := make([]primitive.MessageQueue, 0)
+
+	for k := range info.MQTable {
+		keys = append(keys, k)
+	}
+
+	sort.Slice(keys, func(i, j int) bool {
+		q1 := keys[i]
+		q2 := keys[j]
+		com := strings.Compare(q1.Topic, q2.Topic)
+		if com != 0 {
+			return com < 0
+		}
+
+		com = strings.Compare(q1.BrokerName, q2.BrokerName)
+		if com != 0 {
+			return com < 0
+		}
+
+		return q1.QueueId < q2.QueueId
+	})
+
+	for idx := range keys {
+		dataK, err := json.Marshal(keys[idx])
+		if err != nil {
+			return nil, err
+		}
+		dataV, err := json.Marshal(info.MQTable[keys[idx]])
+		tableJson = fmt.Sprintf("%s,%s:%s", tableJson, string(dataK), string(dataV))
+	}
+	tableJson = strings.TrimLeft(tableJson, ",")
+	jsonData = fmt.Sprintf("%s,\"%s\":%s}", jsonData, "mqTable", fmt.Sprintf("{%s}", tableJson))
+	return []byte(jsonData), nil
+}
+
+func NewConsumerRunningInfo() *ConsumerRunningInfo {
+	return &ConsumerRunningInfo{
+		Properties:       make(map[string]string),
+		SubscriptionData: make(map[*SubscriptionData]bool),
+		MQTable:          make(map[primitive.MessageQueue]ProcessQueueInfo),
+		StatusTable:      make(map[string]ConsumeStatus),
+	}
+}
diff --git a/internal/model_test.go b/internal/model_test.go
index 1dac4ec..56703e0 100644
--- a/internal/model_test.go
+++ b/internal/model_test.go
@@ -20,12 +20,14 @@
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
 	"testing"
 
 	. "github.com/smartystreets/goconvey/convey"
-	"github.com/stretchr/testify/assert"
+	"github.com/tidwall/gjson"
 
 	"github.com/apache/rocketmq-client-go/internal/utils"
+	"github.com/apache/rocketmq-client-go/primitive"
 )
 
 func TestHeartbeatData(t *testing.T) {
@@ -43,7 +45,7 @@
 			set.Add(pData2)
 
 			v, err := json.Marshal(set)
-			assert.Nil(t, err)
+			So(err, ShouldBeNil)
 			fmt.Printf("json producer set: %s", string(v))
 		})
 
@@ -61,8 +63,7 @@
 			hbt.ProducerDatas.Add(p2)
 
 			v, err := json.Marshal(hbt)
-			//ShouldBeNil(t, err)
-			assert.Nil(t, err)
+			So(err, ShouldBeNil)
 			fmt.Printf("json producer: %s\n", string(v))
 		})
 
@@ -79,8 +80,7 @@
 			hbt.ConsumerDatas.Add(c2)
 
 			v, err := json.Marshal(hbt)
-			//ShouldBeNil(t, err)
-			assert.Nil(t, err)
+			So(err, ShouldBeNil)
 			fmt.Printf("json consumer: %s\n", string(v))
 		})
 
@@ -108,10 +108,257 @@
 			hbt.ConsumerDatas.Add(c2)
 
 			v, err := json.Marshal(hbt)
-			//ShouldBeNil(t, err)
-			assert.Nil(t, err)
+			So(err, ShouldBeNil)
 			fmt.Printf("json producer & consumer: %s\n", string(v))
 		})
 	})
 
 }
+
+func TestConsumerRunningInfo_MarshalJSON(t *testing.T) {
+	Convey("test ConsumerRunningInfo MarshalJson", t, func() {
+		props := map[string]string{
+			"maxReconsumeTimes":             "-1",
+			"unitMode":                      "false",
+			"adjustThreadPoolNumsThreshold": "100000",
+			"consumerGroup":                 "mq-client-go-test%GID_GO_TEST",
+			"messageModel":                  "CLUSTERING",
+			"suspendCurrentQueueTimeMillis": "1000",
+			"pullThresholdSizeForTopic":     "-1",
+			"pullThresholdSizeForQueue":     "100",
+			"PROP_CLIENT_VERSION":           "V4_5_1",
+			"consumeConcurrentlyMaxSpan":    "2000",
+			"postSubscriptionWhenPull":      "false",
+			"consumeTimestamp":              "20191127013617",
+			"PROP_CONSUME_TYPE":             "CONSUME_PASSIVELY",
+			"consumeTimeout":                "15",
+			"consumeMessageBatchMaxSize":    "1",
+			"PROP_THREADPOOL_CORE_SIZE":     "20",
+			"pullInterval":                  "0",
+			"pullThresholdForQueue":         "1000",
+			"pullThresholdForTopic":         "-1",
+			"consumeFromWhere":              "CONSUME_FROM_FIRST_OFFSET",
+			"PROP_NAMESERVER_ADDR":          "mq-client-go-test.mq-internet-access.mq-internet.aliyuncs.com:80;",
+			"pullBatchSize":                 "32",
+			"consumeThreadMin":              "20",
+			"PROP_CONSUMER_START_TIMESTAMP": "1574791577504",
+			"consumeThreadMax":              "20",
+			"subscription":                  "{}",
+			"PROP_CONSUMEORDERLY":           "false",
+		}
+		subData := map[*SubscriptionData]bool{
+			&SubscriptionData{
+				ClassFilterMode: false,
+				Codes:           utils.NewSet(),
+				ExpType:         "TAG",
+				SubString:       "*",
+				SubVersion:      1574791579242,
+				Tags:            utils.NewSet(),
+				Topic:           "%RETRY%mq-client-go-test%GID_GO_TEST",
+			}: true,
+			&SubscriptionData{
+				ClassFilterMode: true,
+				Codes:           utils.NewSet(),
+				ExpType:         "TAG",
+				SubString:       "*",
+				SubVersion:      1574791577523,
+				Tags:            utils.NewSet(),
+				Topic:           "mq-client-go-test%go-test",
+			}: true,
+		}
+		statusTable := map[string]ConsumeStatus{
+			"%RETRY%mq-client-go-test%GID_GO_TEST": {
+				PullRT:            11.11,
+				PullTPS:           22.22,
+				ConsumeRT:         33.33,
+				ConsumeOKTPS:      44.44,
+				ConsumeFailedTPS:  55.55,
+				ConsumeFailedMsgs: 666,
+			},
+			"mq-client-go-test%go-test": {
+				PullRT:            123,
+				PullTPS:           123,
+				ConsumeRT:         123,
+				ConsumeOKTPS:      123,
+				ConsumeFailedTPS:  123,
+				ConsumeFailedMsgs: 1234,
+			},
+		}
+		mqTable := map[primitive.MessageQueue]ProcessQueueInfo{
+			{
+				Topic:      "%RETRY%mq-client-go-test%GID_GO_TEST",
+				BrokerName: "qd7internet-01",
+				QueueId:    0,
+			}: {
+				CommitOffset:            0,
+				CachedMsgMinOffset:      0,
+				CachedMsgMaxOffset:      0,
+				CachedMsgCount:          0,
+				CachedMsgSizeInMiB:      0,
+				TransactionMsgMinOffset: 0,
+				TransactionMsgMaxOffset: 0,
+				TransactionMsgCount:     0,
+				Locked:                  false,
+				TryUnlockTimes:          0,
+				LastLockTimestamp:       1574791579221,
+				Dropped:                 false,
+				LastPullTimestamp:       1574791579242,
+				LastConsumeTimestamp:    1574791579221,
+			},
+			{
+				Topic:      "%RETRY%mq-client-go-test%GID_GO_TEST",
+				BrokerName: "qd7internet-01",
+				QueueId:    1,
+			}: {
+				CommitOffset:            1,
+				CachedMsgMinOffset:      2,
+				CachedMsgMaxOffset:      3,
+				CachedMsgCount:          4,
+				CachedMsgSizeInMiB:      5,
+				TransactionMsgMinOffset: 6,
+				TransactionMsgMaxOffset: 7,
+				TransactionMsgCount:     8,
+				Locked:                  true,
+				TryUnlockTimes:          9,
+				LastLockTimestamp:       1574791579221,
+				Dropped:                 false,
+				LastPullTimestamp:       1574791579242,
+				LastConsumeTimestamp:    1574791579221,
+			},
+		}
+		info := ConsumerRunningInfo{
+			Properties:       props,
+			SubscriptionData: subData,
+			StatusTable:      statusTable,
+			MQTable:          mqTable,
+		}
+		data, err := info.Encode()
+		So(err, ShouldBeNil)
+		result := gjson.ParseBytes(data)
+		Convey("test Properties fields", func() {
+			r1 := result.Get("properties")
+			So(r1.Exists(), ShouldBeTrue)
+			m := r1.Map()
+			So(len(m), ShouldEqual, 27)
+
+			So(m["PROP_CLIENT_VERSION"], ShouldNotBeEmpty)
+			So(m["PROP_CLIENT_VERSION"].String(), ShouldEqual, "V4_5_1")
+
+			So(m["PROP_CONSUME_TYPE"], ShouldNotBeNil)
+			So(m["PROP_CONSUME_TYPE"].String(), ShouldEqual, "CONSUME_PASSIVELY")
+
+			So(m["PROP_THREADPOOL_CORE_SIZE"], ShouldNotBeNil)
+			So(m["PROP_THREADPOOL_CORE_SIZE"].String(), ShouldEqual, "20")
+
+			So(m["PROP_NAMESERVER_ADDR"], ShouldNotBeNil)
+			So(m["PROP_NAMESERVER_ADDR"].String(), ShouldEqual, "mq-client-go-test.mq-internet-access.mq-internet.aliyuncs.com:80;")
+
+			So(m["PROP_CONSUMER_START_TIMESTAMP"], ShouldNotBeNil)
+			So(m["PROP_CONSUMER_START_TIMESTAMP"].String(), ShouldEqual, "1574791577504")
+
+			So(m["PROP_CONSUMEORDERLY"], ShouldNotBeNil)
+			So(m["PROP_CONSUMEORDERLY"].String(), ShouldEqual, "false")
+		})
+		Convey("test SubscriptionData fields", func() {
+			r2 := result.Get("subscriptionSet")
+			So(r2.Exists(), ShouldBeTrue)
+			arr := r2.Array()
+			So(len(arr), ShouldEqual, 2)
+
+			m1 := arr[0].Map()
+			So(len(m1), ShouldEqual, 7)
+			So(m1["classFilterMode"].Bool(), ShouldEqual, false)
+			So(len(m1["codes"].Array()), ShouldEqual, 0)
+			So(m1["expressionType"].String(), ShouldEqual, "TAG")
+			So(m1["subString"].String(), ShouldEqual, "*")
+			So(m1["subVersion"].Int(), ShouldEqual, 1574791579242)
+			So(len(m1["tags"].Array()), ShouldEqual, 0)
+			So(m1["topic"].String(), ShouldEqual, "%RETRY%mq-client-go-test%GID_GO_TEST")
+
+			m2 := arr[1].Map()
+			So(len(m2), ShouldEqual, 7)
+			So(m2["classFilterMode"].Bool(), ShouldEqual, true)
+			So(len(m2["codes"].Array()), ShouldEqual, 0)
+			So(m2["expressionType"].String(), ShouldEqual, "TAG")
+			So(m2["subString"].String(), ShouldEqual, "*")
+			So(m2["subVersion"].Int(), ShouldEqual, 1574791577523)
+			So(len(m2["tags"].Array()), ShouldEqual, 0)
+			So(m2["topic"].String(), ShouldEqual, "mq-client-go-test%go-test")
+		})
+		Convey("test StatusTable fields", func() {
+			r3 := result.Get("statusTable")
+			So(r3.Exists(), ShouldBeTrue)
+			m := r3.Map()
+			So(len(m), ShouldEqual, 2)
+
+			status1 := m["mq-client-go-test%go-test"].Map()
+			So(len(status1), ShouldEqual, 6)
+			So(status1["pullRT"].Float(), ShouldEqual, 123)
+			So(status1["pullTPS"].Float(), ShouldEqual, 123)
+			So(status1["consumeRT"].Float(), ShouldEqual, 123)
+			So(status1["consumeOKTPS"].Float(), ShouldEqual, 123)
+			So(status1["consumeFailedTPS"].Float(), ShouldEqual, 123)
+			So(status1["consumeFailedMsgs"].Int(), ShouldEqual, 1234)
+
+			status2 := m["%RETRY%mq-client-go-test%GID_GO_TEST"].Map()
+			So(len(status2), ShouldEqual, 6)
+			So(status2["pullRT"].Float(), ShouldEqual, 11.11)
+			So(status2["pullTPS"].Float(), ShouldEqual, 22.22)
+			So(status2["consumeRT"].Float(), ShouldEqual, 33.33)
+			So(status2["consumeOKTPS"].Float(), ShouldEqual, 44.44)
+			So(status2["consumeFailedTPS"].Float(), ShouldEqual, 55.55)
+			So(status2["consumeFailedMsgs"].Int(), ShouldEqual, 666)
+		})
+		Convey("test MQTable fields", func() {
+			r4 := result.Get("mqTable")
+			So(r4.Exists(), ShouldBeTrue)
+			objNumbers := strings.Split(r4.String(), "},{")
+			So(len(objNumbers), ShouldEqual, 2)
+
+			obj1Str := objNumbers[0][1:len(objNumbers[0])] + "}"
+			obj1KV := strings.Split(obj1Str, "}:{")
+			So(len(obj1KV), ShouldEqual, 2)
+
+			obj1 := gjson.Parse("{" + obj1KV[1][0:len(obj1KV[1])])
+			So(obj1.Exists(), ShouldBeTrue)
+			obj1M := obj1.Map()
+			So(len(obj1M), ShouldEqual, 14)
+			So(obj1M["commitOffset"].Int(), ShouldEqual, 0)
+			So(obj1M["cachedMsgMinOffset"].Int(), ShouldEqual, 0)
+			So(obj1M["cachedMsgMaxOffset"].Int(), ShouldEqual, 0)
+			So(obj1M["cachedMsgCount"].Int(), ShouldEqual, 0)
+			So(obj1M["cachedMsgSizeInMiB"].Int(), ShouldEqual, 0)
+			So(obj1M["transactionMsgMinOffset"].Int(), ShouldEqual, 0)
+			So(obj1M["transactionMsgMaxOffset"].Int(), ShouldEqual, 0)
+			So(obj1M["transactionMsgCount"].Int(), ShouldEqual, 0)
+			So(obj1M["locked"].Bool(), ShouldEqual, false)
+			So(obj1M["tryUnlockTimes"].Int(), ShouldEqual, 0)
+			So(obj1M["lastLockTimestamp"].Int(), ShouldEqual, 1574791579221)
+			So(obj1M["dropped"].Bool(), ShouldEqual, false)
+			So(obj1M["lastPullTimestamp"].Int(), ShouldEqual, 1574791579242)
+			So(obj1M["lastConsumeTimestamp"].Int(), ShouldEqual, 1574791579221)
+
+			obj2Str := "{" + objNumbers[1][0:len(objNumbers[1])-1]
+			obj2KV := strings.Split(obj2Str, "}:{")
+			So(len(obj2KV), ShouldEqual, 2)
+			obj2 := gjson.Parse("{" + obj2KV[1][0:len(obj2KV[1])])
+			So(obj2.Exists(), ShouldBeTrue)
+			obj2M := obj2.Map()
+			So(len(obj2M), ShouldEqual, 14)
+			So(obj2M["commitOffset"].Int(), ShouldEqual, 1)
+			So(obj2M["cachedMsgMinOffset"].Int(), ShouldEqual, 2)
+			So(obj2M["cachedMsgMaxOffset"].Int(), ShouldEqual, 3)
+			So(obj2M["cachedMsgCount"].Int(), ShouldEqual, 4)
+			So(obj2M["cachedMsgSizeInMiB"].Int(), ShouldEqual, 5)
+			So(obj2M["transactionMsgMinOffset"].Int(), ShouldEqual, 6)
+			So(obj2M["transactionMsgMaxOffset"].Int(), ShouldEqual, 7)
+			So(obj2M["transactionMsgCount"].Int(), ShouldEqual, 8)
+			So(obj2M["locked"].Bool(), ShouldEqual, true)
+			So(obj2M["tryUnlockTimes"].Int(), ShouldEqual, 9)
+			So(obj2M["lastLockTimestamp"].Int(), ShouldEqual, 1574791579221)
+			So(obj2M["dropped"].Bool(), ShouldEqual, false)
+			So(obj2M["lastPullTimestamp"].Int(), ShouldEqual, 1574791579242)
+			So(obj2M["lastConsumeTimestamp"].Int(), ShouldEqual, 1574791579221)
+		})
+	})
+}
diff --git a/internal/namesrv.go b/internal/namesrv.go
index 3c20968..b9f1744 100644
--- a/internal/namesrv.go
+++ b/internal/namesrv.go
@@ -52,6 +52,8 @@
 	FindBrokerAddressInSubscribe(brokerName string, brokerId int64, onlyThisBroker bool) *FindBrokerResult
 
 	FetchSubscribeMessageQueues(topic string) ([]*primitive.MessageQueue, error)
+
+	AddrList() []string
 }
 
 // namesrvs rocketmq namesrv instance.
@@ -119,3 +121,7 @@
 func (s *namesrvs) SetCredentials(credentials primitive.Credentials) {
 	s.nameSrvClient.RegisterInterceptor(remote.ACLInterceptor(credentials))
 }
+
+func (s *namesrvs) AddrList() []string {
+	return s.srvs
+}
diff --git a/internal/request.go b/internal/request.go
index 76d0d7d..470fd35 100644
--- a/internal/request.go
+++ b/internal/request.go
@@ -78,10 +78,6 @@
 	return maps
 }
 
-func (request *SendMessageRequestHeader) Decode(properties map[string]string) error {
-	return nil
-}
-
 type EndTransactionRequestHeader struct {
 	ProducerGroup        string
 	TranStateTableOffset int64
@@ -123,23 +119,23 @@
 	return maps
 }
 
-func (request *CheckTransactionStateRequestHeader) Decode(ext map[string]string) {
-	if len(ext) == 0 {
+func (request *CheckTransactionStateRequestHeader) Decode(properties map[string]string) {
+	if len(properties) == 0 {
 		return
 	}
-	if v, existed := ext["tranStateTableOffset"]; existed {
+	if v, existed := properties["tranStateTableOffset"]; existed {
 		request.TranStateTableOffset, _ = strconv.ParseInt(v, 10, 0)
 	}
-	if v, existed := ext["commitLogOffset"]; existed {
+	if v, existed := properties["commitLogOffset"]; existed {
 		request.CommitLogOffset, _ = strconv.ParseInt(v, 10, 0)
 	}
-	if v, existed := ext["msgId"]; existed {
+	if v, existed := properties["msgId"]; existed {
 		request.MsgId = v
 	}
-	if v, existed := ext["transactionId"]; existed {
+	if v, existed := properties["transactionId"]; existed {
 		request.MsgId = v
 	}
-	if v, existed := ext["offsetMsgId"]; existed {
+	if v, existed := properties["offsetMsgId"]; existed {
 		request.MsgId = v
 	}
 }
@@ -273,6 +269,27 @@
 	return maps
 }
 
-func (request *GetRouteInfoRequestHeader) Decode(properties map[string]string) error {
-	return nil
+type GetConsumerRunningInfoHeader struct {
+	consumerGroup string
+	clientID      string
+}
+
+func (request *GetConsumerRunningInfoHeader) Encode() map[string]string {
+	maps := make(map[string]string)
+	maps["consumerGroup"] = request.consumerGroup
+	maps["clientId"] = request.clientID
+	return maps
+}
+
+func (request *GetConsumerRunningInfoHeader) Decode(properties map[string]string) {
+	if len(properties) == 0 {
+		return
+	}
+	if v, existed := properties["consumerGroup"]; existed {
+		request.consumerGroup = v
+	}
+
+	if v, existed := properties["clientId"]; existed {
+		request.clientID = v
+	}
 }