海量并发下如何回答分布式事务一致性（对话式 + 脑图）

package main

import "fmt"

// Participant 模拟一个资源管理器（如一个数据库分库）
type Participant struct {
	name       string
	canCommit  bool // 模拟该参与者是否能做好提交准备（资源能否锁定）
}

// prepare 阶段一：协调者询问参与者是否可以提交
// 真实场景里这里会写 undo/redo 日志并对数据行加锁
func (p *Participant) prepare() bool {
	fmt.Printf("[准备] %s：记录 undo/redo 日志，返回 %v\n", p.name, p.canCommit)
	return p.canCommit
}

func (p *Participant) commit() {
	fmt.Printf("[提交] %s：正式写入数据并释放锁\n", p.name)
}

func (p *Participant) rollback() {
	fmt.Printf("[回滚] %s：利用 undo 日志回滚并释放锁\n", p.name)
}

func main() {
	participants := []*Participant{
		{name: "订单库(数据库一)", canCommit: true},
		{name: "商品库(数据库二)", canCommit: true},
		{name: "促销库(数据库三)", canCommit: false}, // 模拟库存资源锁定失败
	}

	// 阶段一：准备，收集所有投票后再决策
	allReady := true
	for _, p := range participants {
		if !p.prepare() {
			allReady = false
		}
	}

	// 阶段二：全 YES 则提交，否则中段回滚（投 YES 的也要跟着回滚）
	if allReady {
		for _, p := range participants {
			p.commit()
		}
		fmt.Println("✅ 全局事务提交成功")
	} else {
		for _, p := range participants {
			p.rollback()
		}
		fmt.Println("❌ 全局事务回滚（至少一个参与者未就绪）")
	}
}

// 订阅时务必关闭自动应答
msgs, _ := ch.Consume(queue, consumer, /* autoAck = */ false, false, false, false, nil)

for d := range msgs {
	// 1) 先执行业务：扣减优惠券
	err := deductCoupon(context.Background(), d.Body)
	if err != nil {
		// 2) 业务失败：不 Ack，MQ 会在重试策略下重新投递
		//    超过最大重试次数会进死信队列，等待人工干预
		_ = d.Nack(false, true) // multiple=false, requeue=true
		continue
	}
	// 3) 业务成功“之后”才手动 Ack，MQ 才真正删除消息
	if err := d.Ack(false); err != nil {
		// Ack 本身失败也要记录告警，避免消息静默丢失
		log.Printf("ack failed: %v", err)
	}
}

package main

import (
	"fmt"
	"sync"
	"time"
)

// ---- 本地消息表 ----

type MsgStatus string

const (
	StatusPending MsgStatus = "PENDING" // 待发送
	StatusSent    MsgStatus = "SENT"    // 已投递
	StatusDone    MsgStatus = "DONE"    // 已完成（消费者已确认）
)

type LocalMessage struct {
	ID      string
	BizKey  string // 业务键，如 order_id
	Payload string
	Status  MsgStatus
}

// OrderDB 扮演订单系统侧的“本地消息表”
type OrderDB struct {
	mu       sync.Mutex
	messages map[string]*LocalMessage
}

func NewOrderDB() *OrderDB { return &OrderDB{messages: make(map[string]*LocalMessage)} }

// 步骤1：下单时，业务与消息在同一个本地事务里落库
func (db *OrderDB) createOrderWithMsg(orderID, payload string) {
	db.mu.Lock()
	defer db.mu.Unlock()
	db.messages[orderID] = &LocalMessage{
		ID: orderID, BizKey: orderID, Payload: payload, Status: StatusPending,
	}
	fmt.Printf("[订单系统] 本地事务落库：order=%s, 状态=%s\n", orderID, StatusPending)
}

// 步骤2：投递消息到 MQ（此处用打印模拟）
func (db *OrderDB) deliver(orderID string) {
	db.mu.Lock()
	m := db.messages[orderID]
	if m != nil && (m.Status == StatusPending || m.Status == StatusSent) {
		m.Status = StatusSent
	}
	db.mu.Unlock()
	if m != nil {
		fmt.Printf("[订单系统] 投递 MQ：order=%s (状态:%s)\n", orderID, m.Status)
	}
}

// 步骤4：收到消费者确认，标记完成
func (db *OrderDB) confirm(orderID string) {
	db.mu.Lock()
	defer db.mu.Unlock()
	if m, ok := db.messages[orderID]; ok {
		m.Status = StatusDone
		fmt.Printf("[订单系统] 收到消费确认，标记完成：order=%s\n", orderID)
	}
}

// 步骤5：定时任务扫描未完成消息，重新投递（补偿）
func (db *OrderDB) resendPending() {
	db.mu.Lock()
	var pending []string
	for id, m := range db.messages {
		if m.Status == StatusPending || m.Status == StatusSent {
			pending = append(pending, id)
		}
	}
	db.mu.Unlock()
	for _, id := range pending {
		fmt.Printf("[补偿任务] 发现未完成消息，重新投递：order=%s\n", id)
		db.deliver(id)
	}
}

// ---- 优惠券系统侧 ----

// consume 消费消息，成功后回发确认（手动 Ack 思想）
func consume(mq chan string, db *OrderDB, wg *sync.WaitGroup) {
	defer wg.Done()
	for orderID := range mq {
		// 模拟扣减优惠券成功
		fmt.Printf("[优惠券系统] 扣减优惠券成功：order=%s\n", orderID)
		// 关键：业务成功之后才确认（对应 MQ 的手动 Ack）
		db.confirm(orderID)
	}
}

func main() {
	db := NewOrderDB()
	mq := make(chan string, 10)
	var wg sync.WaitGroup
	wg.Add(1)
	go consume(mq, db, &wg)

	// 正常下单：O1001 / O1002 消息顺利送达优惠券系统
	for _, o := range []string{"O1001", "O1002"} {
		db.createOrderWithMsg(o, "deduct-coupon")
		db.deliver(o)
		mq <- o
	}

	// 模拟 MQ 网络异常：O1003 的扣券消息在投递时丢失（未进 MQ）
	db.createOrderWithMsg("O1003", "deduct-coupon")
	// 注意：这里没有 db.deliver("O1003")，也没有 mq <- "O1003"

	close(mq)
	wg.Wait()

	// 补偿任务：扫描本地消息表中未完成的消息，重新投递
	db.resendPending()

	time.Sleep(100 * time.Millisecond)
}

CREATE TABLE local_message (
    id          BIGSERIAL PRIMARY KEY,
    biz_key     VARCHAR(64)  NOT NULL,
    payload     TEXT         NOT NULL,
    status      VARCHAR(16)  NOT NULL DEFAULT 'PENDING', -- PENDING / SENT / DONE
    created_at  TIMESTAMP    NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMP    NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_status ON local_message (status, created_at); -- 补偿扫描靠它