分佈式事務框架 seata-golang 通訊模型詳解

時間  2020-12-13
標籤 java mysql git 程序員 github golang web sql 數據庫 apache 欄目 系統架構 简体版
原文   原文鏈接

頭圖.png

做者 | 劉曉敏 於雨java

1、簡介

Java 的世界裏，你們普遍使用的一個高性能網絡通訊框架 netty，不少 RPC 框架都是基於 netty 來實現的。在 golang 的世界裏，getty 也是一個相似 netty 的高性能網絡通訊庫。getty 最初由 dubbogo 項目負責人於雨開發，做爲底層通訊庫在 dubbo-go 中使用。隨着 dubbo-go 捐獻給 apache 基金會，在社區小夥伴的共同努力下，getty 也最終進入到 apache 這個你們庭，並更名 dubbo-getty 。mysql

18 年的時候，我在公司裏實踐微服務，當時遇到最大的問題就是分佈式事務問題。同年，阿里在社區開源他們的分佈式事務解決方案，我也很快關注到這個項目，起初還叫 fescar，後來改名 seata。因爲我對開源技術很感興趣，加了不少社區羣，當時也很關注 dubbo-go 這個項目，在裏面默默潛水。隨着對 seata 的瞭解，逐漸萌生了作一個 go 版本的分佈式事務框架的想法。git

要作一個 golang 版的分佈式事務框架，首要的一個問題就是如何實現 RPC 通訊。dubbo-go 就是很好的一個例子擺在眼前，遂開始研究 dubbo-go 的底層 getty。程序員

2、如何基於 getty 實現 RPC 通訊

getty 框架的總體模型圖以下：github

1.png

下面結合相關代碼，詳述 seata-golang 的 RPC 通訊過程。golang

1. 創建鏈接

實現 RPC 通訊，首先要創建網絡鏈接吧，咱們從 client.go 開始看起。web

func (c *client) connect() {
    var (
        err error
        ss  Session
    )

    for {
        // 創建一個 session 鏈接
        ss = c.dial()
        if ss == nil {
            // client has been closed
            break
        }
        err = c.newSession(ss)
        if err == nil {
            // 收發報文
            ss.(*session).run()
            // 此處省略部分代碼

            break
        }
        // don't distinguish between tcp connection and websocket connection. Because
        // gorilla/websocket/conn.go:(Conn)Close also invoke net.Conn.Close()
        ss.Conn().Close()
    }
}

connect() 方法經過 dial() 方法獲得了一個 session 鏈接，進入 dial() 方法：sql

func (c *client) dial() Session {
    switch c.endPointType {
    case TCP_CLIENT:
        return c.dialTCP()
    case UDP_CLIENT:
        return c.dialUDP()
    case WS_CLIENT:
        return c.dialWS()
    case WSS_CLIENT:
        return c.dialWSS()
    }

    return nil
}

咱們關注的是 TCP 鏈接，因此繼續進入 c.dialTCP() 方法：數據庫

func (c *client) dialTCP() Session {
    var (
        err  error
        conn net.Conn
    )

    for {
        if c.IsClosed() {
            return nil
        }
        if c.sslEnabled {
            if sslConfig, err := c.tlsConfigBuilder.BuildTlsConfig(); err == nil && sslConfig != nil {
                d := &net.Dialer{Timeout: connectTimeout}
                // 創建加密鏈接
                conn, err = tls.DialWithDialer(d, "tcp", c.addr, sslConfig)
            }
        } else {
            // 創建 tcp 鏈接
            conn, err = net.DialTimeout("tcp", c.addr, connectTimeout)
        }
        if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
            conn.Close()
            err = errSelfConnect
        }
        if err == nil {
            // 返回一個 TCPSession
            return newTCPSession(conn, c)
        }

        log.Infof("net.DialTimeout(addr:%s, timeout:%v) = error:%+v", c.addr, connectTimeout, perrors.WithStack(err))
        <-wheel.After(connectInterval)
    }
}

至此，咱們知道了 getty 如何創建 TCP 鏈接，並返回 TCPSession。apache

2. 收發報文

