Exactly Once和事务消息

package kafka

import (
	"fmt"

	"github.com/Shopify/sarama"
)

var (
	version = ""
)

// NewTxnProducer 新增事务Producer
func NewTxnProducer(brokers []string) (*sarama.AsyncProducer, error) {
	version, err := sarama.ParseKafkaVersion(version)
	if err != nil {
		fmt.Printf("Error parsing Kafka version: %v", err)
		return nil, err
	}

	config := sarama.NewConfig()
	config.Version = version
	// 幂等
	config.Producer.Idempotent = true
	config.Producer.Return.Errors = false
	// 等待所有副本提交成功
	config.Producer.RequiredAcks = sarama.WaitForAll
	// 选择要发送消息的分区，默认为散列
	config.Producer.Partitioner = sarama.NewRoundRobinPartitioner
	// 每次重试之间等待集群稳定的时间
	config.Producer.Transaction.Retry.Backoff = 10
	// 事务中用于识别生产者的实例
	config.Producer.Transaction.ID = "txn_producer"
	// 一个连接在发送阻塞之前,允许有多少个未完成的请求(默认为5)。
	config.Net.MaxOpenRequests = 1

	producer, err := sarama.NewAsyncProducer(brokers, config)
	if err != nil {
		fmt.Printf("NewAsyncProducer err: %v", err)
		return nil, err
	}

	return &producer, nil
}

// SendMessage 发送消息
func SendMessage(producer sarama.AsyncProducer, msg string) error {

	// 开始事务
	err := producer.BeginTxn()
	if err != nil {
		fmt.Printf("BeginTxn err: %v", err)
		return err
	}

	// 发送消息
	producerMessage := &sarama.ProducerMessage{
		Topic: "topic",
		Key:   nil,
		Value: sarama.StringEncoder(msg),
	}
	producer.Input() <- producerMessage

	// 提交事务
	err = producer.CommitTxn()
	if err != nil {
		fmt.Printf("CommitTxn err: %v,msg: %s", err, msg)

		if producer.TxnStatus()&sarama.ProducerTxnFlagFatalError != 0 {
			fmt.Printf("need to recreate producer")
			return err
		}

		if producer.TxnStatus()&sarama.ProducerTxnFlagAbortableError != 0 {
			err = producer.AbortTxn()
			if err != nil { // retry
				fmt.Printf("Producer: unable to abort transaction: %+v", err)
				return err
			}

		}

		// 再次尝试提交
		err = producer.CommitTxn()
		if err != nil {
			fmt.Printf("CommitTxn err: %v,msg: %s", err, msg)
			return err
		}
	}

	return nil
}

package rocketmq

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	rocketmq "github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
)

// DemoListener 测试demo使用
type DemoListener struct {
	localTrans       *sync.Map
	transactionIndex int32
}

// NewDemoListener 创建事务检查对象
func NewDemoListener() *DemoListener {
	return &DemoListener{
		localTrans: new(sync.Map),
	}
}

// CheckLocalTransaction 检查本地事务
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
	}
}

// ExecuteLocalTransaction 执行本地事务逻辑
// msg: 事务消息
// return: 本地事务状态
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
}

// NewRocketmqTxnProducer 新增事务Producer
func NewRocketmqTxnProducer() (rocketmq.TransactionProducer, error) {
	producer, err := rocketmq.NewTransactionProducer(
		NewDemoListener(),
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{"127.0.0.1:9876"})),
		producer.WithRetry(1),
	)
	if err != nil {
		fmt.Printf("NewTransactionProducer err: %v", err)
		return nil, err
	}
	return producer, nil
}

// SendRocketmqMessage 发送消息
func SendRocketmqMessage(producer rocketmq.TransactionProducer, topic, msg string) error {
	result, err := producer.SendMessageInTransaction(
		context.Background(), primitive.NewMessage(topic, []byte(msg)))
	if err != nil {
		fmt.Printf("SendMessageInTransaction err: %v,result: %s", err, result.String())
		return err
	}

	err = producer.Shutdown()
	if err != nil {
		fmt.Printf("Shutdown err: %v", err)
		return err
	}

	return nil
}