You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2018/12/14 02:38:34 UTC
[GitHub] dongeforever closed pull request #6: Pull consumer
dongeforever closed pull request #6: Pull consumer
URL: https://github.com/apache/rocketmq-client-go/pull/6
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core/api.go b/core/api.go
index 92619f3..b64cc97 100644
--- a/core/api.go
+++ b/core/api.go
@@ -29,7 +29,7 @@ func NewProduer(config *ProducerConfig) Producer {
// ProducerConfig define a producer
type ProducerConfig struct {
- GroupID string
+ GroupID string
NameServer string
Credentials *SessionCredentials
}
@@ -74,6 +74,15 @@ type PushConsumer interface {
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)
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services