缘起

最近阅读<> (刘金亮, 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（原子性、一致性、持久性）特征。

摘自 <> 刘金亮, 2021.1

  1. <a name="fGPkq"></a>
  2. # 目标
  4. - 为实现saga模式的分布式事务, 先撸一个pub/sub事务消息队列服务
  5. - 事务消息队列服务的功能性要求
  6.    - 消息不会丢失: 消息的持久化
  7.    - 消息的唯一性: 要求每个消息有全局ID和子事务ID
  8.    - 确保投递成功: 投递队列持久化, 投递状态持久化, 失败重试
  10. <a name="naUyZ"></a>
  11. # 子目标(Day 5)
  13. - 重构和完善消息投递机制
  14.    - iMsgHeap: 使用待投递消息堆缓存消息. 总是优先投递创建时间最小的消息
  15.    - iMsgSource: 定义消息来源接口. 有两种消息来源, 1-数据库；2-eventbus
  16.    - iMsgHistoryRing: 使用ring buffer记录近期已投递成功的消息, 防止重复投递
  17.    - tConcurrentMsgHeap: 最小CreateTime优先的消息堆, 实现iMsgHeap接口, 并且是线程安全的.
  18.    - tDBMsgSource: 从数据库拉取待投递消息, 实现iMsgSource接口
  19.    - tLiveMsgSource: 监听eventbus即时推送的投递消息, 实现iMsgSource接口
  20.    - tMsgHistoryRing: 历史消息的固定大小环形队列, 实现iMsgHistoryRing接口, 缓存近期已投递成功的消息
  21.    - tDeliveryWorker: 
  22.       - 初始化时, 优先从数据库加载待投递消息
  23.       - 使用iMsgHeap缓存待投递消息, 并确保有序
  24.       - 使用iMsgSource接口, 分别从db和eventbus接收投递消息
  25.       - 使用iMsgHistoryRing, 缓存已投递成功的消息, 防止重复投递
  28. <a name="SoPHV"></a>
  29. # iMsgHeap.go
  30. 使用待投递消息堆缓存消息. 总是优先投递创建时间最小的消息
  31. ```go
  32. package delivery
  34. import "learning/gooop/saga/mqs/models"
  36. type iMsgHeap interface {
  37.     Size() int
  38.     IsEmpty() bool
  39.     IsNotEmpty() bool
  41.     Push(msg *models.QueuedMsg)
  42.     Peek() *models.QueuedMsg
  43.     Pop() *models.QueuedMsg
  44. }

iMsgSource.go

定义消息来源接口. 有两种消息来源, 1-数据库；2-eventbus

  1. package delivery
  3. import "learning/gooop/saga/mqs/models"
  5. type iMsgSource interface {
  6.     MsgChan() <- chan *models.QueuedMsg
  7. }
  9. type tSourceExpireFunc func() bool

iMsgHistoryRing.go

使用ring buffer记录近期已投递成功的消息, 防止重复投递

  1. package delivery
  3. import "learning/gooop/saga/mqs/models"
  5. type iMsgHistoryRing interface {
  6.     Push(msg *models.QueuedMsg)
  7.     Has(id int) bool
  8. }

tConcurrentMsgHeap.go

最小CreateTime优先的消息堆, 实现iMsgHeap接口, 并且是线程安全的.

  1. package delivery
  3. import (
  4.     "learning/gooop/saga/mqs/models"
  5.     "sync"
  6. )
  8. type tConcurrentMsgHeap struct {
  9.     items []*models.QueuedMsg
  10.     size int
  11.     mutex *sync.Mutex
  12. }
  14. func newMsgHeap() iMsgHeap {
  15.     it := new(tConcurrentMsgHeap)
  16.     it.init()
  17.     return it
  18. }
  20. func (me *tConcurrentMsgHeap) init() {
  21.     me.items = make([]*models.QueuedMsg, 0)
  22.     me.size = 0
  23.     me.mutex = new(sync.Mutex)
  24. }
  26. func (me *tConcurrentMsgHeap) Size() int {
  27.     return me.size
  28. }
  30. func (me *tConcurrentMsgHeap) IsEmpty() bool {
  31.     return me.size <= 0
  32. }
  34. func (me *tConcurrentMsgHeap) IsNotEmpty() bool {
  35.     return !me.IsEmpty()
  36. }
  39. func (me *tConcurrentMsgHeap) has(msgID int) bool {
  40.     for _,it := range me.items {
  41.         if it.MsgID == msgID {
  42.             return true
  43.         }
  44.     }
  45.     return false
  46. }
  48. func (me *tConcurrentMsgHeap) Push(msg *models.QueuedMsg) {
  49.     me.mutex.Lock()
  50.     defer me.mutex.Unlock()
  52.     if me.has(msg.MsgID) {
  53.         return
  54.     }
  56.     me.ensureSize(me.size + 1)
  57.     me.items[me.size] = msg
  58.     me.size++
  60.     me.shiftUp(me.size - 1)
  61. }
  64. func (me *tConcurrentMsgHeap) ensureSize(size int) {
  65.     for ;len(me.items) < size; {
  66.         me.items = append(me.items, nil)
  67.     }
  68. }
  70. func (me *tConcurrentMsgHeap) parentOf(i int) int {
  71.     return (i - 1) / 2
  72. }
  74. func (me *tConcurrentMsgHeap) leftChildOf(i int) int {
  75.     return i*2 + 1
  76. }
  78. func (me *tConcurrentMsgHeap) rightChildOf(i int) int {
  79.     return me.leftChildOf(i) + 1
  80. }
  82. func (me *tConcurrentMsgHeap) last() (i int, v *models.QueuedMsg) {
  83.     if me.IsEmpty() {
  84.         return -1, nil
  85.     }
  87.     i = me.size - 1
  88.     v = me.items[i]
  89.     return i,v
  90. }
  92. func (me *tConcurrentMsgHeap) shiftUp(i int) {
  93.     if i <= 0 {
  94.         return
  95.     }
  96.     v := me.items[i]
  98.     pi := me.parentOf(i)
  99.     pv := me.items[pi]
  101.     if me.less(v, pv) {
  102.         me.items[pi], me.items[i] = v, pv
  103.         me.shiftUp(pi)
  104.     }
  105. }
  107. func (me *tConcurrentMsgHeap) less(a, b *models.QueuedMsg) bool {
  108.     return a.CreateTime < b.CreateTime
  109. }
  111. func (me *tConcurrentMsgHeap) Pop() *models.QueuedMsg {
  112.     me.mutex.Lock()
  113.     defer me.mutex.Unlock()
  115.     if me.IsEmpty() {
  116.         return nil
  117.     }
  120.     top := me.items[0]
  121.     li, lv := me.last()
  122.     me.items[0] = nil
  123.     me.size--
  125.     if me.IsEmpty() {
  126.         return top
  127.     }
  129.     me.items[0] = lv
  130.     me.items[li] = nil
  132.     me.shiftDown(0)
  133.     return top
  134. }
  137. func (me *tConcurrentMsgHeap) Peek() *models.QueuedMsg {
  138.     me.mutex.Lock()
  139.     defer me.mutex.Unlock()
  141.     if me.IsEmpty() {
  142.         return nil
  143.     }
  145.     return me.items[0]
  146. }
  148. func (me *tConcurrentMsgHeap) shiftDown(i int) {
  149.     pv := me.items[i]
  150.     ok, ci, cv := me.minChildOf(i)
  151.     if ok && me.less(cv, pv) {
  152.         me.items[i], me.items[ci] = cv, pv
  153.         me.shiftDown(ci)
  154.     }
  155. }
  157. func (me *tConcurrentMsgHeap) minChildOf(p int) (ok bool, i int, v *models.QueuedMsg) {
  158.     li := me.leftChildOf(p)
  159.     if li >= me.size {
  160.         return false, 0, nil
  161.     }
  162.     lv := me.items[li]
  164.     ri := me.rightChildOf(p)
  165.     if ri >= me.size {
  166.         return true, li, lv
  167.     }
  168.     rv := me.items[ri]
  170.     if me.less(lv, rv) {
  171.         return true, li, lv
  172.     } else {
  173.         return true, ri, rv
  174.     }
  175. }

tDBMsgSource.go

从数据库拉取待投递消息, 实现iMsgSource接口

  1. package delivery
  3. import (
  4.     "github.com/jmoiron/sqlx"
  5.     "learning/gooop/saga/mqs/database"
  6.     "learning/gooop/saga/mqs/models"
  7.     "time"
  8. )
  10. type tDBMsgSource struct {
  11.     clientID string
  12.     expireFunc tSourceExpireFunc
  13.     msgChan chan *models.QueuedMsg
  14. }
  16. func newDBMsgSource(clientID string, expireFunc tSourceExpireFunc) iMsgSource {
  17.     it := new(tDBMsgSource)
  18.     it.init(clientID, expireFunc)
  19.     return it
  20. }
  22. func (me *tDBMsgSource) init(clientID string, expireFunc tSourceExpireFunc) {
  23.     me.clientID = clientID
  24.     me.expireFunc = expireFunc
  25.     me.msgChan = make(chan *models.QueuedMsg, 1)
  26.     go me.beginPollDB()
  27. }
  29. func (me *tDBMsgSource) MsgChan() <- chan *models.QueuedMsg {
  30.     return me.msgChan
  31. }
  33. func (me *tDBMsgSource) beginPollDB() {
  34.     interval := time.Duration(1) * time.Second
  35.     for !me.expireFunc() {
  36.         if len(me.msgChan) <= 0 {
  37.             ok, msg := me.poll()
  38.             if ok {
  39.                 me.msgChan <- msg
  40.                 continue
  41.             }
  42.         }
  44.         // poll failed, or chan full
  45.         time.Sleep(interval)
  46.     }
  48.     close(me.msgChan)
  49. }
  51. func (me *tDBMsgSource) poll() (bool, *models.QueuedMsg) {
  52.     msg := &models.QueuedMsg{}
  53.     e := database.DB(func(db *sqlx.DB) error {
  54.         rows, err := db.Queryx(
  55.             "select * from delivery_queue where client_id=? order by create_time asc limit 1",
  56.             me.clientID,
  57.         )
  58.         if err != nil {
  59.             return err
  60.         }
  62.         if rows.Next() {
  63.             err = rows.StructScan(msg)
  64.             if err != nil {
  65.                 return err
  66.             }
  67.             return nil
  69.         } else {
  70.             return gEmptyRowsErr
  71.         }
  72.     })
  74.     if e != nil {
  75.         return false, nil
  76.     } else {
  77.         return true, msg
  78.     }
  79. }

tLiveMsgSource.go

监听eventbus即时推送的投递消息, 实现iMsgSource接口

  1. package delivery
  3. import (
  4.     "fmt"
  5.     "learning/gooop/saga/mqs/eventbus"
  6.     "learning/gooop/saga/mqs/logger"
  7.     "learning/gooop/saga/mqs/models"
  8.     "learning/gooop/saga/mqs/models/events"
  9.     "time"
  10. )
  12. type tLiveMsgSource struct {
  13.     clientID string
  14.     expireFunc tSourceExpireFunc
  15.     msgChan chan *models.QueuedMsg
  16. }
  18. func newLiveMsgSource(clientID string, expireFunc tSourceExpireFunc) iMsgSource {
  19.     it := new(tLiveMsgSource)
  20.     it.init(clientID, expireFunc)
  21.     return it
  22. }
  24. func (me *tLiveMsgSource) init(clientID string, expireFunc tSourceExpireFunc) {
  25.     me.clientID = clientID
  26.     me.expireFunc = expireFunc
  27.     me.msgChan = make(chan *models.QueuedMsg, 1)
  29.     eventbus.GlobalEventBus.Sub(events.MsgPublishedEvent,
  30.         me.id(),
  31.         me.handleMsgPublished)
  32.     go me.beginWatchExpire()
  33. }
  36. func (me *tLiveMsgSource) id() string {
  37.     return fmt.Sprintf("tLiveMsgSource.%s", me.clientID)
  38. }
  40. func (me *tLiveMsgSource) beginWatchExpire() {
  41.     for range time.Tick(1 * time.Second) {
  42.         if me.expireFunc() {
  43.             me.afterExpired()
  44.             return
  45.         }
  46.     }
  47. }
  49. func (me *tLiveMsgSource) afterExpired() {
  50.     eventbus.GlobalEventBus.Unsub(events.MsgPublishedEvent, me.id())
  51.     close(me.msgChan)
  52. }
  54. func (me *tLiveMsgSource) handleMsgPublished(_ string, args interface{}) {
  55.     msg, ok := args.(*models.QueuedMsg)
  56.     if !ok {
  57.         return
  58.     }
  60.     if msg.ClientID != me.clientID {
  61.         return
  62.     }
  64.     if len(me.msgChan) >= 0 {
  65.         return
  66.     }
  68.     logger.Logf(
  69.         "tLiveMsgSource.handleMsgPublished, clientID=%s, msg=%s/%s/%s",
  70.         me.clientID, msg.GlobalID, msg.SubID, msg.Topic )
  71.     me.msgChan <- msg
  72. }
  74. func (me *tLiveMsgSource) MsgChan() <- chan *models.QueuedMsg {
  75.     return me.msgChan
  76. }

tMsgHistoryRing.go

历史消息的固定大小环形队列, 实现iMsgHistoryRing接口, 缓存近期已投递成功的消息

  1. package delivery
  3. import "learning/gooop/saga/mqs/models"
  6. type tMsgHistoryRing struct {
  7.     items []*models.QueuedMsg
  8.     capacity int
  9.     index int
  10. }
  12. func newMsgHistoryRing(capacity int) iMsgHistoryRing {
  13.     it := new(tMsgHistoryRing)
  14.     it.init(capacity)
  15.     return it
  16. }
  18. func (me *tMsgHistoryRing) init(capacity int) {
  19.     me.items = make([]*models.QueuedMsg, capacity)
  20.     me.capacity = capacity
  21.     me.index = 0
  22. }
  25. func (me *tMsgHistoryRing) Has(id int) bool {
  26.     for _,it := range me.items {
  27.         if it != nil && it.ID == id {
  28.             return true
  29.         }
  30.     }
  32.     return false
  33. }
  35. func (me *tMsgHistoryRing) Push(msg *models.QueuedMsg) {
  36.     me.items[me.index] = msg
  38.     me.index++
  39.     if me.index >= me.capacity {
  40.         me.index = 0
  41.     }
  42. }

tDeliveryWorker.go

  • 初始化时, 优先从数据库加载待投递消息
  • 使用iMsgHeap缓存待投递消息, 并确保有序
  • 使用iMsgSource接口, 分别从db和eventbus接收投递消息
  • 使用iMsgHistoryRing, 缓存已投递成功的消息, 防止重复投递 ```go package delivery

import ( “bytes” “encoding/json” “errors” “github.com/jmoiron/sqlx” “io/ioutil” “learning/gooop/saga/mqs/database” “learning/gooop/saga/mqs/logger” “learning/gooop/saga/mqs/models” “net/http” “time” )

type tDeliveryWorker struct { info *tWorkerInfo successRing iMsgHistoryRing dbSource iMsgSource liveSource iMsgSource msgHeap iMsgHeap }

func newDeliveryWorker(info tWorkerInfo) tDeliveryWorker { it := new(tDeliveryWorker) it.init(info) return it }

// init: do initialization, and start initial load func (me tDeliveryWorker) init(info tWorkerInfo) { me.info = info me.successRing = newMsgHistoryRing(64)

  1. me.dbSource = newDBMsgSource(info.ClientID, me.isExpired)
  2. me.liveSource = newLiveMsgSource(info.ClientID, me.isExpired)
  3. me.msgHeap = newMsgHeap()
  5. go me.beginInitialLoadFromDB()

}