那它是怎麼收發報文的呢，咱們回到 connection 方法接着往下看，有這樣一行 ss.(*session).run()，在這行代碼以後代碼都是很簡單的操做，咱們猜想這行代碼運行的邏輯裏面必定包含收發報文的邏輯，接着進入 run() 方法：

func (s *session) run() {
    // 省略部分代碼

    go s.handleLoop()
    go s.handlePackage()
}

<br />這裏起了兩個 goroutine，handleLoop 和 handlePackage，看字面意思符合咱們的猜測，進入 handleLoop() 方法：<br />

func (s *session) handleLoop() {
    // 省略部分代碼

    for {
        // A select blocks until one of its cases is ready to run.
        // It choose one at random if multiple are ready. Otherwise it choose default branch if none is ready.
        select {
        // 省略部分代碼

        case outPkg, ok = <-s.wQ:
            // 省略部分代碼

            iovec = iovec[:0]
            for idx := 0; idx < maxIovecNum; idx++ {
        // 經過 s.writer 將 interface{} 類型的 outPkg 編碼成二進制的比特
                pkgBytes, err = s.writer.Write(s, outPkg)
                // 省略部分代碼

                iovec = append(iovec, pkgBytes)

                //省略部分代碼
            }
            // 將這些二進制比特發送出去
            err = s.WriteBytesArray(iovec[:]...)
            if err != nil {
                log.Errorf("%s, [session.handleLoop]s.WriteBytesArray(iovec len:%d) = error:%+v",
                    s.sessionToken(), len(iovec), perrors.WithStack(err))
                s.stop()
                // break LOOP
                flag = false
            }

        case <-wheel.After(s.period):
            if flag {
                if wsFlag {
                    err := wsConn.writePing()
                    if err != nil {
                        log.Warnf("wsConn.writePing() = error:%+v", perrors.WithStack(err))
                    }
                }
                // 定時執行的邏輯，心跳等
                s.listener.OnCron(s)
            }
        }
    }
}

經過上面的代碼，咱們不難發現，handleLoop() 方法處理的是發送報文的邏輯，RPC 須要發送的消息首先由 s.writer 編碼成二進制比特，而後經過創建的 TCP 鏈接發送出去。這個 s.writer 對應的 Writer 接口是 RPC 框架必需要實現的一個接口。

繼續看 handlePackage() 方法：

func (s *session) handlePackage() {
    // 省略部分代碼

    if _, ok := s.Connection.(*gettyTCPConn); ok {
        if s.reader == nil {
            errStr := fmt.Sprintf("session{name:%s, conn:%#v, reader:%#v}", s.name, s.Connection, s.reader)
            log.Error(errStr)
            panic(errStr)
        }

        err = s.handleTCPPackage()
    } else if _, ok := s.Connection.(*gettyWSConn); ok {
        err = s.handleWSPackage()
    } else if _, ok := s.Connection.(*gettyUDPConn); ok {
        err = s.handleUDPPackage()
    } else {
        panic(fmt.Sprintf("unknown type session{%#v}", s))
    }
}

進入 handleTCPPackage() 方法：

func (s *session) handleTCPPackage() error {
    // 省略部分代碼

    conn = s.Connection.(*gettyTCPConn)
    for {
        // 省略部分代碼

        bufLen = 0
        for {
            // for clause for the network timeout condition check
            // s.conn.SetReadTimeout(time.Now().Add(s.rTimeout))
            // 從 TCP 鏈接中收到報文
            bufLen, err = conn.recv(buf)
            // 省略部分代碼

            break
        }
        // 省略部分代碼

        // 將收到的報文二進制比特寫入 pkgBuf
        pktBuf.Write(buf[:bufLen])
        for {
            if pktBuf.Len() <= 0 {
                break
            }
            // 經過 s.reader 將收到的報文解碼成 RPC 消息
            pkg, pkgLen, err = s.reader.Read(s, pktBuf.Bytes())
            // 省略部分代碼

      s.UpdateActive()
            // 將收到的消息放入 TaskQueue 供 RPC 消費端消費
            s.addTask(pkg)
            pktBuf.Next(pkgLen)
            // continue to handle case 5
        }
        if exit {
            break
        }
    }

    return perrors.WithStack(err)
}

