Enrich introduction.md (#21)
* 1: add push order message and pull consumer
2: using `install` rather than `cp` to install dynamic library
* update example code
diff --git a/doc/Introduction.md b/doc/Introduction.md
index c3a825b..9ad1650 100644
--- a/doc/Introduction.md
+++ b/doc/Introduction.md
@@ -27,7 +27,7 @@
- `git clone https://github.com/apache/rocketmq-client-cpp`
- `cd rocketmq-client-cpp`
- `sudo sh build.sh`
- - `cp bin/librocketmq.dylib /usr/local/lib`
+ - `sudo install bin/librocketmq.dylib /usr/local/lib`
- `sudo mkdir /usr/local/include/rocketmq`
- `sudo cp include/* /usr/local/include/rocketmq/`
@@ -44,41 +44,162 @@
- import package
```
- import "github.com/apache/rocketmq-client-go/core"
+ import rocketmq "github.com/apache/rocketmq-client-go/core"
```
- Send message
```go
func SendMessagge(){
- producer := rocketmq.NewProducer(&rocketmq.ProducerConfig{GroupID: "testGroup", NameServer: "localhost:9876"})
+ producer := rocketmq.NewProducer(config)
producer.Start()
defer producer.Shutdown()
fmt.Printf("Producer: %s started... \n", producer)
for i := 0; i < 100; i++ {
- msg := fmt.Sprintf("Hello RocketMQ-%d", i)
- result := producer.SendMessageSync(&rocketmq.Message{Topic: "test", Body: msg})
- fmt.Println(fmt.Sprintf("send message: %s result: %s", msg, result))
+ msg := fmt.Sprintf("%s-*d", *body, i)
+ result, err := producer.SendMessageSync(&rocketmq.Message{Topic: "test", Body: msg})
+ if err != nil {
+ fmt.Println("Error:", err)
+ }
+ fmt.Printf("send message: %s result: %s\n", msg, result)
}
- time.Sleep(10 * time.Second)
- producer.Shutdown()
+ }
+ ```
+- Send ordered message
+ ```go
+ type queueSelectorByOrderID struct{}
+
+ func (s queueSelectorByOrderID) Select(size int, m *rocketmq.Message, arg interface{}) int{
+ return arg.(int) % size
+ }
+ type worker struct {
+ p rocketmq.Producer
+ leftMsgCount int64
+ }
+
+ func (w *worker) run() {
+ selector := queueSelectorByOrderID{}
+ for atomic.AddInt64(&w.leftMsgCount, -1) >= 0 {
+ r, err := w.p.SendMessageOrderly(
+ &rocketmq.Message{Topic: *topic, Body: *body}, selector, 7 /*orderID*/, 3,
+ )
+ if err != nil {
+ println("Send Orderly Error:", err)
+ }
+ fmt.Printf("send orderly result:%+v\n", r)
+ }
+ }
+
+ func sendMessageOrderly(config *rocketmq.ProducerConfig) {
+ producer, err := rocketmq.NewProducer(config)
+ if err != nil {
+ fmt.Println("create Producer failed, error:", err)
+ return
+ }
+
+ producer.Start()
+ defer producer.Shutdown()
+
+ wg := sync.WaitGroup{}
+ wg.Add(*workerCount)
+
+ workers := make([]worker, *workerCount)
+ for i := range workers {
+ workers[i].p = producer
+ workers[i].leftMsgCount = (int64)(*amount)
+ }
+
+ for i := range workers {
+ go func(w *worker) { w.run(); wg.Done() }(&workers[i])
+ }
+
+ wg.Wait()
}
```
- Push Consumer
```go
- func PushConsumeMessage() {
- fmt.Println("Start Receiving Messages...")
- consumer, _ := rocketmq.NewPushConsumer(&rocketmq.ConsumerConfig{
- GroupID: "testGroupId",
- NameServer: "localhost:9876",
- ConsumerThreadCount: 2,
- MessageBatchMaxSize: 16})
+ func ConsumeWithPush(config *rocketmq.PushConsumerConfig) {
+
+ consumer, err := rocketmq.NewPushConsumer(config)
+ if err != nil {
+ println("create Consumer failed, error:", err)
+ return
+ }
+
+ ch := make(chan interface{})
+ var count = (int64)(*amount)
// MUST subscribe topic before consumer started.
- consumer.Subscribe("test", "*", func(msg *rocketmq.MessageExt)rocketmq.ConsumeStatus {
+ consumer.Subscribe("test", "*", func(msg *rocketmq.MessageExt) rocketmq.ConsumeStatus {
fmt.Printf("A message received: \"%s\" \n", msg.Body)
- return rocketmq.ConsumeSuccess})
- consumer.Start()
- defer consumer.Shutdown()
+ if atomic.AddInt64(&count, -1) <= 0 {
+ ch <- "quit"
+ }
+ return rocketmq.ConsumeSuccess
+ })
+
+ err = consumer.Start()
+ if err != nil {
+ println("consumer start failed,", err)
+ return
+ }
+
fmt.Printf("consumer: %s started...\n", consumer)
- time.Sleep(10 * time.Minute)
+ <-ch
+ err = consumer.Shutdown()
+ if err != nil {
+ println("consumer shutdown failed")
+ return
+ }
+ println("consumer has shutdown.")
+ }
+ ```
+- Pull Consumer
+ ```go
+ func ConsumeWithPull(config *rocketmq.PullConsumerConfig, topic string) {
+
+ consumer, err := rocketmq.NewPullConsumer(config)
+ 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)
}
```
- [Full example](../examples)
\ No newline at end of file
