Sending and Receiving Transactional Messages
DMS for RocketMQ ensures transaction consistency between the service logic and message transmission, and implements transaction support in two phases. Figure 1 illustrates the interaction of transactional messages.
The producer sends a half message and then executes the local transaction. If the execution is successful, the transaction is committed. If the execution fails, the transaction is rolled back. If the server does not receive any commit or rollback request after a period of time, it initiates a check. After receiving the check request, the producer resends a transaction commit or rollback request. The message is delivered to the consumer only after being committed. The consumer is unaware of the rollback.
Before sending and receiving transactional messages, collect RocketMQ connection information by referring to Collecting Connection Information.
Preparing the Environment
- Run the following command to check whether Go has been installed:
go version
If the following information is displayed, Go has been installed:
go version go1.16.5 linux/amd64
If Go is not installed, download and install it.
- Add the following code to go.mod to add the dependency:
module rocketmq-example-go go 1.13 require ( github.com/apache/rocketmq-client-go/v2 v2.1.1 )
Sending Transactional Messages
The following code is an example. Replace the information in bold with the actual values.
package main
import (
"context"
"fmt"
"os"
"strconv"
"sync"
"sync/atomic"
"time"
"github.com/apache/rocketmq-client-go/v2"
"github.com/apache/rocketmq-client-go/v2/primitive"
"github.com/apache/rocketmq-client-go/v2/producer"
)
type DemoListener struct {
localTrans *sync.Map
transactionIndex int32
}
func NewDemoListener() *DemoListener {
return &DemoListener{
localTrans: new(sync.Map),
}
}
func (dl *DemoListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
nextIndex := atomic.AddInt32(&dl.transactionIndex, 1)
fmt.Printf("nextIndex: %v for transactionID: %v\n", nextIndex, msg.TransactionId)
status := nextIndex % 3
dl.localTrans.Store(msg.TransactionId, primitive.LocalTransactionState(status+1))
fmt.Printf("dl")
return primitive.UnknowState
}
func (dl *DemoListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
fmt.Printf("%v msg transactionID : %v\n", time.Now(), msg.TransactionId)
v, existed := dl.localTrans.Load(msg.TransactionId)
if !existed {
fmt.Printf("unknow msg: %v, return Commit", msg)
return primitive.CommitMessageState
}
state := v.(primitive.LocalTransactionState)
switch state {
case 1:
fmt.Printf("checkLocalTransaction COMMIT_MESSAGE: %v\n", msg)
return primitive.CommitMessageState
case 2:
fmt.Printf("checkLocalTransaction ROLLBACK_MESSAGE: %v\n", msg)
return primitive.RollbackMessageState
case 3:
fmt.Printf("checkLocalTransaction unknow: %v\n", msg)
return primitive.UnknowState
default:
fmt.Printf("checkLocalTransaction default COMMIT_MESSAGE: %v\n", msg)
return primitive.CommitMessageState
}
}
func main() {
p, _ := rocketmq.NewTransactionProducer(
NewDemoListener(),
producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.0.1:8100"})),
producer.WithRetry(1),
)
err := p.Start()
if err != nil {
fmt.Printf("start producer error: %s\n", err.Error())
os.Exit(1)
}
for i := 0; i < 10; i++ {
res, err := p.SendMessageInTransaction(context.Background(),
primitive.NewMessage("topic1", []byte("Hello RocketMQ again "+strconv.Itoa(i))))
if err != nil {
fmt.Printf("send message error: %s\n", err)
} else {
fmt.Printf("send message success: result=%s\n", res.String())
}
}
time.Sleep(5 * time.Minute)
err = p.Shutdown()
if err != nil {
fmt.Printf("shutdown producer error: %s", err.Error())
}
}
The parameters in the example code are described as follows. For details about how to obtain the parameter values, see Collecting Connection Information.
- 192.168.0.1:8100: instance metadata address and port.
- topic1: topic name.
The producer needs to implement two callback functions. The ExecuteLocalTransaction callback function is called after the half message is sent (see step 3 in the diagram). The CheckLocalTransaction callback function is called after the check request is received (see step 6 in the diagram). The two callback functions can return three transaction states:
- primitive.CommitMessageState: Transaction committed. The consumer can retrieve the message.
- primitive.RollbackMessageState: Transaction rolled back. The message will be discarded and cannot be retrieved.
- primitive.UnknowState: The status cannot be determined and the server is expected to check the message status from the producer again.
Subscribing to Transactional Messages
The code for subscribing to transactional messages is the same as that for subscribing to normal messages.
Feedback
Was this page helpful?
Provide feedbackThank you very much for your feedback. We will continue working to improve the documentation.