從上面的代碼邏輯咱們分析出，RPC 消費端須要將從 TCP 鏈接收到的二進制比特報文解碼成 RPC 能消費的消息，這個工做由 s.reader 實現，因此，咱們要構建 RPC 通訊層也須要實現 s.reader 對應的 Reader 接口。

3. 底層處理網絡報文的邏輯如何與業務邏輯解耦

咱們都知道，netty 經過 boss 線程和 worker 線程實現了底層網絡邏輯和業務邏輯的解耦。那麼，getty 是如何實現的呢？

在 handlePackage() 方法最後，咱們看到，收到的消息被放入了 s.addTask(pkg) 這個方法，接着往下分析：

func (s *session) addTask(pkg interface{}) {
    f := func() {
        s.listener.OnMessage(s, pkg)
        s.incReadPkgNum()
    }
    if taskPool := s.EndPoint().GetTaskPool(); taskPool != nil {
        taskPool.AddTaskAlways(f)
        return
    }
    f()
}

pkg 參數傳遞到了一個匿名方法，這個方法最終放入了 taskPool。這個方法很關鍵，在我後來寫 seata-golang 代碼的時候，就遇到了一個坑，這個坑後面分析。

接着咱們看一下 taskPool 的定義：

// NewTaskPoolSimple build a simple task pool
func NewTaskPoolSimple(size int) GenericTaskPool {
    if size < 1 {
        size = runtime.NumCPU() * 100
    }
    return &taskPoolSimple{
        work: make(chan task),
        sem:  make(chan struct{}, size),
        done: make(chan struct{}),
    }
}

構建了一個緩衝大小爲 size （默認爲  runtime.NumCPU() * 100） 的 channel sem。再看方法 AddTaskAlways(t task)：

func (p *taskPoolSimple) AddTaskAlways(t task) {
    select {
    case <-p.done:
        return
    default:
    }

    select {
    case p.work <- t:
        return
    default:
    }
    select {
    case p.work <- t:
    case p.sem <- struct{}{}:
        p.wg.Add(1)
        go p.worker(t)
    default:
        goSafely(t)
    }
}

加入的任務，會先由 len(p.sem) 個 goroutine 去消費，若是沒有 goroutine 空閒，則會啓動一個臨時的 goroutine 去運行 t()。至關於有  len(p.sem) 個 goroutine 組成了 goroutine pool，pool 中的 goroutine 去處理業務邏輯，而不是由處理網絡報文的 goroutine 去運行業務邏輯，從而實現瞭解耦。寫 seata-golang 時遇到的一個坑，就是忘記設置 taskPool 形成了處理業務邏輯和處理底層網絡報文邏輯的 goroutine 是同一個，我在業務邏輯中阻塞等待一個任務完成時，阻塞了整個 goroutine，使得阻塞期間收不到任何報文。

4. 具體實現

下面的代碼見 getty.go：

// Reader is used to unmarshal a complete pkg from buffer
type Reader interface {
    Read(Session, []byte) (interface{}, int, error)
}

// Writer is used to marshal pkg and write to session
type Writer interface {
    // if @Session is udpGettySession, the second parameter is UDPContext.
    Write(Session, interface{}) ([]byte, error)
}

// ReadWriter interface use for handle application packages
type ReadWriter interface {
    Reader
    Writer
}
// EventListener is used to process pkg that received from remote session
type EventListener interface {
    // invoked when session opened
    // If the return error is not nil, @Session will be closed.
    OnOpen(Session) error

    // invoked when session closed.
    OnClose(Session)

    // invoked when got error.
    OnError(Session, error)

    // invoked periodically, its period can be set by (Session)SetCronPeriod
    OnCron(Session)

    // invoked when getty received a package. Pls attention that do not handle long time
    // logic processing in this func. You'd better set the package's maximum length.
    // If the message's length is greater than it, u should should return err in
    // Reader{Read} and getty will close this connection soon.
    //
    // If ur logic processing in this func will take a long time, u should start a goroutine
    // pool(like working thread pool in cpp) to handle the processing asynchronously. Or u
    // can do the logic processing in other asynchronous way.
    // !!!In short, ur OnMessage callback func should return asap.
    //
    // If this is a udp event listener, the second parameter type is UDPContext.
    OnMessage(Session, interface{})
}

