事务消息逻辑
首先我们来docker 部署rocketMQ与rocketMQDashBoard
docker ps查看rocketMQ 容器名称
docker ps
进入容器内部
docker exec -it rmqnamesrv /bin/bash创建事务消息
sh mqadmin updateTopic -c DefaultCluster -t TRANSACTIONTopic -n 127.0.0.1:9876 -a +message.type=TRANSACTION创建事务消息成功
在github rocktmq5.0 golang客户端已经支持事务消息
官方事务消息demo
https://github.com/apache/rocketmq-clients/blob/master/golang/example/producer/transaction/main.go
半事务触发时间限制
func TRANSACTIONTopic() {
// log to console
os.Setenv("mq.consoleAppender.enabled", "true")
golang.ResetLogger()
topic := "TRANSACTIONTopic"
producer, err := golang.NewProducer(&golang.Config{
Endpoint: Endpoint,
Credentials: &credentials.SessionCredentials{},
},
golang.WithTransactionChecker(&golang.TransactionChecker{
Check: CheckHalf, //检查half消息
}),
golang.WithTopics(topic),
)
if err != nil {
log.Fatal(err)
}
// start producer
err = producer.Start()
if err != nil {
log.Fatal(err)
}
// graceful stop producer
defer producer.GracefulStop()
msg := &golang.Message{Topic: topic, Body: []byte("this is a message TRANSACTION ")}
msg.SetKeys("a", "b")
msg.SetTag("ab")
transaction := producer.BeginTransaction()
resp, err := producer.SendWithTransaction(context.TODO(), msg, transaction)
if err != nil {
log.Fatal(err)
}
for i := 0; i < len(resp); i++ {
fmt.Printf("%#v\n", resp[i])
}
//commit transaction message
//TODO 本地事务执行成功
TODOLocalTransaction()
//提交事务
err = transaction.Commit()
if err != nil {
log.Fatal(err)
}
time.Sleep(time.Minute * 10)
}
// 检查本地事务
func CheckHalf(msg *golang.MessageView) golang.TransactionResolution {
log.Printf("check transaction message: %v", msg)
CheckLocalTransaction() //检查本地事务
return golang.COMMIT
}
func TODOLocalTransaction() bool {
return true
}
func CheckLocalTransaction() bool {
return true
}
第一步：producer.SendWithTransaction 发送半事务（发送成功）
第二步：TODO 本地事务执行成功 TODOLocalTransaction(执行本地事务成功)
第三步：当我们注释掉第三步commit代码，模拟提交事务异常（模拟commit事务消息失败）
第四步：rocketmq触发CheckHalf代码
check事务状态类型
UNKNOWN // 暂时无法判断状态，等待固定时间以后Broker端根据回查规则向生产者进行消息回查 COMMIT //提交事务，允许消费者消费该消息 ROLLBACK //回滚事务，消息将被丢弃不允许消费。
