手撸golang GO与微服务 Saga模式之6

老罗话编程 · · 402 次点击 · · 开始浏览

这是一个创建于的文章，其中的信息可能已经有所发展或是发生改变。

缘起

最近阅读<<Go微服务实战>> (刘金亮, 2021.1)
本系列笔记拟采用golang练习之

Saga模式

saga模式将分布式长事务切分为一系列独立短事务
每个短事务是可通过补偿动作进行撤销的
事务动作和补动作偿都是幂等的, 允许重复执行而不会有副作用

Saga由一系列的子事务“Ti”组成，
每个Ti都有对应的补偿“Ci”，
当Ti出现问题时Ci用于处理Ti执行带来的问题。

可以通过下面的两个公式理解Saga模式。
T = T1 T2 … Tn
T = TCT

Saga模式的核心理念是避免使用长期持有锁（如14.2.2节介绍的两阶段提交）的长事务，
而应该将事务切分为一组按序依次提交的短事务，
Saga模式满足ACD（原子性、一致性、持久性）特征。

摘自 <<Go微服务实战>> 刘金亮, 2021.1

目标

为实现saga模式的分布式事务, 先撸一个pub/sub事务消息队列服务
事务消息队列服务的功能性要求
- 消息不会丢失: 消息的持久化
- 消息的唯一性: 要求每个消息有全局ID和子事务ID
- 确保投递成功: 投递队列持久化, 投递状态持久化, 失败重试

子目标(Day 6)

终于可以进行基本功能测试了
- 订阅接口测试: /subscribe
- 发布接口测试: /publish
- 通知接口测试: /notify
重构
- database.go: 重构DDL语句, 以兼容sqlx的StructScan字段映射
- 所有Queryx返回的rows, 必须加上defer rows.Close()
- 添加大量的过程诊断日志
- tLiveMsgSource: fix handleMsgPublished未接收数据就return的bug

单元测试

mqs_test.go, 依次清空数据库, 测试订阅接口, 发布接口和通知接口, 并诊断过程日志和数据库记录变化.

package saga

import (
    "bytes"
    "encoding/json"
    "errors"
    "fmt"
    "github.com/jmoiron/sqlx"
    "io/ioutil"
    "learning/gooop/saga/mqs/cmd"
    "learning/gooop/saga/mqs/database"
    "learning/gooop/saga/mqs/logger"
    "learning/gooop/saga/mqs/models"
    "net/http"
    "sync"
    "testing"
    "time"
)


var gRunOnce sync.Once
func fnBootMQS() {
    gRunOnce.Do(func() {
        // boot mqs
        go cmd.BootMQS()

        // wait for mqs up
        time.Sleep(1 * time.Second)
    })
}

func fnAssertTrue (t *testing.T, b bool, msg string) {
    if !b {
        t.Fatal(msg)
    }
}

func Test_MQS(t *testing.T) {
    // prepare mqs
    fnClearDB(t)
    fnBootMQS()

    // subscribe
    fnTestSubscribe(t)
    t.Log("passed fnTestSubscribe")

    // publish and notify
    fnTestPublishAndNotify(t)
    t.Log("passed fnTestPublishAndNotify")
}


func fnTestPublishAndNotify(t *testing.T) {
    t.Log("testing fnTestPublishAndNotify")

    msg := &models.TxMsg{
        GlobalID: "test-global-id",
        SubID: "test-sub-id",
        SenderID: "test-sender-id",
        Topic: "test-topic",
        Content: "test-content",
    }
    fnPost(t, msg, "http://localhost:3333/publish")

    // check log
    fnAssertTrue(t, logger.Count("handlers.Publish, msg=test-global-id")==1, "expecting log: handlers.Publish, msg=test-global-id")

    // check notify
    time.Sleep(100 * time.Millisecond)
    fnAssertTrue(t, logger.Count("tLiveMsgSource.handleMsgPublished, clientID=test-client, msg=test-global-id")==1, "expecting log: tLiveMsgSource.handleMsgPublished")
    fnAssertTrue(t, logger.Count("tDeliveryWorker.afterDeliverySuccess, done, id=test-client, msg=test-global-id")==1, "expecting log: tDeliveryWorker.afterDeliverySuccess")
    fnAssertTrue(t, logger.Count("handlers.Notify, msg=")==1, "expecting log: handlers.Notify, msg=")

    // check success queue
    fnAssertTrue(t, fnDBCount(t, "select count(1) from success_queue where ClientID='test-client'")==1, "expectiang db.success_queue.count == 1")
}