// beginInitialLoadFromDB: initially, load queued msg from database func (me tDeliveryWorker) beginInitialLoadFromDB() { buf := [][]models.QueuedMsg{ nil } for !me.isExpired() { err := database.DB(func(db *sqlx.DB) error { e, rows := me.loadFromDB(db) if e != nil { return e }

  1.         buf[0] = rows
  2.         return nil
  3.     })
  5.     if err != nil {
  6.         logger.Logf("tDeliveryWorker.initialLoadFromDB, clientID=%s, err=%s", me.info.ClientID, err.Error())
  7.         time.Sleep(3 * time.Second)
  8.     } else {
  9.         me.afterInitialLoad(buf[0])
  10.     }
  11. }

}

// loadFromDB: load queued msg from database func (me tDeliveryWorker) loadFromDB(db sqlx.DB) (error, []models.QueuedMsg) { rows, err := db.Queryx( “select from delivery_queue where client_id=? order by create_time asc limit ?”, me.info.ClientID, gInitialLoadRows, ) if err != nil { return err, nil }

  1. msgList := []*models.QueuedMsg{}
  2. for rows.Next() {
  3.     msg := &models.QueuedMsg{}
  4.     err = rows.StructScan(msg)
  5.     if err != nil {
  6.         return err, nil
  7.     }
  8.     msgList = append(msgList, msg)
  9. }
  11. return nil, msgList

}