經過對整個 getty 代碼的分析，咱們只要實現  ReadWriter 來對 RPC  消息編解碼，再實現 EventListener 來處理 RPC 消息的對應的具體邏輯，將 ReadWriter 實現和 EventLister 實現注入到 RPC 的 Client 和 Server 端，則可實現 RPC 通訊。

4.1 編解碼協議實現

下面是 seata 協議的定義：

2.png

在 ReadWriter 接口的實現 RpcPackageHandler 中，調用 Codec 方法對消息體按照上面的格式編解碼：

// 消息編碼爲二進制比特
func MessageEncoder(codecType byte, in interface{}) []byte {
    switch codecType {
    case SEATA:
        return SeataEncoder(in)
    default:
        log.Errorf("not support codecType, %s", codecType)
        return nil
    }
}

// 二進制比特解碼爲消息體
func MessageDecoder(codecType byte, in []byte) (interface{}, int) {
    switch codecType {
    case SEATA:
        return SeataDecoder(in)
    default:
        log.Errorf("not support codecType, %s", codecType)
        return nil, 0
    }
}

4.2 Client 端實現

再來看 client 端 EventListener 的實現 RpcRemotingClient：

func (client *RpcRemoteClient) OnOpen(session getty.Session) error {
    go func() 
        request := protocal.RegisterTMRequest{AbstractIdentifyRequest: protocal.AbstractIdentifyRequest{
            ApplicationId:           client.conf.ApplicationId,
            TransactionServiceGroup: client.conf.TransactionServiceGroup,
        }}
    // 創建鏈接後向 Transaction Coordinator 發起註冊 TransactionManager 的請求
        _, err := client.sendAsyncRequestWithResponse(session, request, RPC_REQUEST_TIMEOUT)
        if err == nil {
      // 將與 Transaction Coordinator 創建的鏈接保存在鏈接池供後續使用
            clientSessionManager.RegisterGettySession(session)
            client.GettySessionOnOpenChannel <- session.RemoteAddr()
        }
    }()

    return nil
}

// OnError ...
func (client *RpcRemoteClient) OnError(session getty.Session, err error) {
    clientSessionManager.ReleaseGettySession(session)
}

// OnClose ...
func (client *RpcRemoteClient) OnClose(session getty.Session) {
    clientSessionManager.ReleaseGettySession(session)
}

// OnMessage ...
func (client *RpcRemoteClient) OnMessage(session getty.Session, pkg interface{}) {
    log.Info("received message:{%v}", pkg)
    rpcMessage, ok := pkg.(protocal.RpcMessage)
    if ok {
        heartBeat, isHeartBeat := rpcMessage.Body.(protocal.HeartBeatMessage)
        if isHeartBeat && heartBeat == protocal.HeartBeatMessagePong {
            log.Debugf("received PONG from %s", session.RemoteAddr())
        }
    }

    if rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST ||
        rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST_ONEWAY {
        log.Debugf("msgId:%s, body:%v", rpcMessage.Id, rpcMessage.Body)

        // 處理事務消息，提交 or 回滾
        client.onMessage(rpcMessage, session.RemoteAddr())
    } else {
        resp, loaded := client.futures.Load(rpcMessage.Id)
        if loaded {
            response := resp.(*getty2.MessageFuture)
            response.Response = rpcMessage.Body
            response.Done <- true
            client.futures.Delete(rpcMessage.Id)
        }
    }
}

// OnCron ...
func (client *RpcRemoteClient) OnCron(session getty.Session) {
  // 發送心跳
    client.defaultSendRequest(session, protocal.HeartBeatMessagePing)
}

clientSessionManager.RegisterGettySession(session) 的邏輯將在下文中分析。

