# 收发事务消息
分布式消息服务RocketMQ版的事务消息支持在业务逻辑与发送消息之间提供事务保证，通过两阶段的方式提供对事务消息的支持，事务消息交互流程如[图1]所示。
图1事务消息交互流程   
![](https://support.huaweicloud.com/devg-hrm/zh-cn_image_0000001129284708.png "点击放大")
事务消息生产者首先发送半消息，然后执行本地事务。如果执行成功，则发送事务提交，否则发送事务回滚。服务端在一段时间后如果一直收不到提交或回滚，则发起回查，生产者在收到回查后重新发送事务提交或回滚。消息只有在提交之后才投递给消费者，消费者对回滚的消息不可见。
收发事务消息前，请参考[收集连接信息](https://support.huaweicloud.com/devg-hrm/hrm-devg-008.html)收集RocketMQ所需的连接信息。
#### 准备环境
1. 执行以下命令，检查是否已安装Go。
   ```
   go version
   ```
   返回如下回显时，说明Go已经安装。
   ```
   go version go1.16.5 linux/amd64
   ```
   如果未安装Go，请[下载并安装](https://go.dev/dl/go1.16.5.linux-amd64.tar.gz)。
   
2. 进入Go脚本所在的bin目录下。
3. 执行"touch go.mod"命令新建一个"go.mod"，并增加以下代码，添加依赖。
   ```
   module rocketmq-example-go
   go 1.13
   require (
   github.com/apache/rocketmq-client-go/v2 v2.1.2
   )
   ```
   
4. 执行如下命令增加代理。
   ```
   export GOPROXY=https://goproxy.cn,direct
   ```
   
5. 执行如下命令下载依赖。
   ```
   go mod tidy
   ```
   
 
#### 发送事务消息
参考如下示例代码（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。
```
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),
//producer.WithTls(true), //创建实例时，如果开启了SSL，请添加此行代码。
)
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())
}
}
```
示例代码中的参数说明如下，请参考[收集连接信息](https://support.huaweicloud.com/devg-hrm/hrm-devg-008.html)获取参数值。
- 192.168.0.1:8100：表示实例连接地址和端口。
- topic1：表示Topic名称。
事务消息生产者需要实现两个回调函数，其中ExecuteLocalTransaction回调函数在发送完半事务消息后被调用，即上图中的第3阶段，CheckLocalTransaction回调函数在收到回查时调用，即上图中的第6阶段。两个回调函数均可返回3种事务状态：
- primitive.CommitMessageState：提交事务，允许消费者消费该消息。
- primitive.RollbackMessageState：回滚事务，消息将被丢弃不允许消费。
- primitive.UnknowState：无法判断状态，期待服务端向生产者再次回查该消息的状态。
#### 订阅事务消息
订阅事务消息的代码与[订阅普通消息的代码](https://support.huaweicloud.com/devg-hrm/hrm-devg-013.html#hrm-devg-013__section77006348111)相同。