基于 TCC 写一个所谓的分布式事务#2
紧接上篇，实现 TCC 的使用方。
实现 TCC Component
首先需要我们需要定义一些状态，比如事务的状态，业务方的状态（此处只是作为模拟，不涉及具体逻辑），并定义最基本的 Component 结构。
状态定义
// tcc 组件侧记录的一笔事务的状态
type TXStatus string
func (t TXStatus) String() string {
return string(t)
}
const (
TXTried TXStatus = "tried" // 已执行 try 操作
TXConfirmed TXStatus = "confirmed" // 已执行 confirm 操作
TXCanceled TXStatus = "canceled" // 已执行 cancel 操作
)
// 一笔事务对应数据的状态
type DataStatus string
func (d DataStatus) String() string {
return string(d)
}
const (
DataFrozen DataStatus = "frozen" // 冻结态
DataSuccessful DataStatus = "successful" // 成功态
)
结构定义
首先实现一个简单的 TCC Component 结构，里面应该包含一个 id，还有一个 client，这里的 client 其实是一个 Redis Client，主要用于数据存储和分布式锁，locker 则是我们上一篇基于 Redis 实现的分布式锁。
type MockComponent struct {
id string
client redislib.UniversalClient
locker dlock.Locker
}
func NewMockComponent(id string, client redislib.UniversalClient, locker dlock.Locker) *MockComponent {
return &MockComponent{
id: id,
client: client,
locker: locker,
}
}
下面按照顺序来分别实现 TCC Component 的基本方法。
Try
Try 方法的流程之前已经介绍过，事务方调用 Component Try 时，首先会添加分布式锁，然后进行幂等校验，将资源冻结（等到后续 Confirm 后才最终落库提交），然后记录事务的状态。
func (m *MockComponent) Try(ctx context.Context, req *gotcc.TCCReq) (*gotcc.TCCResp, error) {
// 基于 txID 维度加锁，这里不涉及耗时操作，直接继承 locker 的配置即可
mutex := m.locker.NewMutex(pkg.BuildTXLockKey(m.id, req.TXID))
if err := mutex.Lock(ctx); err != nil {
return nil, err
}
defer func() {
unlockCtx, unlockCancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second)
defer unlockCancel()
_ = mutex.Unlock(unlockCtx)
}()
// 基于 txID 幂等性去重，注意这里需要特殊排除一下键不存在的情况
txStatus, err := m.client.Get(ctx, pkg.BuildTXKey(m.id, req.TXID)).Result()
if err != nil && !errors.Is(err, redislib.Nil) {
return nil, err
}
res := &gotcc.TCCResp{
ComponentID: m.id,
TXID: req.TXID,
}
switch txStatus {
// 重复的 try 请求，给予成功的响应
case TXTried.String(), TXConfirmed.String():
res.ACK = true
return res, nil
// 先 cancel，后收到 try 请求，拒绝
case TXCanceled.String():
return res, nil
default:
}
// 执行 try 操作，将数据状态置为 frozen
bizID := gocast.ToString(req.Data["biz_id"])
// 存储 bizID 和事务的关系，Confirm/Cancel 后删除，这里没有设置 TTL
if err := m.client.SetNX(ctx, pkg.BuildTXDetailKey(m.id, req.TXID), bizID, 0).Err(); err != nil {
return nil, err
}
// 要求必须从零到一把 bizID 对应的数据置为冻结态
created, err := m.client.SetNX(ctx, pkg.BuildDataKey(m.id, req.TXID, bizID), DataFrozen.String(), 0).Result()
if err != nil {
return nil, err
}
if !created {
return res, nil
}
// 更新事务状态至 Tried
if err := m.client.Set(ctx, pkg.BuildTXKey(m.id, req.TXID), TXTried.String(), 0).Err(); err != nil {
return nil, fmt.Errorf("set transaction tried, txid=%s: %w", req.TXID, err)
}
// try 请求执行成功
res.ACK = true
// TEST ComponentB Try Failed
// if strings.Contains(req.ComponentID, "componentB") {
// res.ACK = false
// }
return res, nil
}
Confirm
Confirm 则是将冻结的业务数据推向终态，本方法由 TXManager 在所有 Components Try 均返回成功响应后调用。
func (m *MockComponent) Confirm(ctx context.Context, txID string) (*gotcc.TCCResp, error) {
// 基于 txID 维度加锁
mutex := m.locker.NewMutex(pkg.BuildTXLockKey(m.id, txID))
if err := mutex.Lock(ctx); err != nil {
return nil, err
}
defer func() {
unlockCtx, unlockCancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second)
defer unlockCancel()
_ = mutex.Unlock(unlockCtx)
}()
// 要求 txID 此前状态为 tried
// 如果该键不存在也直接抛出错误
txStatus, err := m.client.Get(ctx, pkg.BuildTXKey(m.id, txID)).Result()
if err != nil {
return nil, err
}
res := &gotcc.TCCResp{
ComponentID: m.id,
TXID: txID,
}
switch txStatus {
// 已 confirm，直接幂等响应为成功
case TXConfirmed.String():
res.ACK = true
return res, nil
// 只有状态为 try 放行
case TXTried.String():
// 其他情况直接拒绝
default:
return res, nil
}
// 获取事务对应的 bizID
bizID, err := m.client.Get(ctx, pkg.BuildTXDetailKey(m.id, txID)).Result()
if err != nil {
return nil, err
}
// 要求对应的数据状态此前为 frozen
dataStatus, err := m.client.Get(ctx, pkg.BuildDataKey(m.id, txID, bizID)).Result()
if err != nil {
return nil, err
}
switch dataStatus {
case DataFrozen.String():
// 把对应数据处理状态置为 successful
if err = m.client.Set(ctx, pkg.BuildDataKey(m.id, txID, bizID), DataSuccessful.String(), 0).Err(); err != nil {
return nil, err
}
// 数据已经推进状态了，只需推进 TX 状态即可
case DataSuccessful.String():
// 状态非法
default:
return res, nil
}
// 把事务状态更新为成功
if err := m.client.Set(
ctx,
pkg.BuildTXKey(m.id, txID),
TXConfirmed.String(),
0,
).Err(); err != nil {
return nil, fmt.Errorf("set transaction confirmed, txid=%s: %w", txID, err)
}
// 处理成功，给予成功的响应
res.ACK = true
return res, nil
}
Cancel
和 Confirm 类似，这里是处理 TCC 事务失败的链路。在实际业务实现中这里应该对冻结的数据操作进行回滚。
func (m *MockComponent) Cancel(ctx context.Context, txID string) (*gotcc.TCCResp, error) {
// 基于 txID 维度加锁
mutex := m.locker.NewMutex(pkg.BuildTXLockKey(m.id, txID))
if err := mutex.Lock(ctx); err != nil {
return nil, err
}
defer func() {
unlockCtx, unlockCancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second)
defer unlockCancel()
_ = mutex.Unlock(unlockCtx)
}()
tccResp := &gotcc.TCCResp{
ACK: true,
ComponentID: m.id,
TXID: txID,
}
// 查看事务的状态，只要不是 confirmed，就无脑置为 canceled
// 其次事务如果根本不存在，也应该抛出错误
txStatus, err := m.client.Get(ctx, pkg.BuildTXKey(m.id, txID)).Result()
if err != nil {
return nil, err
}
switch txStatus {
// 非法流转链路
case TXConfirmed.String():
return nil, fmt.Errorf("invalid tx status: %s, txid: %s", txStatus, txID)
// 直接幂等返回
case TXCanceled.String():
return tccResp, nil
// 进入正常 Cancel 链路
case TXTried.String():
default:
return nil, fmt.Errorf(
"unexpected tx status: %s, txid: %s",
txStatus,
txID,
)
}
// 根据事务获取对应的 bizID
bizID, err := m.client.Get(ctx, pkg.BuildTXDetailKey(m.id, txID)).Result()
if err != nil {
return nil, err
}
// 删除对应的 frozen 冻结记录
if err = m.client.Del(ctx, pkg.BuildDataKey(m.id, txID, bizID)).Err(); err != nil {
return nil, err
}
// 把事务状态更新为 canceled
if err := m.client.Set(
ctx,
pkg.BuildTXKey(m.id, txID),
TXCanceled.String(),
0,
).Err(); err != nil {
return nil, fmt.Errorf("set transaction canceled, txid=%s: %w", txID, err)
}
return tccResp, nil
}
想必大伙儿已经发现，在 Try/Confirm/Cancel 方法中，同样维护了一份事务状态，即：
const (
TXTried TXStatus = "tried" // 已执行 try 操作
TXConfirmed TXStatus = "confirmed" // 已执行 confirm 操作
TXCanceled TXStatus = "canceled" // 已执行 cancel 操作
)
看过之前 TCC 框架原理应该清楚，在 TXStore 中也维护了一份事务状态，那这个事务状态和 Component 中的有何不同？其实就是视角不同。
TXStore 中维护的是全局视角下的事务状态，即由多个 Components 组成的分布式事务状态，而 Component 中维护的仅仅是当前实例下的事务状态（单机事务）。
实现 TXStore
数据库及 DAO
这里我们简单设计一下数据库结构及其 DAO 实现，只需要记录分布式事务状态和各个 Components Try 状态即可，为了简单，Components Try 状态直接采用 JSON 存储：
CREATE TABLE IF NOT EXISTS `tx_record`
(
`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键ID',
`status` varchar(16) NOT NULL COMMENT '事务状态 hanging/successful/failure',
`component_try_statuses` json DEFAULT NULL COMMENT '各组件 try 接口请求状态 hanging/successful/failure',
`deleted_at` datetime DEFAULT NULL COMMENT '删除时间',
`created_at` datetime NOT NULL COMMENT '创建时间',
`updated_at` datetime DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (`id`) USING BTREE COMMENT '主键索引',
KEY `idx_status` (`status`) COMMENT '事务状态索引'
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT '事务日志记录';
type QueryOption func(db *gorm.DB) *gorm.DB
func WithID(id uint) QueryOption {
return func(db *gorm.DB) *gorm.DB {
return db.Where("id = ?", id)
}
}
func WithStatus(status gotcc.TXStatus) QueryOption {
return func(db *gorm.DB) *gorm.DB {
return db.Where("status = ?", status.String())
}
}
type TXRecordPO struct {
gorm.Model
Status string `gorm:"status"`
ComponentTryStatuses string `gorm:"component_try_statuses"`
}
func (t TXRecordPO) TableName() string {
return "tx_record"
}
type ComponentTryStatus struct {
ComponentID string `json:"componentID"`
TryStatus string `json:"tryStatus"`
}
type TXRecordDAO struct {
db *gorm.DB
}
func NewTXRecordDAO(db *gorm.DB) *TXRecordDAO {
return &TXRecordDAO{
db: db,
}
}
func (t *TXRecordDAO) GetTXRecords(ctx context.Context, opts ...QueryOption) ([]*TXRecordPO, error) {
db := t.db.WithContext(ctx).Model(&TXRecordPO{})
for _, opt := range opts {
db = opt(db)
}
var records []*TXRecordPO
return records, db.Scan(&records).Error
}
func (t *TXRecordDAO) CreateTXRecord(ctx context.Context, record *TXRecordPO) (uint, error) {
return record.ID, t.db.WithContext(ctx).Model(&TXRecordPO{}).Create(record).Error
}
// UpdateComponentStatus 更新事务中的 Component 状态
func (t *TXRecordDAO) UpdateComponentStatus(ctx context.Context, id uint, componentID string, status string) error {
return t.LockAndDo(ctx, id, func(ctx context.Context, dao *TXRecordDAO, record *TXRecordPO) error {
var statuses map[string]*ComponentTryStatus
if err := json.Unmarshal([]byte(record.ComponentTryStatuses), &statuses); err != nil {
return err
}
componentStatus, ok := statuses[componentID]
if !ok {
return fmt.Errorf("invalid component: %s in txid: %d", componentID, id)
}
if componentStatus.TryStatus == status {
return nil
}
if componentStatus.TryStatus == gotcc.TryHanging.String() {
componentStatus.TryStatus = status
body, _ := json.Marshal(statuses)
record.ComponentTryStatuses = string(body)
return dao.UpdateTXRecord(ctx, record)
}
return fmt.Errorf("invalid status: %s of component: %s, txid: %d", statuses[componentID].TryStatus, componentID, id)
})
}
func (t *TXRecordDAO) UpdateTXRecord(ctx context.Context, record *TXRecordPO) error {
return t.db.WithContext(ctx).Updates(record).Error
}
// LockAndDo 加锁拿到对应 id 的事务记录，然后对其执行 do 函数
func (t *TXRecordDAO) LockAndDo(ctx context.Context, id uint, do func(ctx context.Context, dao *TXRecordDAO, record *TXRecordPO) error) error {
return t.db.Transaction(func(tx *gorm.DB) error {
// 加写锁
var record TXRecordPO
if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).First(&record, id).Error; err != nil {
return err
}
txDAO := NewTXRecordDAO(tx)
return do(ctx, txDAO, &record)
})
}
结构定义
TXStore 采用关系型数据库的方式记录事务状态，所以这里我们将之前定义的 DAO 写入结构体，此外同样需要分布式锁 locker，这里的分布式锁仅仅是为了避免 TXManager 的全局轮询被多实例同时执行：
type TXRecordDAO interface {
GetTXRecords(ctx context.Context, opts ...expdao.QueryOption) ([]*expdao.TXRecordPO, error)
CreateTXRecord(ctx context.Context, record *expdao.TXRecordPO) (uint, error)
UpdateComponentStatus(ctx context.Context, id uint, componentID string, status string) error
UpdateTXRecord(ctx context.Context, record *expdao.TXRecordPO) error
LockAndDo(ctx context.Context, id uint, do func(ctx context.Context, dao *expdao.TXRecordDAO, record *expdao.TXRecordPO) error) error
}
type MockTXStore struct {
locker dlock.Locker
mutex dlock.Mutex
dao TXRecordDAO
}
func NewMockTXStore(dao TXRecordDAO, locker dlock.Locker) *MockTXStore {
return &MockTXStore{
dao: dao,
locker: locker,
}
}
事务操作
// CreateTX 创建一条事务，并注入 TCC Components
func (m *MockTXStore) CreateTX(ctx context.Context, components ...gotcc.TCCComponent) (string, error) {
// 创建一项内容，里面以唯一事务 id 为 key
componentTryStatuses := make(map[string]*expdao.ComponentTryStatus, len(components))
for _, component := range components {
componentTryStatuses[component.ID()] = &expdao.ComponentTryStatus{
ComponentID: component.ID(),
TryStatus: gotcc.TryHanging.String(),
}
}
// 记录每个 Component 的处理状态
statusesBody, _ := json.Marshal(componentTryStatuses)
txID, err := m.dao.CreateTXRecord(ctx, &expdao.TXRecordPO{
Status: gotcc.TXHanging.String(),
ComponentTryStatuses: string(statusesBody),
})
if err != nil {
return "", err
}
return gocast.ToString(txID), nil
}
// TXUpdate 在事务日志中更新 Component 状态
func (m *MockTXStore) TXUpdate(ctx context.Context, txID string, componentID string, accept bool) error {
_txID := gocast.ToUint(txID)
status := gotcc.TXFailure.String()
if accept {
status = gotcc.TXSuccessful.String()
}
return m.dao.UpdateComponentStatus(ctx, _txID, componentID, status)
}
// GetHangingTXs 轮询用，获取还在 hanging 状态的事务
func (m *MockTXStore) GetHangingTXs(ctx context.Context) ([]*gotcc.Transaction, error) {
records, err := m.dao.GetTXRecords(ctx, expdao.WithStatus(gotcc.TXHanging))
if err != nil {
return nil, err
}
txs := make([]*gotcc.Transaction, 0, len(records))
for _, record := range records {
componentTryStatuses := make(map[string]*expdao.ComponentTryStatus)
_ = json.Unmarshal([]byte(record.ComponentTryStatuses), &componentTryStatuses)
components := make([]*gotcc.ComponentTryEntity, 0, len(componentTryStatuses))
for _, component := range componentTryStatuses {
components = append(components, &gotcc.ComponentTryEntity{
ComponentID: component.ComponentID,
TryStatus: gotcc.ComponentTryStatus(component.TryStatus),
})
}
txs = append(txs, &gotcc.Transaction{
TXID: gocast.ToString(record.ID),
Status: gotcc.TXHanging,
CreatedAt: record.CreatedAt,
Components: components,
})
}
return txs, nil
}
// TXSubmit 提交事务的最终状态
func (m *MockTXStore) TXSubmit(ctx context.Context, txID string, success bool) error {
do := func(ctx context.Context, dao *expdao.TXRecordDAO, record *expdao.TXRecordPO) error {
if success {
if record.Status == gotcc.TXFailure.String() {
return fmt.Errorf("invalid tx status: %s, txid: %s", record.Status, txID)
}
record.Status = gotcc.TXSuccessful.String()
} else {
if record.Status == gotcc.TXSuccessful.String() {
return fmt.Errorf("invalid tx status: %s, txid: %s", record.Status, txID)
}
record.Status = gotcc.TXFailure.String()
}
return dao.UpdateTXRecord(ctx, record)
}
return m.dao.LockAndDo(ctx, gocast.ToUint(txID), do)
}
// GetTX 获取指定的一笔事务
func (m *MockTXStore) GetTX(ctx context.Context, txID string) (*gotcc.Transaction, error) {
records, err := m.dao.GetTXRecords(ctx, expdao.WithID(gocast.ToUint(txID)))
if err != nil {
return nil, err
}
if len(records) != 1 {
return nil, errors.New("get tx failed")
}
componentTryStatuses := make(map[string]*expdao.ComponentTryStatus)
_ = json.Unmarshal([]byte(records[0].ComponentTryStatuses), &componentTryStatuses)
components := make([]*gotcc.ComponentTryEntity, 0, len(componentTryStatuses))
for _, tryItem := range componentTryStatuses {
components = append(components, &gotcc.ComponentTryEntity{
ComponentID: tryItem.ComponentID,
TryStatus: gotcc.ComponentTryStatus(tryItem.TryStatus),
})
}
return &gotcc.Transaction{
TXID: txID,
Status: gotcc.TXStatus(records[0].Status),
Components: components,
CreatedAt: records[0].CreatedAt,
}, nil
}
实现全局锁
TXStore 全局分布式锁在 TXManager 轮询中使用。
func (m *MockTXStore) Lock(ctx context.Context, expireDuration time.Duration) error {
m.mutex = m.locker.NewMutex(pkg.BuildTXRecordLockKey(), dlock.WithExpiry(expireDuration))
return m.mutex.Lock(ctx)
}
func (m *MockTXStore) Unlock(ctx context.Context) error {
if m.mutex == nil {
return errors.New("tx store mutex is not locked")
}
return m.mutex.Unlock(ctx)
}
测试运行
下列测试所有 Components Try 都会通过，如果需要测试失败的情况，可以修改一下 Try 方法，使得其中某个 Component Try 执行失败。
const (
dsn = "请输入你的 mysql dsn"
address = "请输入你的 redis ip"
password = "请输入你的 redis 密码"
)
func main() {
redisClient := pkg.NewRedisClient(address, password)
locker := dlock.NewRedisLocker(redisClient, dlock.DefaultConfig())
mysqlDB, err := pkg.NewDB(dsn)
if err != nil {
fmt.Println(err)
return
}
componentAID := "componentA"
componentBID := "componentB"
componentCID := "componentC"
// 构造出对应的 tcc component
componentA := NewMockComponent(componentAID, redisClient, locker)
componentB := NewMockComponent(componentBID, redisClient, locker)
componentC := NewMockComponent(componentCID, redisClient, locker)
// 构造出事务日志存储模块
txRecordDAO := dao.NewTXRecordDAO(mysqlDB)
txStore := NewMockTXStore(txRecordDAO, locker)
txManager := gotcc.NewTXManager(txStore, gotcc.WithMonitorTick(time.Second))
defer txManager.Stop()
// 完成各组件的注册
if err := txManager.Register(componentA); err != nil {
fmt.Println(err)
return
}
if err := txManager.Register(componentB); err != nil {
fmt.Println(err)
return
}
if err := txManager.Register(componentC); err != nil {
fmt.Println(err)
return
}
ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
defer cancel()
_, success, err := txManager.Transaction(ctx, []*gotcc.RequestEntity{
{ComponentID: componentAID,
Request: map[string]interface{}{
"biz_id": componentAID + "_biz",
},
},
{ComponentID: componentBID,
Request: map[string]interface{}{
"biz_id": componentBID + "_biz",
},
},
{ComponentID: componentCID,
Request: map[string]interface{}{
"biz_id": componentCID + "_biz",
},
},
}...)
if err != nil {
fmt.Printf("tx failed, err: %v", err)
return
}
if !success {
fmt.Println("tx failed")
}
// 这里等待一分钟，观察 TxManager 异步轮询 Hanging Transaction
<-time.After(time.Minute)
fmt.Println("success")
}
结束！垮台！
评论区