4.3 Server 端 Transaction Coordinator 實現

代碼見 DefaultCoordinator：

func (coordinator *DefaultCoordinator) OnOpen(session getty.Session) error {
    log.Infof("got getty_session:%s", session.Stat())
    return nil
}

func (coordinator *DefaultCoordinator) OnError(session getty.Session, err error) {
    // 釋放 TCP 鏈接
  SessionManager.ReleaseGettySession(session)
    session.Close()
    log.Errorf("getty_session{%s} got error{%v}, will be closed.", session.Stat(), err)
}

func (coordinator *DefaultCoordinator) OnClose(session getty.Session) {
    log.Info("getty_session{%s} is closing......", session.Stat())
}

func (coordinator *DefaultCoordinator) OnMessage(session getty.Session, pkg interface{}) {
    log.Debugf("received message:{%v}", pkg)
    rpcMessage, ok := pkg.(protocal.RpcMessage)
    if ok {
        _, isRegTM := rpcMessage.Body.(protocal.RegisterTMRequest)
        if isRegTM {
      // 將 TransactionManager 信息和 TCP 鏈接創建映射關係
            coordinator.OnRegTmMessage(rpcMessage, session)
            return
        }

        heartBeat, isHeartBeat := rpcMessage.Body.(protocal.HeartBeatMessage)
        if isHeartBeat && heartBeat == protocal.HeartBeatMessagePing {
            coordinator.OnCheckMessage(rpcMessage, session)
            return
        }

        if rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST ||
            rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST_ONEWAY {
            log.Debugf("msgId:%s, body:%v", rpcMessage.Id, rpcMessage.Body)
            _, isRegRM := rpcMessage.Body.(protocal.RegisterRMRequest)
            if isRegRM {
        // 將 ResourceManager 信息和 TCP 鏈接創建映射關係
                coordinator.OnRegRmMessage(rpcMessage, session)
            } else {
                if SessionManager.IsRegistered(session) {
                    defer func() {
                        if err := recover(); err != nil {
                            log.Errorf("Catch Exception while do RPC, request: %v,err: %w", rpcMessage, err)
                        }
                    }()
          // 處理事務消息，全局事務註冊、分支事務註冊、分支事務提交、全局事務回滾等
                    coordinator.OnTrxMessage(rpcMessage, session)
                } else {
                    session.Close()
                    log.Infof("close a unhandled connection! [%v]", session)
                }
            }
        } else {
            resp, loaded := coordinator.futures.Load(rpcMessage.Id)
            if loaded {
                response := resp.(*getty2.MessageFuture)
                response.Response = rpcMessage.Body
                response.Done <- true
                coordinator.futures.Delete(rpcMessage.Id)
            }
        }
    }
}

func (coordinator *DefaultCoordinator) OnCron(session getty.Session) {

}

coordinator.OnRegTmMessage(rpcMessage, session) 註冊 Transaction Manager，coordinator.OnRegRmMessage(rpcMessage, session) 註冊 Resource Manager。具體邏輯分析見下文。

消息進入 coordinator.OnTrxMessage(rpcMessage, session) 方法，將按照消息的類型碼路由到具體的邏輯當中：

switch msg.GetTypeCode() {
    case protocal.TypeGlobalBegin:
        req := msg.(protocal.GlobalBeginRequest)
        resp := coordinator.doGlobalBegin(req, ctx)
        return resp
    case protocal.TypeGlobalStatus:
        req := msg.(protocal.GlobalStatusRequest)
        resp := coordinator.doGlobalStatus(req, ctx)
        return resp
    case protocal.TypeGlobalReport:
        req := msg.(protocal.GlobalReportRequest)
        resp := coordinator.doGlobalReport(req, ctx)
        return resp
    case protocal.TypeGlobalCommit:
        req := msg.(protocal.GlobalCommitRequest)
        resp := coordinator.doGlobalCommit(req, ctx)
        return resp
    case protocal.TypeGlobalRollback:
        req := msg.(protocal.GlobalRollbackRequest)
        resp := coordinator.doGlobalRollback(req, ctx)
        return resp
    case protocal.TypeBranchRegister:
        req := msg.(protocal.BranchRegisterRequest)
        resp := coordinator.doBranchRegister(req, ctx)
        return resp
    case protocal.TypeBranchStatusReport:
        req := msg.(protocal.BranchReportRequest)
        resp := coordinator.doBranchReport(req, ctx)
        return resp
    default:
        return nil
    }

