diff --git a/core/api.go b/core/api.go
index 0f4b6f9..39240ba 100644
--- a/core/api.go
+++ b/core/api.go
@@ -77,6 +77,15 @@
 	Subscribe(topic, expression string, consumeFunc func(msg *MessageExt) ConsumeStatus) error
 }
 
+// PullConsumer consumer pulling the message
+type PullConsumer interface {
+	baseAPI
+	// Pull returns the messages from the consume queue by specify the offset and the max number
+	Pull(mq MessageQueue, subExpression string, offset int64, maxNums int) PullResult
+	// FetchSubscriptionMessageQueues returns the consume queue of the topic
+	FetchSubscriptionMessageQueues(topic string) []MessageQueue
+}
+
 type SessionCredentials struct {
 	AccessKey string
 	SecretKey string
diff --git a/core/log.go b/core/log.go
new file mode 100644
index 0000000..e081e4c
--- /dev/null
+++ b/core/log.go
@@ -0,0 +1,50 @@
+/*
+ * 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 rocketmq
+
+/*
+#include "rocketmq/CCommon.h"
+*/
+import "C"
+import "fmt"
+
+// LogLevel the log level
+type LogLevel int
+
+// predefined log level
+const (
+	LogLevelFatal = LogLevel(C.E_LOG_LEVEL_FATAL)
+	LogLevelError = LogLevel(C.E_LOG_LEVEL_ERROR)
+	LogLevelWarn  = LogLevel(C.E_LOG_LEVEL_WARN)
+	LogLevelInfo  = LogLevel(C.E_LOG_LEVEL_INFO)
+	LogLevelDebug = LogLevel(C.E_LOG_LEVEL_DEBUG)
+	LogLevelTrace = LogLevel(C.E_LOG_LEVEL_TRACE)
+	LogLevelNum   = LogLevel(C.E_LOG_LEVEL_LEVEL_NUM)
+)
+
+// LogConfig the log configuration for the pull consumer
+type LogConfig struct {
+	Path     string
+	FileNum  int
+	FileSize int64
+	Level    LogLevel
+}
+
+func (lc *LogConfig) String() string {
+	return fmt.Sprintf("%+v", *lc)
+}
diff --git a/core/pull_consumer.go b/core/pull_consumer.go
new file mode 100644
index 0000000..19a033c
--- /dev/null
+++ b/core/pull_consumer.go
@@ -0,0 +1,225 @@
+/*
+ * 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 rocketmq
+
+/*
+#cgo LDFLAGS: -L/usr/local/lib -lrocketmq
+
+#include <stdio.h>
+#include <stdlib.h>
+#include "rocketmq/CMessageExt.h"
+#include "rocketmq/CPullConsumer.h"
+*/
+import "C"
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"unsafe"
+)
+
+// PullStatus pull status
+type PullStatus int
+
+// predefined pull status
+const (
+	PullFound         = PullStatus(C.E_FOUND)
+	PullNoNewMsg      = PullStatus(C.E_NO_NEW_MSG)
+	PullNoMatchedMsg  = PullStatus(C.E_NO_MATCHED_MSG)
+	PullOffsetIllegal = PullStatus(C.E_OFFSET_ILLEGAL)
+	PullBrokerTimeout = PullStatus(C.E_BROKER_TIMEOUT)
+)
+
+func (ps PullStatus) String() string {
+	switch ps {
+	case PullFound:
+		return "Found"
+	case PullNoNewMsg:
+		return "NoNewMsg"
+	case PullNoMatchedMsg:
+		return "NoMatchedMsg"
+	case PullOffsetIllegal:
+		return "OffsetIllegal"
+	case PullBrokerTimeout:
+		return "BrokerTimeout"
+	default:
+		return "Unknown status"
+	}
+}
+
+// PullConsumerConfig the configuration for the pull consumer
+type PullConsumerConfig struct {
+	GroupID     string
+	NameServer  string
+	Credentials *SessionCredentials
+	Log         *LogConfig
+}
+
+// DefaultPullConsumer default consumer pulling the message
+type DefaultPullConsumer struct {
+	PullConsumerConfig
+	cconsumer *C.struct_CPullConsumer
+	funcsMap  sync.Map
+}
+
+func (c *DefaultPullConsumer) String() string {
+	topics := ""
+	c.funcsMap.Range(func(key, value interface{}) bool {
+		topics += key.(string) + ", "
+		return true
+	})
+	return fmt.Sprintf("[%+v, subcribed topics: [%s]]", c.PullConsumerConfig, topics)
+}
+
+// NewPullConsumer creates one pull consumer
+func NewPullConsumer(conf *PullConsumerConfig) (*DefaultPullConsumer, error) {
+	cs := C.CString(conf.GroupID)
+	cconsumer := C.CreatePullConsumer(cs)
+	C.free(unsafe.Pointer(cs))
+
+	cs = C.CString(conf.NameServer)
+	C.SetPullConsumerNameServerAddress(cconsumer, cs)
+	C.free(unsafe.Pointer(cs))
+
+	log := conf.Log
+	if log != nil {
+		cs = C.CString(log.Path)
+		if C.SetPullConsumerLogPath(cconsumer, cs) != 0 {
+			return nil, errors.New("new pull consumer error:set log path failed")
+		}
+		C.free(unsafe.Pointer(cs))
+
+		if C.SetPullConsumerLogFileNumAndSize(cconsumer, C.int(log.FileNum), C.long(log.FileSize)) != 0 {
+			return nil, errors.New("new pull consumer error:set log file num and size failed")
+		}
+		if C.SetPullConsumerLogLevel(cconsumer, C.CLogLevel(log.Level)) != 0 {
+			return nil, errors.New("new pull consumer error:set log level failed")
+		}
+	}
+
+	if conf.Credentials != nil {
+		ak := C.CString(conf.Credentials.AccessKey)
+		sk := C.CString(conf.Credentials.SecretKey)
+		ch := C.CString(conf.Credentials.Channel)
+		C.SetPullConsumerSessionCredentials(cconsumer, ak, sk, ch)
+
+		C.free(unsafe.Pointer(ak))
+		C.free(unsafe.Pointer(sk))
+		C.free(unsafe.Pointer(ch))
+	}
+
+	return &DefaultPullConsumer{PullConsumerConfig: *conf, cconsumer: cconsumer}, nil
+}
+
+// Start starts the pulling conumser
+func (c *DefaultPullConsumer) Start() error {
+	r := C.StartPullConsumer(c.cconsumer)
+	if r != 0 {
+		return fmt.Errorf("start failed, code:%d", r)
+	}
+	return nil
+}
+
+// Shutdown shutdown the pulling conumser
+func (c *DefaultPullConsumer) Shutdown() error {
+	r := C.ShutdownPullConsumer(c.cconsumer)
+	if r != 0 {
+		return fmt.Errorf("shutdown failed, code:%d", r)
+	}
+
+	r = C.DestroyPullConsumer(c.cconsumer)
+	if r != 0 {
+		return fmt.Errorf("destory failed, code:%d", r)
+	}
+	return nil
+}
+
+// FetchSubscriptionMessageQueues returns the topic's consume queue
+func (c *DefaultPullConsumer) FetchSubscriptionMessageQueues(topic string) []MessageQueue {
+	var (
+		q    *C.struct__CMessageQueue_
+		size C.int
+	)
+
+	ctopic := C.CString(topic)
+	C.FetchSubscriptionMessageQueues(c.cconsumer, ctopic, &q, &size)
+	C.free(unsafe.Pointer(ctopic))
+	if size == 0 {
+		return nil
+	}
+
+	qs := make([]MessageQueue, size)
+	for i := range qs {
+		cq := (*C.struct__CMessageQueue_)(
+			unsafe.Pointer(uintptr(unsafe.Pointer(q)) + uintptr(i)*unsafe.Sizeof(*q)),
+		)
+		qs[i].ID, qs[i].Broker, qs[i].Topic = int(cq.queueId), C.GoString(&cq.brokerName[0]), topic
+	}
+	C.ReleaseSubscriptionMessageQueue(q)
+
+	return qs
+}
+
+// PullResult the pull result
+type PullResult struct {
+	NextBeginOffset int64
+	MinOffset       int64
+	MaxOffset       int64
+	Status          PullStatus
+	Messages        []*MessageExt
+}
+
+func (pr *PullResult) String() string {
+	return fmt.Sprintf("%+v", *pr)
+}
+
+// Pull pulling the message from the specified message queue
+func (c *DefaultPullConsumer) Pull(mq MessageQueue, subExpression string, offset int64, maxNums int) PullResult {
+	cmq := C.struct__CMessageQueue_{
+		queueId: C.int(mq.ID),
+	}
+
+	copy(cmq.topic[:], *(*[]C.char)(unsafe.Pointer(&mq.Topic)))
+	copy(cmq.brokerName[:], *(*[]C.char)(unsafe.Pointer(&mq.Broker)))
+
+	csubExpr := C.CString(subExpression)
+	cpullResult := C.Pull(c.cconsumer, &cmq, csubExpr, C.longlong(offset), C.int(maxNums))
+
+	pr := PullResult{
+		NextBeginOffset: int64(cpullResult.nextBeginOffset),
+		MinOffset:       int64(cpullResult.minOffset),
+		MaxOffset:       int64(cpullResult.maxOffset),
+		Status:          PullStatus(cpullResult.pullStatus),
+	}
+	if cpullResult.size > 0 {
+		msgs := make([]*MessageExt, cpullResult.size)
+		for i := range msgs {
+			msgs[i] = cmsgExtToGo(*(**C.struct_CMessageExt)(
+				unsafe.Pointer(
+					uintptr(unsafe.Pointer(cpullResult.msgFoundList)) + uintptr(i)*unsafe.Sizeof(*cpullResult.msgFoundList),
+				),
+			))
+		}
+		pr.Messages = msgs
+	}
+
+	C.free(unsafe.Pointer(csubExpr))
+	C.ReleasePullResult(cpullResult)
+	return pr
+}
diff --git a/core/queue.go b/core/queue.go
new file mode 100644
index 0000000..47f10ef
--- /dev/null
+++ b/core/queue.go
@@ -0,0 +1,31 @@
+/*
+ * 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 rocketmq
+
+import "fmt"
+
+// MessageQueue the queue of the message
+type MessageQueue struct {
+	Topic  string
+	Broker string
+	ID     int
+}
+
+func (q *MessageQueue) String() string {
+	return fmt.Sprintf("broker:%s, topic:%s, id:%d", q.Broker, q.Topic, q.ID)
+}
diff --git a/examples/pullconsumer/consumer.go b/examples/pullconsumer/consumer.go
new file mode 100644
index 0000000..3643a22
--- /dev/null
+++ b/examples/pullconsumer/consumer.go
@@ -0,0 +1,113 @@
+/*
+ * 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 main
+
+import (
+	"flag"
+	"fmt"
+	"time"
+
+	rocketmq "github.com/apache/rocketmq-client-go/core"
+)
+
+var (
+	namesrvAddrs string
+	topic        string
+	body         string
+	groupID      string
+	msgCount     int64
+	workerCount  int
+)
+
+func init() {
+	flag.StringVar(&namesrvAddrs, "n", "", "name server address")
+	flag.StringVar(&topic, "t", "", "topic")
+	flag.StringVar(&groupID, "g", "", "group")
+}
+
+// ./consumer -n "localhost:9876" -t test -g local_test
+func main() {
+	flag.Parse()
+
+	if namesrvAddrs == "" {
+		fmt.Println("empty namesrv")
+		return
+	}
+
+	if topic == "" {
+		fmt.Println("empty topic")
+		return
+	}
+
+	if groupID == "" {
+		fmt.Println("empty groupID")
+		return
+	}
+
+	consumer, err := rocketmq.NewPullConsumer(&rocketmq.PullConsumerConfig{
+		GroupID:    groupID,
+		NameServer: namesrvAddrs,
+		Log: &rocketmq.LogConfig{
+			Path: "example",
+		},
+	})
+	if err != nil {
+		fmt.Printf("new pull consumer error:%s\n", err)
+		return
+	}
+
+	err = consumer.Start()
+	if err != nil {
+		fmt.Printf("start consumer error:%s\n", err)
+		return
+	}
+	defer consumer.Shutdown()
+
+	mqs := consumer.FetchSubscriptionMessageQueues(topic)
+	fmt.Printf("fetch subscription mqs:%+v\n", mqs)
+
+	total, offsets, now := 0, map[int]int64{}, time.Now()
+
+PULL:
+	for {
+		for _, mq := range mqs {
+			pr := consumer.Pull(mq, "*", offsets[mq.ID], 32)
+			total += len(pr.Messages)
+			fmt.Printf("pull %s, result:%+v\n", mq.String(), pr)
+
+			switch pr.Status {
+			case rocketmq.PullNoNewMsg:
+				break PULL
+			case rocketmq.PullFound:
+				fallthrough
+			case rocketmq.PullNoMatchedMsg:
+				fallthrough
+			case rocketmq.PullOffsetIllegal:
+				offsets[mq.ID] = pr.NextBeginOffset
+			case rocketmq.PullBrokerTimeout:
+				fmt.Println("broker timeout occur")
+			}
+		}
+	}
+
+	var timePerMessage time.Duration
+	if total > 0 {
+		timePerMessage = time.Since(now) / time.Duration(total)
+	}
+	fmt.Printf("total message:%d, per message time:%d\n", total, timePerMessage)
+}