func fnClearDB(t *testing.T) {
    fnDBExec(t, "delete from subscriber")
    fnDBExec(t, "delete from tx_msg")
    fnDBExec(t, "delete from delivery_queue")
    fnDBExec(t, "delete from success_queue")
}


func fnDBCount(t *testing.T, sql string, args... interface{}) int {
    sum := []int{ 0 }
    err := database.DB(func(db *sqlx.DB) error {
        r,e := db.Queryx(sql, args...)
        if e != nil {
            return e
        }

        defer r.Close()
        if !r.Next() {
            return errors.New("empty rows")
        }

        e = r.Scan(&sum[0])
        if e != nil {
            return e
        }

        return nil
    })

    if err != nil {
        t.Fatal(err)
    }

    return sum[0]
}


func fnDBExec(t *testing.T, sql string, args... interface{}) int {
    rows := []int64{ 0 }
    err := database.DB(func(db *sqlx.DB) error {
        r,e := db.Exec(sql, args...)
        if e != nil {
            return e
        }

        rows[0], e = r.RowsAffected()
        if e != nil {
            return e
        }

        return nil
    })

    if err != nil {
        t.Fatal(err)
    }
    return int(rows[0])
}

func fnTestSubscribe(t *testing.T) {
    t.Log("testing fnTestSubscribe")

    // clear subscriber
    fnDBExec(t, "delete from subscriber")

    msg := &models.SubscribeMsg{
        ClientID: "test-client",
        Topic: "test-topic",
        NotifyUrl: "http://localhost:3333/notify",
        ExpireTime: time.Now().UnixNano() + int64(30*time.Second),
    }
    fnPost(t, msg, "http://localhost:3333/subscribe")

    // check log
    fnAssertTrue(t, logger.Count("handlers.Subscribe, event=subscriber.registered") == 1, "expecting event=subscriber.registered")

    // check db
    count := fnDBCount(t, "select count(1) as n from subscriber where ClientID=? and topic=?", msg.ClientID, msg.Topic)
    fnAssertTrue(t, count == 1, "expecting subscriber.count == 1")
}


func fnPost(t *testing.T, msg interface{}, url string) {
    body,_ := json.Marshal(msg)
    rsp, err := http.Post(url, "application/json;charset=utf-8", bytes.NewReader(body))
    if err != nil {
        t.Fatal(err)
    }

    defer rsp.Body.Close()
    j, err := ioutil.ReadAll(rsp.Body)
    if err != nil {
        t.Fatal(err)
    }
    ok := &models.OkMsg{}
    err = json.Unmarshal(j, ok)
    if err != nil {
        t.Fatal(err)
    }
    fnAssertTrue(t, ok.OK, fmt.Sprintf("expecting replying ok from %s", url))
}

测试输出

根据assert失败提示, 逐步排查诊断日志, 主要是sqlx字段映射错误, 和未及时调用rows.Close()错误.

$ go test -v mqs_test.go 
=== RUN   Test_MQS
eventbus.Pub, event=system.boot, handler=gDeliveryService.handleBootEvent
[GIN-debug] [WARNING] Creating an Engine instance with the Logger and Recovery middleware already attached.

[GIN-debug] [WARNING] Running in "debug" mode. Switch to "release" mode in production.
 - using env:   export GIN_MODE=release
 - using code:  gin.SetMode(gin.ReleaseMode)

