You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by di...@apache.org on 2019/12/18 09:37:18 UTC
[rocketmq-client-go] branch native updated: feat(producer): support
batch message send (#332)
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch native
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-go.git
The following commit(s) were added to refs/heads/native by this push:
new c57a000 feat(producer): support batch message send (#332)
c57a000 is described below
commit c57a000c270666429c49c35e92323ee4f740658a
Author: xujianhai666 <52...@users.noreply.github.com>
AuthorDate: Wed Dec 18 17:37:11 2019 +0800
feat(producer): support batch message send (#332)
- add batch message
- modify producer interface
---
api.go | 6 +-
examples/consumer/simple/main.go | 5 +-
.../{consumer/simple => producer/batch}/main.go | 39 +++++++------
internal/request.go | 46 +++++++++++----
primitive/message.go | 35 ++++++++++++
producer/producer.go | 65 ++++++++++++++++++----
6 files changed, 152 insertions(+), 44 deletions(-)
diff --git a/api.go b/api.go
index 931dbcf..4a7a90e 100644
--- a/api.go
+++ b/api.go
@@ -29,10 +29,10 @@ import (
type Producer interface {
Start() error
Shutdown() error
- SendSync(ctx context.Context, mq *primitive.Message) (*primitive.SendResult, error)
+ SendSync(ctx context.Context, mq ...*primitive.Message) (*primitive.SendResult, error)
SendAsync(ctx context.Context, mq func(ctx context.Context, result *primitive.SendResult, err error),
- msg *primitive.Message) error
- SendOneWay(ctx context.Context, mq *primitive.Message) error
+ msg ...*primitive.Message) error
+ SendOneWay(ctx context.Context, mq ...*primitive.Message) error
}
func NewProducer(opts ...producer.Option) (Producer, error) {
diff --git a/examples/consumer/simple/main.go b/examples/consumer/simple/main.go
index 33ff6ee..c3ec35c 100644
--- a/examples/consumer/simple/main.go
+++ b/examples/consumer/simple/main.go
@@ -35,7 +35,10 @@ func main() {
)
err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx context.Context,
msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
- fmt.Printf("subscribe callback: %v \n", msgs)
+ for i := range msgs {
+ fmt.Printf("subscribe callback: %v \n", msgs[i])
+ }
+
return consumer.ConsumeSuccess, nil
})
if err != nil {
diff --git a/examples/consumer/simple/main.go b/examples/producer/batch/main.go
similarity index 58%
copy from examples/consumer/simple/main.go
copy to examples/producer/batch/main.go
index 33ff6ee..da44480 100644
--- a/examples/consumer/simple/main.go
+++ b/examples/producer/batch/main.go
@@ -21,35 +21,38 @@ import (
"context"
"fmt"
"os"
- "time"
+ "strconv"
"github.com/apache/rocketmq-client-go"
- "github.com/apache/rocketmq-client-go/consumer"
"github.com/apache/rocketmq-client-go/primitive"
+ "github.com/apache/rocketmq-client-go/producer"
)
func main() {
- c, _ := rocketmq.NewPushConsumer(
- consumer.WithGroupName("testGroup"),
- consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+ p, _ := rocketmq.NewProducer(
+ producer.WithNameServer([]string{"127.0.0.1:9876"}),
+ producer.WithRetry(2),
)
- err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx context.Context,
- msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
- fmt.Printf("subscribe callback: %v \n", msgs)
- return consumer.ConsumeSuccess, nil
- })
+ err := p.Start()
if err != nil {
- fmt.Println(err.Error())
+ fmt.Printf("start producer error: %s", err.Error())
+ os.Exit(1)
}
- // Note: start after subscribe
- err = c.Start()
+ var msgs []*primitive.Message
+ for i := 0; i < 10; i++ {
+ msgs = append(msgs, primitive.NewMessage("test",
+ []byte("Hello RocketMQ Go Client! num: "+strconv.Itoa(i))))
+ }
+
+ res, err := p.SendSync(context.Background(), msgs...)
+
if err != nil {
- fmt.Println(err.Error())
- os.Exit(-1)
+ fmt.Printf("send message error: %s\n", err)
+ } else {
+ fmt.Printf("send message success: result=%s\n", res.String())
}
- time.Sleep(time.Hour)
- err = c.Shutdown()
+ err = p.Shutdown()
if err != nil {
- fmt.Printf("shundown Consumer error: %s", err.Error())
+ fmt.Printf("shutdown producer error: %s", err.Error())
}
}
diff --git a/internal/request.go b/internal/request.go
index 470fd35..3e477ea 100644
--- a/internal/request.go
+++ b/internal/request.go
@@ -46,17 +46,19 @@ const (
)
type SendMessageRequestHeader struct {
- ProducerGroup string
- Topic string
- QueueId int
- SysFlag int
- BornTimestamp int64
- Flag int32
- Properties string
- ReconsumeTimes int
- UnitMode bool
- MaxReconsumeTimes int
- Batch bool
+ ProducerGroup string
+ Topic string
+ QueueId int
+ SysFlag int
+ BornTimestamp int64
+ Flag int32
+ Properties string
+ ReconsumeTimes int
+ UnitMode bool
+ MaxReconsumeTimes int
+ Batch bool
+ DefaultTopic string
+ DefaultTopicQueueNums string
}
func (request *SendMessageRequestHeader) Encode() map[string]string {
@@ -88,6 +90,28 @@ type EndTransactionRequestHeader struct {
TransactionId string
}
+type SendMessageRequestV2Header struct {
+ *SendMessageRequestHeader
+}
+
+func (request *SendMessageRequestV2Header) Encode() map[string]string {
+ maps := make(map[string]string)
+ maps["a"] = request.ProducerGroup
+ maps["b"] = request.Topic
+ maps["c"] = request.DefaultTopic
+ maps["d"] = request.DefaultTopicQueueNums
+ maps["e"] = strconv.Itoa(request.QueueId)
+ maps["f"] = fmt.Sprintf("%d", request.SysFlag)
+ maps["g"] = strconv.FormatInt(request.BornTimestamp, 10)
+ maps["h"] = fmt.Sprintf("%d", request.Flag)
+ maps["i"] = request.Properties
+ maps["j"] = strconv.Itoa(request.ReconsumeTimes)
+ maps["k"] = strconv.FormatBool(request.UnitMode)
+ maps["l"] = strconv.Itoa(request.MaxReconsumeTimes)
+ maps["m"] = strconv.FormatBool(request.Batch)
+ return maps
+}
+
func (request *EndTransactionRequestHeader) Encode() map[string]string {
maps := make(map[string]string)
maps["producerGroup"] = request.ProducerGroup
diff --git a/primitive/message.go b/primitive/message.go
index 0608a61..8633dae 100644
--- a/primitive/message.go
+++ b/primitive/message.go
@@ -75,6 +75,12 @@ type Message struct {
mutex sync.RWMutex
}
+func (m *Message) WithProperties(p map[string]string) {
+ m.mutex.Lock()
+ m.properties = p
+ m.mutex.Unlock()
+}
+
func (m *Message) WithProperty(key, value string) {
if key == "" || value == "" {
return
@@ -199,6 +205,35 @@ func (m *Message) String() string {
m.Topic, string(m.Body), m.Flag, m.properties, m.TransactionId)
}
+func (m *Message) Marshal() []byte {
+ // storeSize all size of message info.
+ // TOTALSIZE MAGICCOD BODYCRC FLAG BODYSIZE BODY PROPERTYSIZE PROPERTY
+ v := m.MarshallProperties()
+ properties := []byte(v)
+ storeSize := 4 + 4 + 4 + 4 + 4 + len(m.Body) + 2 + len(properties)
+
+ buffer := make([]byte, storeSize)
+ pos := 0
+ binary.BigEndian.PutUint32(buffer[pos:], uint32(storeSize)) // 1. TOTALSIZE
+ pos += 4
+ binary.BigEndian.PutUint32(buffer[pos:], 0) // 2. MAGICCODE
+ pos += 4
+ binary.BigEndian.PutUint32(buffer[pos:], 0) // 3. BODYCRC
+ pos += 4
+ binary.BigEndian.PutUint32(buffer[pos:], uint32(m.Flag)) // 4. FLAG
+ pos += 4
+ binary.BigEndian.PutUint32(buffer[pos:], uint32(len(m.Body))) // 5. BODYSIZE
+ pos += 4
+ copy(buffer[pos:], m.Body)
+ pos += len(m.Body)
+
+ binary.BigEndian.PutUint16(buffer[pos:], uint16(len(properties))) // 7. PROPERTYSIZE
+ pos += 2
+ copy(buffer[pos:], properties)
+
+ return buffer
+}
+
type MessageExt struct {
Message
MsgId string
diff --git a/producer/producer.go b/producer/producer.go
index 287df58..013d3aa 100644
--- a/producer/producer.go
+++ b/producer/producer.go
@@ -18,6 +18,7 @@ limitations under the License.
package producer
import (
+ "bytes"
"context"
"fmt"
"strconv"
@@ -90,26 +91,58 @@ func (p *defaultProducer) Shutdown() error {
return nil
}
-func (p *defaultProducer) checkMsg(msg *primitive.Message) error {
+func (p *defaultProducer) checkMsg(msgs ...*primitive.Message) error {
if atomic.LoadInt32(&p.state) != int32(internal.StateRunning) {
return ErrNotRunning
}
- if msg == nil {
+ if len(msgs) == 0 {
return errors.New("message is nil")
}
- if msg.Topic == "" {
+ if len(msgs[0].Topic) == 0 {
return errors.New("topic is nil")
}
return nil
}
-func (p *defaultProducer) SendSync(ctx context.Context, msg *primitive.Message) (*primitive.SendResult, error) {
- if err := p.checkMsg(msg); err != nil {
+func (p *defaultProducer) encodeBatch(msgs ...*primitive.Message) *primitive.Message {
+ if len(msgs) == 1 {
+ return msgs[0]
+ }
+
+ // encode batch
+ batch := new(primitive.Message)
+ batch.Topic = msgs[0].Topic
+ batch.Queue = msgs[0].Queue
+ if len(msgs) > 1 {
+ batch.Body = MarshalMessageBatch(msgs...)
+ batch.Batch = true
+ } else {
+ batch.Body = msgs[0].Body
+ batch.Flag = msgs[0].Flag
+ batch.WithProperties(msgs[0].GetProperties())
+ batch.TransactionId = msgs[0].TransactionId
+ }
+ return batch
+}
+
+func MarshalMessageBatch(msgs ...*primitive.Message) []byte {
+ buffer := bytes.NewBufferString("")
+ for _, msg := range msgs {
+ data := msg.Marshal()
+ buffer.Write(data)
+ }
+ return buffer.Bytes()
+}
+
+func (p *defaultProducer) SendSync(ctx context.Context, msgs ...*primitive.Message) (*primitive.SendResult, error) {
+ if err := p.checkMsg(msgs...); err != nil {
return nil, err
}
+ msg := p.encodeBatch(msgs...)
+
resp := new(primitive.SendResult)
if p.interceptor != nil {
primitive.WithMethod(ctx, primitive.SendSync)
@@ -177,11 +210,13 @@ func (p *defaultProducer) sendSync(ctx context.Context, msg *primitive.Message,
return err
}
-func (p *defaultProducer) SendAsync(ctx context.Context, f func(context.Context, *primitive.SendResult, error), msg *primitive.Message) error {
- if err := p.checkMsg(msg); err != nil {
+func (p *defaultProducer) SendAsync(ctx context.Context, f func(context.Context, *primitive.SendResult, error), msgs ...*primitive.Message) error {
+ if err := p.checkMsg(msgs...); err != nil {
return err
}
+ msg := p.encodeBatch(msgs...)
+
if p.interceptor != nil {
primitive.WithMethod(ctx, primitive.SendAsync)
@@ -217,11 +252,13 @@ func (p *defaultProducer) sendAsync(ctx context.Context, msg *primitive.Message,
})
}
-func (p *defaultProducer) SendOneWay(ctx context.Context, msg *primitive.Message) error {
- if err := p.checkMsg(msg); err != nil {
+func (p *defaultProducer) SendOneWay(ctx context.Context, msgs ...*primitive.Message) error {
+ if err := p.checkMsg(msgs...); err != nil {
return err
}
+ msg := p.encodeBatch(msgs...)
+
if p.interceptor != nil {
primitive.WithMethod(ctx, primitive.SendOneway)
return p.interceptor(ctx, msg, nil, func(ctx context.Context, req, reply interface{}) error {
@@ -286,10 +323,16 @@ func (p *defaultProducer) buildSendRequest(mq *primitive.MessageQueue,
Properties: msg.MarshallProperties(),
ReconsumeTimes: 0,
UnitMode: p.options.UnitMode,
- Batch: false,
+ Batch: msg.Batch,
+ }
+ cmd := internal.ReqSendMessage
+ if msg.Batch {
+ cmd = internal.ReqSendBatchMessage
+ reqv2 := &internal.SendMessageRequestV2Header{SendMessageRequestHeader: req}
+ return remote.NewRemotingCommand(cmd, reqv2, msg.Body)
}
- return remote.NewRemotingCommand(internal.ReqSendMessage, req, msg.Body)
+ return remote.NewRemotingCommand(cmd, req, msg.Body)
}
func (p *defaultProducer) selectMessageQueue(msg *primitive.Message) *primitive.MessageQueue {