diff --git a/core/log.go b/core/log.go
new file mode 100644
index 0000000..d9b87e8
--- /dev/null
+++ b/core/log.go
@@ -0,0 +1,45 @@
+/*
+ * 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"
+
+// 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
+}
diff --git a/core/pull_consumer.go b/core/pull_consumer.go
index b9c7338..e236ee6 100644
--- a/core/pull_consumer.go
+++ b/core/pull_consumer.go
@@ -14,6 +14,7 @@
  *  See the License for the specific language governing permissions and
  *  limitations under the License.
  */
+
 package rocketmq
 
 /*
@@ -27,18 +28,37 @@
 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)
+)
+
+// 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 {
-	config    *ConsumerConfig
+	PullConsumerConfig
 	cconsumer *C.struct_CPullConsumer
 	funcsMap  sync.Map
-
-	resources []*C.char
 }
 
 func (c *DefaultPullConsumer) String() string {
@@ -47,48 +67,141 @@
 		topics += key.(string) + ", "
 		return true
 	})
-	return fmt.Sprintf("[%s, subcribed topics: [%s]]", c.config, topics)
+	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 {
-	C.StartPullConsumer(c.cconsumer)
+	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 {
-	C.ShutdownPullConsumer(c.cconsumer)
-	C.DestroyPullConsumer(c.cconsumer)
-	for _, r := range c.resources {
-		C.free(unsafe.Pointer(r))
+	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 fetchs the topic's subcripted message queues
 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
+}
 
-CPullConsumer *CreatePullConsumer(const char *groupId);
-int DestroyPullConsumer(CPullConsumer *consumer);
-int StartPullConsumer(CPullConsumer *consumer);
-int ShutdownPullConsumer(CPullConsumer *consumer);
-int SetPullConsumerGroupID(CPullConsumer *consumer, const char *groupId);
-const char *GetPullConsumerGroupID(CPullConsumer *consumer);
-int SetPullConsumerNameServerAddress(CPullConsumer *consumer, const char *namesrv);
-int SetPullConsumerSessionCredentials(CPullConsumer *consumer, const char *accessKey, const char *secretKey,
-                                     const char *channel);
-int SetPullConsumerLogPath(CPullConsumer *consumer, const char *logPath);
-int SetPullConsumerLogFileNumAndSize(CPullConsumer *consumer, int fileNum, long fileSize);
-int SetPullConsumerLogLevel(CPullConsumer *consumer, CLogLevel level);
+// 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),
+	}
 
-int FetchSubscriptionMessageQueues(CPullConsumer *consumer, const char *topic, CMessageQueue **mqs , int* size);
-int ReleaseSubscriptionMessageQueue(CMessageQueue *mqs);
+	copy(cmq.topic[:], *(*[]C.char)(unsafe.Pointer(&mq.Topic)))
+	copy(cmq.brokerName[:], *(*[]C.char)(unsafe.Pointer(&mq.Broker)))
+	fmt.Printf("%+v\n", []byte(mq.Topic))
+	fmt.Printf("%+v\n", cmq.topic)
+	fmt.Printf("%+v\n", []byte(mq.Broker))
+	fmt.Printf("%+v\n", cmq.brokerName)
 
-CPullResult Pull(CPullConsumer *consumer,const CMessageQueue *mq, const char *subExpression, long long offset, int maxNums);
-int ReleasePullResult(CPullResult pullResult);
+	csubExpr := C.CString(subExpression)
+	cpullResult := C.Pull(c.cconsumer, &cmq, csubExpr, C.longlong(offset), C.int(maxNums))
 
-*/
+	pullResult := 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),
+				),
+			))
+		}
+	}
+
+	C.free(unsafe.Pointer(csubExpr))
+	C.ReleasePullResult(cpullResult)
+	return pullResult
+}
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..875d5eb
--- /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:9988" -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)
+}