// afterInitialLoad: after initial load done, push msgs into heap, and start delivery loop func (me tDeliveryWorker) afterInitialLoad(msgList []models.QueuedMsg) { logger.Logf(“tDeliveryWorker.afterInitialLoad, clientID=%s, rows=%d”, me.info.ClientID, len(msgList)) for _,it := range msgList { me.msgHeap.Push(it) }

  1. go me.beginPollAndDeliver()

}

// beginPollAndDeliver: poll msg from heap, and then deliver it func (me *tDeliveryWorker) beginPollAndDeliver() { for !me.isExpired() { select { case msg := <- me.dbSource.MsgChan(): me.msgHeap.Push(msg) break

  1.     case msg := <- me.liveSource.MsgChan():
  2.         me.msgHeap.Push(msg)
  3.         break
  4.     }
  6.     if me.msgHeap.IsEmpty() {
  7.         continue
  8.     }
  10.     msg := me.msgHeap.Pop()
  11.     if msg == nil {
  12.         continue
  13.     }
  15.     switch msg.StatusFlag {
  16.     case 0:
  17.         // 未处理的消息
  18.         me.handleUndeliveredMsg(msg)
  19.         break
  21.     case 1:
  22.         // 处理中的消息
  23.         me.handleDeliveringMsg(msg)
  24.         break
  25.     }
  26. }

}