4.4 session manager 分析

Client 端同 Transaction Coordinator 創建鏈接起鏈接後，經過 clientSessionManager.RegisterGettySession(session) 將鏈接保存在 serverSessions = sync.Map{} 這個 map 中。map 的 key 爲從 session 中獲取的 RemoteAddress 即 Transaction Coordinator 的地址，value 爲 session。這樣，Client 端就能夠經過 map 中的一個 session 來向 Transaction Coordinator 註冊 Transaction Manager 和 Resource Manager 了。具體代碼見 getty_client_session_manager.go。

Transaction Manager 和 Resource Manager 註冊到 Transaction Coordinator 後，一個鏈接既有可能用來發送 TM 消息也有可能用來發送 RM 消息。咱們經過 RpcContext 來標識一個鏈接信息：

type RpcContext struct {
    Version                 string
    TransactionServiceGroup string
    ClientRole              meta.TransactionRole
    ApplicationId           string
    ClientId                string
    ResourceSets            *model.Set
    Session                 getty.Session
}

當收到事務消息時，咱們須要構造這樣一個 RpcContext 供後續事務處理邏輯使用。因此，咱們會構造下列 map 來緩存映射關係：

var (
    // session -> transactionRole
    // TM will register before RM, if a session is not the TM registered,
    // it will be the RM registered
    session_transactionroles = sync.Map{}

    // session -> applicationId
    identified_sessions = sync.Map{}

    // applicationId -> ip -> port -> session
    client_sessions = sync.Map{}

    // applicationId -> resourceIds
    client_resources = sync.Map{}
)

這樣，Transaction Manager 和 Resource Manager 分別經過 coordinator.OnRegTmMessage(rpcMessage, session) 和 coordinator.OnRegRmMessage(rpcMessage, session) 註冊到 Transaction Coordinator 時，會在上述 client_sessions map 中緩存 applicationId、ip、port 與 session 的關係，在 client_resources map 中緩存 applicationId 與 resourceIds（一個應用可能存在多個 Resource Manager） 的關係。在須要時，咱們就能夠經過上述映射關係構造一個 RpcContext。這部分的實現和 java 版 seata 有很大的不一樣，感興趣的能夠深刻了解一下。具體代碼見 getty_session_manager.go。

至此，咱們就分析完了 seata-golang 整個 RPC 通訊模型的機制。

3、seata-golang 的將來

seata-golang  從今年 4 月份開始開發，到 8 月份基本實現和 java 版 seata 1.2 協議的互通，對 mysql 數據庫實現了 AT 模式（自動協調分佈式事務的提交回滾），實現了 TCC 模式，TC 端使用 mysql 存儲數據，使 TC 變成一個無狀態應用支持高可用部署。下圖展現了 AT 模式的原理：

3.png

後續，還有許多工做能夠作，好比：對註冊中心的支持、對配置中心的支持、和 java 版 seata 1.4 的協議互通、其餘數據庫的支持、raft transaction coordinator 的實現等，但願對分佈式事務問題感興趣的開發者能夠加入進來一塊兒來打造一個完善的 golang 的分佈式事務框架。

參考資料

做者簡介

劉曉敏 (GitHubID dk-lockdown)，目前就任於 h3c 成都分公司，擅長使用 Java/Go 語言，在雲原生和微服務相關技術方向均有涉獵，目前專攻分佈式事務。

於雨(github @AlexStocks)，dubbo-go 項目和社區負責人，一個有十多年服務端基礎架構研發一線工做經驗的程序員，陸續參與改進過 Muduo/Pika/Dubbo/Sentinel-go 等知名項目，目前在螞蟻金服可信原生部從事容器編排和 service mesh 工做。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程