[GIN-debug] GET    /ping                     --> learning/gooop/saga/mqs/handlers.Ping (4 handlers)
[GIN-debug] POST   /subscribe                --> learning/gooop/saga/mqs/handlers.Subscribe (4 handlers)
[GIN-debug] POST   /publish                  --> learning/gooop/saga/mqs/handlers.Publish (4 handlers)
[GIN-debug] POST   /notify                   --> learning/gooop/saga/mqs/handlers.Notify (4 handlers)
[GIN-debug] Listening and serving HTTP on :3333
tDeliveryService.beginCreatingWorkers
tDeliveryService.beginCleanExpiredWorkers
    mqs_test.go:139: testing fnTestSubscribe
handlers.Subscribe, event=subscriber.registered, msg=&{test-client test-topic http://localhost:3333/notify 1615868155394998875}
eventbus.Pub, event=subscriber.registered, handler=gDeliveryService.handleSubscriberRegistered
[GIN] 2021/03/16 - 12:15:25 | 200 |    8.118873ms |             ::1 | POST     "/subscribe"
[GIN] 2021/03/16 - 12:15:25 | 200 |    8.214968ms |             ::1 | POST     "/subscribe"
tDeliveryWorker.afterInitialLoad, clientID=test-client, rows=0
database.DB, err=empty rows
    mqs_test.go:45: passed fnTestSubscribe
    mqs_test.go:54: testing fnTestPublishAndNotify
handlers.Publish, msg=test-global-id/test-sub-id/test-topic, msgId=15
[GIN] 2021/03/16 - 12:15:25 | 200 |   15.200109ms |             ::1 | POST     "/publish"
[GIN] 2021/03/16 - 12:15:25 | 200 |   15.216578ms |             ::1 | POST     "/publish"
handlers.Publish, pubLiveMsg 15
handlers.Publish, pubLiveMsg, msgId=15, rows=1
handlers.Publish, event=msg.published, clientID=test-client, msg=test-global-id/test-sub-id/http://localhost:3333/notify
eventbus.Pub, event=msg.published, handler=tLiveMsgSource.test-client
tLiveMsgSource.handleMsgPublished, clientID=test-client, msg=test-global-id/test-sub-id/test-topic
handlers.Notify, msg=&{test-global-id test-sub-id  0 test-topic test-content}
[GIN] 2021/03/16 - 12:15:25 | 200 |       48.38µs |             ::1 | POST     "/notify"
[GIN] 2021/03/16 - 12:15:25 | 200 |     110.659µs |             ::1 | POST     "/notify"
tDeliveryWorker.afterDeliverySuccess, done, id=test-client, msg=test-global-id/test-sub-id
    mqs_test.go:49: passed fnTestPublishAndNotify
--- PASS: Test_MQS (1.18s)
PASS
ok      command-line-arguments  1.187s

(未完待续)

有疑问加站长微信联系（非本文作者）

本文来自：简书

感谢作者：老罗话编程

查看原文：手撸golang GO与微服务 Saga模式之6

入群交流（和以上内容无关）：加入Go大咖交流群，或添加微信：liuxiaoyan-s 备注：入群；或加QQ群：692541889

402 次点击

加入收藏微博

收入我的专栏

上一篇：修改文件夹下面所有.jpg文件名demo ->欢迎加入微信交流群一起交流学习go

下一篇：Go text/template

gin

微服务

测试

功能测试

0 回复

添加一条新回复（您需要登录后才能回复没有账号？）

请尽量让自己的回复能够对别人有帮助
支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
支持 @ 本站用户；支持表情（输入 : 提示），见 Emoji cheat sheet
图片支持拖拽、截图粘贴等方式上传

手撸golang GO与微服务 Saga模式之6

缘起

Saga模式

目标

子目标(Day 6)

单元测试

测试输出

用户登录

今日阅读排行

一周阅读排行

关注我

缘起

Saga模式

目标

子目标(Day 6)

单元测试

测试输出

手撸golang GO与微服务 Saga模式之6

缘起

Saga模式

目标

子目标(Day 6)

单元测试

测试输出

用户登录

今日阅读排行

一周阅读排行

关注我

给该专栏投稿 写篇新文章

收入到我管理的专栏 新建专栏

缘起

Saga模式

目标

子目标(Day 6)

单元测试

测试输出

给该专栏投稿写篇新文章

收入到我管理的专栏新建专栏