// isExpired: is me expired? func (me *tDeliveryWorker) isExpired() bool { return time.Now().UnixNano() >= me.info.ExpireTime }

// handleUndeliveredMsg: if msg unhandled, then try to deliver it func (me tDeliveryWorker) handleUndeliveredMsg(msg models.QueuedMsg) { err := database.DB(func(db *sqlx.DB) error { now := time.Now().UnixNano() r,e := db.Exec( “update delivery_queue set status_flag=1, update_time=? where id=? and status_flag=0 and update_time=?”, now, msg.ID, msg.UpdateTime, ) if e != nil { return e }

  1.     rows, e := r.RowsAffected()
  2.     if e != nil {
  3.         return e
  4.     }
  5.     if rows != 1 {
  6.         return gOneRowsErr
  7.     }
  9.     msg.StatusFlag = 1
  10.     msg.UpdateTime = now
  12.     return nil
  13. })
  15. if err != nil {
  16.     logger.Logf("tDeliveryWorker.handleNewMsg, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
  17.     return
  18. }
  20. if me.deliver(msg) {
  21.     me.afterDeliverySuccess(msg)
  23. } else {
  24.     me.afterDeliveryFailed(msg)
  25. }

}

// deliver: use http.Post function to delivery msg func (me tDeliveryWorker) deliver(msg models.QueuedMsg) bool { if me.successRing.Has(msg.ID) { return true }

  1. t := &models.TxMsg{
  2.     GlobalID: msg.GlobalID,
  3.     SubID: msg.SubID,
  4.     Topic: msg.Topic,
  5.     CreateTime: msg.CreateTime,
  6.     Content: msg.Content,
  7. }
  8. j,e := json.Marshal(t)
  9. if e != nil {
  10.     logger.Logf("tDeliveryWorker.deliver, failed json.Marshal, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  11.     return false
  12. }
  14. r, e := http.Post(me.info.NotifyURL, "application/json", bytes.NewReader(j))
  15. if e != nil {
  16.     logger.Logf("tDeliveryWorker.deliver, failed http.Post, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  17.     return false
  18. }
  20. defer r.Body.Close()
  21. rep, e := ioutil.ReadAll(r.Body)
  22. if e != nil {
  23.     logger.Logf("tDeliveryWorker.deliver, failed ioutil.ReadAll, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  24.     return false
  25. }
  27. m := &models.OkMsg{}
  28. e = json.Unmarshal(rep, m)
  29. if e != nil {
  30.     logger.Logf("tDeliveryWorker.deliver, failed json.Unmarshal, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  31.     return false
  32. }
  34. if m.OK {
  35.     return true
  36. } else {
  37.     logger.Logf("tDeliveryWorker.deliver, failed OkMsg.OK, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  38.     return false
  39. }

}

// handleDeliveringMsg: if delivery timeout, then retry delivery func (me tDeliveryWorker) handleDeliveringMsg(msg models.QueuedMsg) { now := time.Now().UnixNano() if msg.UpdateTime + gDeliveryTimeoutNanos > now { return }

  1. // delivery timeout
  2. me.afterDeliveryTimeout(msg)

}

// afterDeliverySuccess: if done, move msg to success queue func (me tDeliveryWorker) afterDeliverySuccess(msg models.QueuedMsg) { if me.successRing.Has(msg.ID) { return } me.successRing.Push(msg)

  1. err := database.TX(func(db *sqlx.DB, tx *sqlx.Tx) error {
  2.     r,e := db.Exec(
  3. "delete from delivery_queue where id=? and update_time=? and status_flag=1",
  4.         msg.ID,
  5.         msg.UpdateTime,
  6.     )
  7.     if e != nil {
  8.         return e
  9.     }
  11.     rows, e := r.RowsAffected()
  12.     if e != nil {
  13.         return e
  14.     }
  15.     if rows != 1 {
  16.         return gOneRowsErr
  17.     }
  19.     r, e = db.Exec(
  20.         "insert into success_queue (msg_id, client_id, create_time) values(?, ?, ?)",
  21.         msg.ID,
  22.         msg.ClientID,
  23.         time.Now().UnixNano(),
  24.     )
  25.     if e != nil {
  26.         return e
  27.     }
  29.     rows, e = r.RowsAffected()
  30.     if e != nil {
  31.         return e
  32.     }
  33.     if rows != 1 {
  34.         return gOneRowsErr
  35.     }
  37.     return nil
  38. })
  40. if err != nil {
  41.     logger.Logf("tDeliveryWorker.afterDeliverySuccess, failed, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
  42. } else {
  43.     logger.Logf("tDeliveryWorker.afterDeliverySuccess, done, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  44. }

}

// afterDeliveryFailed: if failed, do nothing but just log it func (me tDeliveryWorker) afterDeliveryFailed(msg models.QueuedMsg) { logger.Logf(“tDeliveryWorker.afterDeliveryFailed, id=%v, msg=%s/%s”, me.info.ClientID, msg.GlobalID, msg.SubID) }

// afterDeliveryTimeout: if timeout, then reset status and retry func (me tDeliveryWorker) afterDeliveryTimeout(msg models.QueuedMsg) { err := database.DB(func(db *sqlx.DB) error { r,e := db.Exec( “update delivery_queue set status_flag=0 where id=? and status_flag=1 and update_time=?”, msg.ID, msg.UpdateTime, ) if e != nil { return e }

  1.     rows,e := r.RowsAffected()
  2.     if e != nil {
  3.         return e
  4.     }
  6.     if rows != 1 {
  7.         return gOneRowsErr
  8.     }
  10.     return nil
  11. })
  13. if err != nil {
  14.     logger.Logf("tDeliveryWorker.afterDeliveryTimeout, failed, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
  15. } else {
  16.     logger.Logf("tDeliveryWorker.afterDeliveryTimeout, done, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  17. }

}

var gEmptyRowsErr = errors.New(“empty rows”) var gOneRowsErr = errors.New(“expecting one row affected”) var gDeliveryTimeoutNanos = int64(10 * (time.Second / time.Nanosecond)) var gInitialLoadRows = 100 ```

(未完待续)