You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by il...@apache.org on 2021/01/15 03:39:34 UTC
[dubbo-website] branch master updated: dubbo-go blog batch #5
This is an automated email from the ASF dual-hosted git repository.
iluo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/dubbo-website.git
The following commit(s) were added to refs/heads/master by this push:
new 95a92df dubbo-go blog batch #5
95a92df is described below
commit 95a92df0aab3f88f9855aad3b405beabfa22d477
Author: Ian Luo <ia...@gmail.com>
AuthorDate: Fri Jan 15 11:39:06 2021 +0800
dubbo-go blog batch #5
---
content/zh/blog/news/dubbo-go-cloud-native.md | 152 ++++++
content/zh/blog/news/dubbo-go-codewalk-2.md | 559 ++++++++++++++++++++++
content/zh/blog/news/dubbo-go-seata.md | 659 ++++++++++++++++++++++++++
static/imgs/blog/dubbo-go/3.0-plan/p1.webp | Bin 0 -> 55426 bytes
static/imgs/blog/dubbo-go/3.0-plan/p2.webp | Bin 0 -> 26698 bytes
static/imgs/blog/dubbo-go/3.0-plan/p3.webp | Bin 0 -> 45754 bytes
static/imgs/blog/dubbo-go/3.0-plan/p4.webp | Bin 0 -> 26536 bytes
static/imgs/blog/dubbo-go/3.0-plan/p5.webp | Bin 0 -> 55550 bytes
static/imgs/blog/dubbo-go/3.0-plan/p6.webp | Bin 0 -> 6544 bytes
static/imgs/blog/dubbo-go/3.0-plan/p7.webp | Bin 0 -> 10782 bytes
static/imgs/blog/dubbo-go/code2/p1.png | Bin 0 -> 155856 bytes
static/imgs/blog/dubbo-go/code2/p2.png | Bin 0 -> 106127 bytes
static/imgs/blog/dubbo-go/code2/p3.png | Bin 0 -> 35147 bytes
static/imgs/blog/dubbo-go/code2/p4.png | Bin 0 -> 110362 bytes
static/imgs/blog/dubbo-go/code2/p5.png | Bin 0 -> 82163 bytes
static/imgs/blog/dubbo-go/code2/p6.png | Bin 0 -> 22061 bytes
static/imgs/blog/dubbo-go/code2/p7.png | Bin 0 -> 24077 bytes
static/imgs/blog/dubbo-go/code2/p8.png | Bin 0 -> 4754 bytes
static/imgs/blog/dubbo-go/code2/p9.png | Bin 0 -> 141104 bytes
static/imgs/blog/dubbo-go/seata/p1.webp | Bin 0 -> 32696 bytes
static/imgs/blog/dubbo-go/seata/p2.webp | Bin 0 -> 12302 bytes
static/imgs/blog/dubbo-go/seata/p3.webp | Bin 0 -> 28034 bytes
22 files changed, 1370 insertions(+)
diff --git a/content/zh/blog/news/dubbo-go-cloud-native.md b/content/zh/blog/news/dubbo-go-cloud-native.md
new file mode 100644
index 0000000..87c9e69
--- /dev/null
+++ b/content/zh/blog/news/dubbo-go-cloud-native.md
@@ -0,0 +1,152 @@
+---
+title: "dubbogo 3.0：牵手 gRPC 走向云原生时代"
+linkTitle: "dubbogo 3.0：牵手 gRPC 走向云原生时代"
+date: 2021-01-15
+description: >
+ 本文介绍了 dubbo-go 3.0 对云原生的支持和规划
+---
+
+自从 2011 年 Dubbo 开源之后，被大量中小公司采用，一直是国内最受欢迎的 RPC 框架。2014 年，由于阿里内部组织架构调整，Dubbo 暂停维护了一段时间，之后随着 Spring Cloud 的面世，两个体系在融合中一起助推了微服务的火热。
+
+不过这世界变化快，自从以 docker 为代表的的容器技术和以 K8s 为代表的容器编排技术登上舞台之后，云原生时代到来了。在云原生时代，不可变的基础设施给原有的中间件带来了不可变的中间件基础设施：gRPC 统一了底层通信层；protobuf 统一了序列化协议；以 envoy + istio 为代表的 service mesh 逐渐统一了服务的控制面与数据面。
+
+dubbogo 的天然使命是：Bridging the gap between Java and Go。保持 Go 应用与 Java 应用互联互通的同时，借助 Go 语言（事实上的第一云原生语言）的优势拥抱云原生时代。dubbogo 社区 2020 年勠力打造三支箭：
+
+- 已经发布的对齐 dubbo 2.7 的 dubbogo v1.5 版本；
+- 近期将要发布的 sidecar 形态的 dubbo-go-proxy 项目；
+- 以及处于进行时的 dubbogo 3.0。
+
+用一句话概括 dubbogo 3.0 即是：新通信协议、新序列化协议、新应用注册模型以及新的服务治理能力！本文主要着重讨论 dubbogo 3.0 的新通信协议和应用级服务注册发现模型。
+
+## dubbogo 3.0 vs gRPC
+
+知己知彼，方能进步。dubbogo 3.0 的通信层改进主要借鉴了 gRPC。
+
+gRPC 协议，简单来说就是 http2 协议的基础之上，增加了特定的协议 header：“grpc-” 开头的 header 字段，采用特定的打解包工具（protobuf）对数据进行序列化，从而实现 RPC 调用。
+
+
+
+众所周知，gRPC 几乎没有服务治理能力，而阿里云现有 dubbo 框架兼具 RPC 和服务治理能力，整体实力不逊于 gRPC。但在“大家都用 gRPC” 这样的背景之下，dubbogo 3.0 的新通信协议就必须**完美兼容 gRPC**，对开发者已部署的服务完全兼容，并在此基础之上延续已有 dubbo 协议和服务治理能力，进而推出一系列新策略：比如 mesh 支持、应用级服务注册等。
+
+
+
+## dubbogo 3.0 vs dubbogo 1.5
+
+目前已有的 dubbo 2.7 协议已经尽可能实现了 gRPC 的支持。开发者可以通过 protoc-gen-dubbo 工具将 pb IDL 协议转换为框架支持的 stub，再借助底层 gRPC conn 的 RPC 过程，将已有的服务治理能力自上而下传递给 gRPC，因此实现了 gRPC 服务的支持。
+
+dubbo-go v1.5.x 也支持 gRPC 的 Stream 调用。和 unary RPC 类似，通过产生框架支持的 stub，在底层 gRPC stream 调用的基础之上，将流式 RPC 的能力和并入框架。但由于 dubbo v2.7.x / dubbo-go v1.5.x 本身并不支持流式调用，所以没有对 gRPC stream 调用的进行上层服务治理支持。
+
+开发者所面临的问题就是：我们在使用 dubbo-go2.7 进行 grpc 协议传输的时候，或多或少不是那么放心。
+
+而即将推出的 dubbo-go 3.0 协议将从根源解决这个问题。
+
+## 协议兼容的三种层次
+
+笔者认为，一款服务框架对于第三方协议的支持可分为三个程度：应用层次、协议层次、传输层次。
+
+一款框架如果在一个协议的 sdk 之上封装接口，可以认为它处于应用层次支持，这样的框架需要遵循下层 sdk 的接口，可扩展性较差。
+
+处于协议层次的框架，从配置层到服务治理层均由本框架提供，而在此之下的协议层到网络传输层均使用某个固定的通信协议，这样的框架可以解决服务治理的问题，但框架本身无法与第三方协议完全适配，如果不适配就会出现对第三方协议支持的削弱，比如上面说到的 dubbo-go 1.5 对 stream rpc 支持的缺陷。
+
+如果想进一步支持更多的第三方协议，需要从传输层下手，真正了解第三方协议的具体字段、所依赖的底层协议（比如 HTTP2）的帧模型和数据流，再开发出与第三方协议完全一致的数据交互模块，作为本框架的底层。这样做的好处是最大程度赋予了协议的可扩展性，可以在兼容已有协议的基础之上，可选地增加开发者需要的字段，从而实现已有协议无法实现的功能，就比如 dubbogo 3.0 将支持的反压策略。
+
+## 基于 HTTP2 的通信流程
+
+gRPC 一次基于 HTTP2 的 unary rpc 调用传输主要流程如下：
+
+- client 发送 Magic 信息：
+ PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n；
+- server 收到并检查是否正确；
+- client 和 server 互相发送 setting 帧，收到后发送 ACK 确认；
+- client 发送 Header 帧，包含 gRPC 协议字段，以 End Headers 作为 Header 结束标志；
+- client 紧接着发送 Data 帧，包含 RPC 调用的 request 信息，以 End Stream 作为 Data 结束标志；
+- server 调用函数获得结果；
+- server 发送 Header 帧，包含 gRPC 协议字段，以 End Headers 作为 Header 结束标志；
+- server 紧接着发送 Data 帧，包含 RPC 调用回传的 response 信息；
+- server 紧接着再次发送 Header 帧，包含 RPC 状态和 message 信息，以 End Stream 作为本次 RPC 调用结束标志。
+
+其中包含 gRPC 调用信息的 HTTP2 Header 帧如下图：
+
+
+
+另外，在 gRPC 的 stream 调用中，可在 server 端回传的过程中发送多次 Data，调用结束后再发送 Header 终止 RPC 过程，并汇报状态信息。
+
+dubbogo 3.0 的通信层将在 HTTP2 通信协议之上采用同样的通信流程，以保证与 gRPC 的底层通信沟通能力。
+
+## dubbogo 3.0 预期通信架构
+
+除了通信协议采用 HTTP2 外，dubbogo 3.0 将采用基于 google protobuf 的 triple 协议【下面称为 dubbo3 协议】作为 dubbogo 3.0 的序列化协议，为 dubbo 将来支持更多的编程语言打下通信协议层面的基础。
+
+目前设计的 dubbogo 3.0 传输模型如下：
+
+
+
+- 为保证同时支持 unary RPC 和 stream RPC，在 server 端和 client 端增加数据流结构，以异步调用的形式完成数据传递；
+- 继续支持原有的 TCP 通信能力；
+- 在 HTTP2 的通信协议之上支持 dubbo3 协议，decode 过程兼容 gRPC 使用的 protobuf，保证与 gRPC 服务打通。
+
+## 应用级服务注册发现
+
+#### 1. 应用级服务注册发现介绍
+
+dubbogo 3.0 使用的新一代服务注册发现体系，将摒弃旧版的“接口级注册发现”，使用“应用级别注册发现”。
+
+简单地说，接口级别注册发现，在注册中心中以 RPC 服务为 key，以实例列表作为 value 来组织数据的，而我们新引入的“应用粒度的服务发现”，它以应用名（Application）作为 key，以这个应用部署的一组实例（Instance）列表作为 value。这带来两点不同：
+
+- 数据映射关系变了，从 RPC Service -> Instance 变为 Application -> Instance；
+- 数据变少了，注册中心没有了 RPC Service 及其相关配置信息。
+
+可以认为，基于应用粒度的模型所存储和推送的数据量是和应用、实例数成正比的，只有当我们的应用数增多或应用的实例数增长时，地址推送压力才会上涨。
+
+而对于基于接口粒度的模型，数据量是和接口数量正相关的，鉴于一个应用通常发布多个接口的现状，其数量级一般是比应用粒度的数十倍。另外一个关键点在于，接口的定义更多的是业务侧的内部行为，接口粒度导致的集群规模评估的不透明，而实例、应用增长都通常是在运维侧的规划之中，可控性较好。
+
+工商银行曾经对这两个模型进行生产测算：应用级服务注册模型可以让注册中心上的数据量变成原来的 1.68%，新模型可以让 zookeeper 轻松至成 10 万级别的服务量和 10 万级别的节点量。
+
+#### 2. 元数据中心同步机制的引入
+
+数据中心的数据量变少所造成的结果，是 RPC 服务相关的数据在注册中心消失了，只有 application - instance 这两个层级的数据。为了保证这部分缺少的 RPC 服务数据仍然能被 Consumer 端正确的感知，我们在 Consumer 和 Provider 间建立了一条单独的通信通道，目前针对元数据同步有两种具体的可选方案，分别是：
+
+- 内建 MetadataService；
+- 独立的元数据中心，通过中细化的元数据集群协调数据。
+
+#### 3. 兼容旧版本 dubbo-go
+
+为了使整个开发流程对老的 dubbo-go 用户更透明，同时避免指定 provider 对可扩展性带来的影响），我们设计了一套 RPC服务到应用名的映射关系，以尝试在 consumer 自动完成 RPC 服务到 provider 应用名的转换。
+
+
+
+这套设计可以让 dubbogo 3.0 中同时保持对 dubbo v2.6.x、dubbo v2.7.x 和 dubbo v3.0.x 三个大版的互联互通。
+
+## 统一路由的支持
+
+路由在概念上可以理解为从已有的所有 IP 地址列表中，根据特定的路由规则，挑选出需要的 ip 地址子集。路由的过程需要根据配置好的路由规则进行筛选，最终取所有路由规则的交集获得结果。多个路由如同流水线一样，形成一条路由链，从所有的地址表中筛选出最终目的地址集合，再通过负载均衡策略选择访问的地址。
+
+#### 1. 路由链
+
+
+
+可以把路由链的逻辑简单理解为 target = rn(...r3(r2(r1(src))))。对于每一个 router 内部的逻辑，可以抽象为输入地址 addrs-in 与 router 中按全量地址 addrs-all 实现切分好的 n 个**互不相交**的地址池 addrs-pool-1 ... addrs-pool-n 按实现定义好的规则取交集作为输出地址。以此类推，完成整个路由链的计算。
+
+
+
+#### 2. failover
+
+在路由规则配置文件中可以配置 failover 字段。在寻找地址失败时可以 failover， 选择其他 subset，并且顺序执行下来，直到找到地址，否则最后报地址找不到异常。
+
+#### 3. 兜底路由
+
+在的路由规则配置中，可以配置一个没有任何条件的 match, 最终的结果是至少会有一个 subset 被选到，以达到地址空保护的作用。
+
+作为 2020 年 dubbogo 社区打造并将在 2021 年初亮出的的三支箭之一，dubbogo 3.0 将带来不同平常且焕然一新的开发体验，敬请广大开发者期待！
+
+如果你有任何疑问，欢迎钉钉扫码加入交流群【钉钉群号 31363295】：
+
+dubbogo 3.0 目前是社区和 dubbo 官方团队-- 阿里中间件团队共同合作开发。
+
+阿里云-中间件团队招募对 dubbo3 (java & go)、dapr、arthas 有兴趣的开发者。可以钉钉联系 northlatitude，或者发送邮件至 beiwei.ly@alibaba-inc.com。
+
+> 作者简介
+>
+> **李志信** (GitHubID LaurenceLiZhixin)，阿里云云原生中间件团队开发工程师，dubbogo 社区开发者，中山大学软件工程专业在校学生，擅长使用 Go 语言，专注于云原生和微服务等技术方向。
+>
+> **于雨**(github @AlexStocks)，dubbo-go 项目和社区负责人，一个有十多年服务端做着基础架构研发一线工作经验的程序员，陆续参与改进过 Muduo/Pika/Dubbo/Sentinel-go 等知名项目，目前在蚂蚁金服可信原生部从事容器编排和 service mesh 工作。
\ No newline at end of file
diff --git a/content/zh/blog/news/dubbo-go-codewalk-2.md b/content/zh/blog/news/dubbo-go-codewalk-2.md
new file mode 100644
index 0000000..5cf54db
--- /dev/null
+++ b/content/zh/blog/news/dubbo-go-codewalk-2.md
@@ -0,0 +1,559 @@
+---
+title: "dubbo-go源码笔记（二）客户端调用过程"
+linkTitle: "dubbo-go源码笔记（二）客户端调用过程"
+date: 2021-01-15
+description: >
+ 随着微服务架构的流行，许多高性能 rpc 框架应运而生，由阿里开源的 dubbo 框架 go 语言版本的 dubbo-go 也成为了众多开发者不错的选择。本文将介绍 dubbo-go 框架的基本使用方法，以及从 export 调用链的角度进行 server 端源码导读，希望能引导读者进一步认识这款框架。
+---
+
+## 前言
+
+有了上一篇文章[《dubbo-go 源码笔记（一）Server服务暴露过程详解》]({{<ref "/blog/news/dubbo-go-codewalk-1.md" >}} "") 的铺垫，可以大致上类比客户端服务类似于服务端启动过程。其中最大的区别是服务端通过zk注册服务，发布自己的ivkURL并订阅事件开启监听；而服务端应该是通过zk注册组件，**拿到需要调用的serviceURL**，**更新invoker**并**重写用户的RPCService**，从而实现对远程过程调用细节的封装。
+
+## 1. 配置文件和客户端源码
+
+#### 1.1 client配置文件
+
+helloworld提供的demo：profiles/client.yaml
+
+```yaml
+registries :
+ "demoZk":
+ protocol: "zookeeper"
+ timeout : "3s"
+ address: "127.0.0.1:2181"
+ username: ""
+ password: ""
+references:
+ "UserProvider":
+ # 可以指定多个registry，使用逗号隔开;不指定默认向所有注册中心注册
+ registry: "demoZk"
+ protocol : "dubbo"
+ interface : "com.ikurento.user.UserProvider"
+ cluster: "failover"
+ methods :
+ - name: "GetUser"
+ retries: 3
+```
+
+可看到配置文件与之前讨论过的server端非常类似，其refrences部分字段就是对当前服务要主调的服务的配置，其中详细说明了调用协议、注册协议、接口id、调用方法、集群策略等，这些配置都会在之后与注册组件交互，重写ivk、调用的过程中使用到。
+
+#### 1.2 客户端使用框架源码
+
+user.go
+
+```go
+func init() {
+ config.SetConsumerService(userProvider)
+ hessian.RegisterPOJO(&User{})
+}
+```
+
+main.go
+
+```go
+func main() {
+ hessian.RegisterPOJO(&User{})
+ config.Load()
+ time.Sleep(3e9)
+ println("\n\n\nstart to test dubbo")
+ user := &User{}
+ err := userProvider.GetUser(context.TODO(), []interface{}{"A001"}, user)
+ if err != nil {
+ panic(err)
+ }
+ println("response result: %v\n", user)
+ initSignal()
+}
+```
+
+官网提供的helloworld demo的源码。可看到与服务端类似，在user.go内注册了rpc-service，以及需要rpc传输的结构体user。
+
+在main函数中，同样调用了config.Load()函数，之后就可以直接通过实现好的rpc-service：userProvider 直接调用对应的功能函数，即可实现rpc调用。
+
+可以猜到，从hessian注册结构、SetConsumerService，到调用函数.GetUser()期间，用户定义的rpc-service也就是userProvider对应的函数被重写，重写后的GetUser函数已经包含了实现了远程调用逻辑的invoker。
+
+接下来，就要通过阅读源码，看看dubbo-go是如何做到的。
+
+## 2. 实现远程过程调用
+
+#### 2.1 加载配置文件
+
+config/config_loader.go :Load()
+
+```go
+// Load Dubbo Init
+func Load() {
+ // init router
+ initRouter()
+ // init the global event dispatcher
+ extension.SetAndInitGlobalDispatcher(GetBaseConfig().EventDispatcherType)
+ // start the metadata report if config set
+ if err := startMetadataReport(GetApplicationConfig().MetadataType, GetBaseConfig().MetadataReportConfig); err != nil {
+ logger.Errorf("Provider starts metadata report error, and the error is {%#v}", err)
+ return
+ }
+ // reference config
+ loadConsumerConfig()
+```
+
+在main函数中调用的config.Load()函数，进而调用了loadConsumerConfig，类似于之前讲到的server端配置读入函数。
+
+在loadConsumerConfig函数中，进行了三步操作：
+
+
+
+1. 检查配置文件并将配置写入内存
+2. **在for循环内部**，依次引用（refer）并且实例化（implement）每个被调reference。
+3. 等待三秒钟所有invoker就绪
+
+其中重要的就是for循环里面的引用和实例化，两步操作，会在接下来展开讨论。
+
+至此，配置已经被写入了框架。
+
+#### 2.2 获取远程Service URL，实现可供调用的invoker
+
+上述的ref.Refer完成的就是这部分的操作。
+
+图（一）
+
+
+
+##### 2.2.1 构造注册url
+
+和server端类似，存在注册url和服务url，dubbo习惯将服务url作为注册url的sub。
+
+config/reference_config.go: Refer()
+
+```go
+/ Refer ...
+func (c *ReferenceConfig) Refer(_ interface{}) {
+ //（一）配置url参数(serviceUrl)，将会作为sub
+ cfgURL := common.NewURLWithOptions(
+ common.WithPath(c.id),
+ common.WithProtocol(c.Protocol),
+ common.WithParams(c.getUrlMap()),
+ common.WithParamsValue(constant.BEAN_NAME_KEY, c.id),
+ )
+ ...
+ // （二）注册地址可以通过url格式给定，也可以通过配置格式给定
+ // 这一步的意义就是配置->提取信息生成URL
+ if c.Url != "" {// 用户给定url信息，可以是点对点的地址，也可以是注册中心的地址
+ // 1. user specified URL, could be peer-to-peer address, or register center's address.
+ urlStrings := gxstrings.RegSplit(c.Url, "\\s*[;]+\\s*")
+ for _, urlStr := range urlStrings {
+ serviceUrl, err := common.NewURL(urlStr)
+ ...
+ }
+ } else {// 配置读入注册中心的信息
+ // assemble SubURL from register center's configuration mode
+ // 这是注册url，protocol = registry,包含了zk的用户名、密码、ip等等
+ c.urls = loadRegistries(c.Registry, consumerConfig.Registries, common.CONSUMER)
+ ...
+ // set url to regUrls
+ for _, regUrl := range c.urls {
+ regUrl.SubURL = cfgURL// regUrl的subURl存当前配置url
+ }
+ }
+ //至此，无论通过什么形式，已经拿到了全部的regURL
+ // （三）获取registryProtocol实例，调用其Refer方法，传入新构建好的regURL
+ if len(c.urls) == 1 {
+ // 这一步访问到registry/protocol/protocol.go registryProtocol.Refer
+ // 这里是registry
+ c.invoker = extension.GetProtocol(c.urls[0].Protocol).Refer(*c.urls[0])
+ } else {
+ // 如果有多个注册中心，即有多个invoker,则采取集群策略
+ invokers := make([]protocol.Invoker, 0, len(c.urls))
+ ...
+ }
+```
+
+
+
+这个函数中，已经处理完从Register配置到RegisterURL的转换,即图（一）中部分：
+
+
+
+接下来，已经拿到的url将被传递给RegistryProtocol，进一步refer。
+
+##### 2.2.2 registryProtocol获取到zkRegistry实例，进一步Refer
+
+registry/protocol/protocol.go: Refer
+
+```go
+// Refer provider service from registry center
+// 拿到的是配置文件registries的url，他能够生成一个invoker = 指向目的addr，以供客户端直接调用。
+func (proto *registryProtocol) Refer(url common.URL) protocol.Invoker {
+ var registryUrl = url
+ // 这里拿到的是referenceConfig，serviceUrl里面包含了Reference的所有信息，包含interfaceName、method等等
+ var serviceUrl = registryUrl.SubURL
+ if registryUrl.Protocol == constant.REGISTRY_PROTOCOL {// registryUrl.Proto = "registry"
+ protocol := registryUrl.GetParam(constant.REGISTRY_KEY, "")
+ registryUrl.Protocol = protocol//替换成了具体的值，比如"zookeeper"
+ }
+ // 接口对象
+ var reg registry.Registry
+ // （一）实例化接口对象，缓存策略
+ if regI, loaded := proto.registries.Load(registryUrl.Key()); !loaded {
+ // 缓存中不存在当前registry，新建一个reg
+ reg = getRegistry(®istryUrl)
+ // 缓存起来
+ proto.registries.Store(registryUrl.Key(), reg)
+ } else {
+ reg = regI.(registry.Registry)
+ }
+ // 到这里，获取到了reg实例 zookeeper的registry
+ //（二）根据Register的实例zkRegistry和传入的regURL新建一个directory
+ // 这一步存在复杂的异步逻辑，从注册中心拿到了目的service的真实addr，获取了invoker并放入directory，
+ // 这一步将在下面详细给出步骤
+ // new registry directory for store service url from registry
+ directory, err := extension.GetDefaultRegistryDirectory(®istryUrl, reg)
+ if err != nil {
+ logger.Errorf("consumer service %v create registry directory error, error message is %s, and will return nil invoker!",
+ serviceUrl.String(), err.Error())
+ return nil
+ }
+ // （三）DoRegister 在zk上注册当前client service
+ err = reg.Register(*serviceUrl)
+ if err != nil {
+ logger.Errorf("consumer service %v register registry %v error, error message is %s",
+ serviceUrl.String(), registryUrl.String(), err.Error())
+ }
+ // （四）new cluster invoker，将directory写入集群，获得具有集群策略的invoker
+ cluster := extension.GetCluster(serviceUrl.GetParam(constant.CLUSTER_KEY, constant.DEFAULT_CLUSTER))
+ invoker := cluster.Join(directory)
+ // invoker保存
+ proto.invokers = append(proto.invokers, invoker)
+ return invoker
+}
+```
+
+可详细阅读上述注释，这个函数完成了从url到invoker的全部过程
+
+（一）首先获得Registry对象，默认是之前实例化的zkRegistry，和之前server获取Registry的处理很类似。
+（二）通过构造一个新的directory，异步拿到之前在zk上注册的server端信息，生成invoker
+（三）在zk上注册当前service
+（四）集群策略，获得最终invoker
+
+这一步完成了图（一）中所有余下的绝大多数操作，接下来就需要详细的查看directory的构造过程：
+
+##### 2.2.3 构造directory（包含较复杂的异步操作）
+
+
+
+图（二）
+
+上述的 `extension.GetDefaultRegistryDirectory(®istryUrl, reg)`函数，本质上调用了已经注册好的`NewRegistryDirectory`函数:
+
+registry/directory/directory.go: NewRegistryDirectory()
+
+```go
+// NewRegistryDirectory will create a new RegistryDirectory
+// 这个函数作为default注册在extension上面
+// url为注册url，reg为zookeeper registry
+func NewRegistryDirectory(url *common.URL, registry registry.Registry) (cluster.Directory, error) {
+ if url.SubURL == nil {
+ return nil, perrors.Errorf("url is invalid, suburl can not be nil")
+ }
+ dir := &RegistryDirectory{
+ BaseDirectory: directory.NewBaseDirectory(url),
+ cacheInvokers: []protocol.Invoker{},
+ cacheInvokersMap: &sync.Map{},
+ serviceType: url.SubURL.Service(),
+ registry: registry,
+ }
+ dir.consumerConfigurationListener = newConsumerConfigurationListener(dir)
+ go dir.subscribe(url.SubURL)
+ return dir, nil
+}
+```
+
+首先构造了一个注册directory，开启携程调用其subscribe函数，传入serviceURL。
+
+这个directory目前包含了对应的zkRegistry，以及传入的URL，他cacheInvokers的部分是空的。
+
+进入dir.subscribe(url.SubURL)这个异步函数：
+
+registry/directory/directory.go: subscribe()
+
+```go
+// subscribe from registry
+func (dir *RegistryDirectory) subscribe(url *common.URL) {
+ // 增加两个监听，
+ dir.consumerConfigurationListener.addNotifyListener(dir)
+ dir.referenceConfigurationListener = newReferenceConfigurationListener(dir, url)
+ // subscribe调用
+ dir.registry.Subscribe(url, dir)
+}
+```
+
+重点来了，他调用了zkRegistry的Subscribe方法,与此同时将自己作为ConfigListener传入
+
+> 我认为这种传入listener的设计模式非常值得学习，而且很有java的味道。
+>
+> 针对等待zk返回订阅信息这样的异步操作，需要传入一个Listener，这个Listener需要实现Notify方法，进而在作为参数传入内部之后，可以被异步地调用Notify，将内部触发的异步事件“传递出来”，再进一步处理加工。
+>
+> 层层的Listener事件链，能将传入的原始serviceURL通过zkConn发送给zk服务，获取到服务端注册好的url对应的二进制信息。
+>
+> 而Notify回调链，则将这串byte[]一步一步解析、加工；以事件的形式向外传递，最终落到directory上的时候，已经是成型的newInvokers了。
+>
+> 具体细节不再以源码形式展示，可参照上图查阅源码。
+
+至此已经拿到了server端注册好的真实invoker。
+
+完成了图（一）中的部分：
+
+
+
+##### 2.2.4 构造带有集群策略的clusterinvoker
+
+经过上述操作，已经拿到了server端Invokers，放入了directory的cacheinvokers数组里面缓存。
+
+后续的操作对应本文2.2.2的第四步，由directory生成带有特性集群策略的invoker
+
+```go
+// （四）new cluster invoker，将directory写入集群，获得具有集群策略的invoker
+ cluster := extension.GetCluster(serviceUrl.GetParam(constant.CLUSTER_KEY, constant.DEFAULT_CLUSTER))
+ invoker := cluster.Join(directory)
+123
+```
+
+Join函数的实现就是如下函数：
+
+cluster/cluster_impl/failover_cluster_invokers.go: newFailoverClusterInvoker()
+
+```go
+func newFailoverClusterInvoker(directory cluster.Directory) protocol.Invoker {
+ return &failoverClusterInvoker{
+ baseClusterInvoker: newBaseClusterInvoker(directory),
+ }
+}
+12345
+```
+
+dubbo-go框架默认选择failover策略，既然返回了一个invoker，我们查看一下failoverClusterInvoker的Invoker方法，看他是如何将集群策略封装到Invoker函数内部的：
+
+cluster/cluster_impl/failover_cluster_invokers.go: Invoker()
+
+```go
+// Invoker 函数
+func (invoker *failoverClusterInvoker) Invoke(ctx context.Context, invocation protocol.Invocation) protocol.Result {
+ ...
+ //调用List方法拿到directory缓存的所有invokers
+ invokers := invoker.directory.List(invocation)
+ if err := invoker.checkInvokers(invokers, invocation); err != nil {// 检查是否可以实现调用
+ return &protocol.RPCResult{Err: err}
+ }
+ // 获取来自用户方向传入的
+ methodName := invocation.MethodName()
+ retries := getRetries(invokers, methodName)
+ loadBalance := getLoadBalance(invokers[0], invocation)
+ for i := 0; i <= retries; i++ {
+ // 重要！这里是集群策略的体现，失败后重试！
+ //Reselect before retry to avoid a change of candidate `invokers`.
+ //NOTE: if `invokers` changed, then `invoked` also lose accuracy.
+ if i > 0 {
+ if err := invoker.checkWhetherDestroyed(); err != nil {
+ return &protocol.RPCResult{Err: err}
+ }
+ invokers = invoker.directory.List(invocation)
+ if err := invoker.checkInvokers(invokers, invocation); err != nil {
+ return &protocol.RPCResult{Err: err}
+ }
+ }
+ // 这里是负载均衡策略的体现！选择特定ivk进行调用。
+ ivk := invoker.doSelect(loadBalance, invocation, invokers, invoked)
+ if ivk == nil {
+ continue
+ }
+ invoked = append(invoked, ivk)
+ //DO INVOKE
+ result = ivk.Invoke(ctx, invocation)
+ if result.Error() != nil {
+ providers = append(providers, ivk.GetUrl().Key())
+ continue
+ }
+ return result
+ }
+ ...
+}
+```
+
+> 看了很多Invoke函数的实现，所有类似的Invoker函数都包含两个方向，一个是用户方向的invcation，一个是函数方向的底层invokers。
+>
+> 而集群策略的invoke函数本身作为接线员，把invocation一步步解析，根据调用需求和集群策略，选择特定的invoker来执行
+>
+> proxy函数也是这样，一个是用户方向的ins[] reflect.Type, 一个是函数方向的invoker。
+>
+> proxy函数负责将ins转换为invocation，调用对应invoker的invoker函数，实现连通。
+>
+> 而出于这样的设计，可以在一步步Invoker封装的过程中，每个Invoker只关心自己负责操作的部分，从而使整个调用栈解耦。
+>
+> 妙啊！！！
+
+至此，我们理解了failoverClusterInvoker 的Invoke函数实现，也正是和这个集群策略Invoker被返回，接受来自上方的调用。
+
+已完成图（一）中的：
+
+
+
+##### 2.2.5 在zookeeper上注册当前client
+
+拿到invokers后，可以回到：
+
+config/refrence_config.go: Refer()函数了。
+
+```go
+ if len(c.urls) == 1 {
+ // 这一步访问到registry/protocol/protocol.go registryProtocol.Refer
+ c.invoker = extension.GetProtocol(c.urls[0].Protocol).Refer(*c.urls[0])
+ // （一）拿到了真实的invokers
+ } else {
+ // 如果有多个注册中心，即有多个invoker,则采取集群策略
+ invokers := make([]protocol.Invoker, 0, len(c.urls))
+ ...
+ cluster := extension.GetCluster(hitClu)
+ // If 'zone-aware' policy select, the invoker wrap sequence would be:
+ // ZoneAwareClusterInvoker(StaticDirectory) ->
+ // FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
+ c.invoker = cluster.Join(directory.NewStaticDirectory(invokers))
+ }
+ // （二）create proxy，为函数配置代理
+ if c.Async {
+ callback := GetCallback(c.id)
+ c.pxy = extension.GetProxyFactory(consumerConfig.ProxyFactory).GetAsyncProxy(c.invoker, callback, cfgURL)
+ } else {
+ // 这里c.invoker已经是目的addr了
+ c.pxy = extension.GetProxyFactory(consumerConfig.ProxyFactory).GetProxy(c.invoker, cfgURL)
+ }
+```
+
+我们有了可以打通的invokers，但还不能直接调用，因为invoker的入参是invocation，而调用函数使用的是具体的参数列表。需要通过一层proxy来规范入参和出参。
+
+接下来新建一个默认proxy，放置在c.proxy内，以供后续使用
+
+至此，完成了图（一）中最后的操作
+
+
+
+### 2.3 将调用逻辑以代理函数的形式写入rpc-service
+
+上面完成了config.Refer操作
+
+回到config/config_loader.go: loadConsumerConfig()
+
+
+
+下一个重要的函数是Implement，他完的操作较为简单：旨在使用上面生成的c.proxy代理，链接用户自己定义的rpcService到clusterInvoker的信息传输。
+
+函数较长，只选取了重要的部分:
+
+common/proxy/proxy.go: Implement()
+
+```go
+// Implement
+// proxy implement
+// In consumer, RPCService like:
+// type XxxProvider struct {
+// Yyy func(ctx context.Context, args []interface{}, rsp *Zzz) error
+// }
+// Implement 实现的过程，就是proxy根据函数名和返回值，通过调用invoker 构造出拥有远程调用逻辑的代理函数
+// 将当前rpc所有可供调用的函数注册到proxy.rpc内
+func (p *Proxy) Implement(v common.RPCService) {
+ // makeDubboCallProxy 这是一个构造代理函数，这个函数的返回值是func(in []reflect.Value) []reflect.Value 这样一个函数
+ // 这个被返回的函数是请求实现的载体，由他来发起调用获取结果
+ makeDubboCallProxy := func(methodName string, outs []reflect.Type) func(in []reflect.Value) []reflect.Value {
+ return func(in []reflect.Value) []reflect.Value {
+ // 根据methodName和outs的类型，构造这样一个函数，这个函数能将in 输入的value转换为输出的value
+ // 这个函数具体的实现如下：
+ ...
+ // 目前拿到了 methodName、所有入参的interface和value，出参数reply
+ // （一）根据这些生成一个 rpcinvocation
+ inv = invocation_impl.NewRPCInvocationWithOptions(
+ invocation_impl.WithMethodName(methodName),
+ invocation_impl.WithArguments(inIArr),
+ invocation_impl.WithReply(reply.Interface()),
+ invocation_impl.WithCallBack(p.callBack),
+ invocation_impl.WithParameterValues(inVArr))
+ for k, value := range p.attachments {
+ inv.SetAttachments(k, value)
+ }
+ // add user setAttachment
+ atm := invCtx.Value(constant.AttachmentKey) // 如果传入的ctx里面有attachment，也要写入inv
+ if m, ok := atm.(map[string]string); ok {
+ for k, value := range m {
+ inv.SetAttachments(k, value)
+ }
+ }
+ // 至此构造inv完毕
+ // (二）触发Invoker 之前已经将cluster_invoker放入proxy，使用Invoke方法，通过getty远程过程调用
+ result := p.invoke.Invoke(invCtx, inv)
+ // 如果有attachment，则加入
+ if len(result.Attachments()) > 0 {
+ invCtx = context.WithValue(invCtx, constant.AttachmentKey, result.Attachments())
+ }
+ ...
+ }
+ }
+ numField := valueOfElem.NumField()
+ for i := 0; i < numField; i++ {
+ t := typeOf.Field(i)
+ methodName := t.Tag.Get("dubbo")
+ if methodName == "" {
+ methodName = t.Name
+ }
+ f := valueOfElem.Field(i)
+ if f.Kind() == reflect.Func && f.IsValid() && f.CanSet() { // 针对于每个函数
+ outNum := t.Type.NumOut()
+ // 规定函数输出只能有1/2个
+ if outNum != 1 && outNum != 2 {
+ logger.Warnf("method %s of mtype %v has wrong number of in out parameters %d; needs exactly 1/2",
+ t.Name, t.Type.String(), outNum)
+ continue
+ }
+ // The latest return type of the method must be error.
+ // 规定最后一个返回值一定是error
+ if returnType := t.Type.Out(outNum - 1); returnType != typError {
+ logger.Warnf("the latest return type %s of method %q is not error", returnType, t.Name)
+ continue
+ }
+ // 获取到所有的出参类型，放到数组里
+ var funcOuts = make([]reflect.Type, outNum)
+ for i := 0; i < outNum; i++ {
+ funcOuts[i] = t.Type.Out(i)
+ }
+ // do method proxy here:
+ // （三）调用make函数，传入函数名和返回值，获得能调用远程的proxy，将这个proxy替换掉原来的函数位置
+ f.Set(reflect.MakeFunc(f.Type(), makeDubboCallProxy(methodName, funcOuts)))
+ logger.Debugf("set method [%s]", methodName)
+ }
+ }
+ ...
+}
+```
+
+正如之前所说，proxy的作用是将用户定义的函数参数列表，转化为抽象的invocation传入Invoker，进行调用。
+
+其中已标明有三处较为重要的地方：
+
+1. 在代理函数中实现由参数列表生成Invocation的逻辑
+2. 在代理函数实现调用Invoker的逻辑
+3. 将代理函数替换为原始rpc-service对应函数
+ 至此，也就解决了一开始的问题：
+ client.go: main()
+
+```go
+ config.Load()
+ user := &User{}
+ err := userProvider.GetUser(context.TODO(), []interface{}{"A001"}, user)
+```
+
+这里直接调用用户定义的rpcService的函数GetUser，这里实际调用的是经过重写入的函数代理，所以就能实现远程调用了。
+
+### 3. 从client到server的invoker嵌套链- 小结
+
+在阅读dubbo-go源码的过程中，我能发现一条清晰的invoker-proxy嵌套链，我希望通过图的形式来展现：
+
+
+
+> 作者简介 李志信 (GitHubID LaurenceLiZhixin)，中山大学软件工程专业在校学生，擅长使用 Java/Go 语言，专注于云原生和微服务等技术方向。
diff --git a/content/zh/blog/news/dubbo-go-seata.md b/content/zh/blog/news/dubbo-go-seata.md
new file mode 100644
index 0000000..4d8878f
--- /dev/null
+++ b/content/zh/blog/news/dubbo-go-seata.md
@@ -0,0 +1,659 @@
+---
+title: "分布式事务框架 seata-golang 通信模型详解"
+linkTitle: "分布式事务框架 seata-golang 通信模型详解"
+date: 2021-01-15
+description: >
+ 本文介绍了 seata 的 go 语言客户端通信模型的实现
+---
+
+## 简介
+
+Java 的世界里，大家广泛使用一个高性能网络通信框架 —— netty，很多 RPC 框架都是基于 netty 来实现的。在 golang 的世界里，getty 也是一个类似 netty 的高性能网络通信库。getty 最初由 dubbo-go 项目负责人于雨开发，作为底层通信库在 dubbo-go 中使用。随着 dubbo-go 捐献给 apache 基金会，在社区小伙伴的共同努力下，getty 也最终进入到 apache 这个大家庭，并改名 dubbo-getty。
+
+18 年的时候，我在公司里实践微服务，当时遇到最大的问题就是分布式事务问题。同年，阿里在社区开源他们的分布式事务解决方案，我也很快关注到这个项目，起初还叫 fescar，后来更名 seata。由于我对开源技术很感兴趣，加了很多社区群，当时也很关注 dubbo-go 这个项目，在里面默默潜水。随着对 seata 的了解，逐渐萌生了做一个 go 版本的分布式事务框架的想法。
+
+要做一个 golang 版的分布式事务框架，首先需要解决的一个问题就是如何实现 RPC 通信。dubbo-go 就是摆在眼前很好的一个例子，遂开始研究 dubbo-go 的底层 getty。
+
+## 如何基于 getty 实现 RPC 通信
+
+getty 框架的整体模型图如下：
+
+
+
+下面结合相关代码，详述 seata-golang 的 RPC 通信过程。
+
+### 1. 建立连接
+
+实现 RPC 通信，首先要建立网络连接，这里先从 client.go 开始看起。
+
+```go
+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()` 方法：
+
+```go
+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()` 方法：
+
+```go
+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。
+
+### 2. 收发报文
+
+那它是怎么收发报文的呢，我们回到 connection 方法接着往下看，有这样一行 `ss.(*session).run()`，在这行代码之后，代码都是很简单的操作，我们猜测这行代码运行的逻辑里面一定包含收发报文的逻辑，接着进入 `ru``n()` 方法：
+
+```go
+func (s *session) run() {
+ // 省略部分代码
+
+ go s.handleLoop()
+ go s.handlePackage()
+}
+```
+
+这里起了两个 goroutine：`handleLoop` 和 `handlePackage`，看字面意思符合我们的猜想，进入 `handleLoop()` 方法：
+
+```go
+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()` 方法：
+
+```go
+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()` 方法：
+
+```go
+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)` 这个方法，接着往下分析：
+
+```go
+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 的定义：
+
+```go
+// 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)`：
+
+```go
+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 ：
+
+```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
+}
+```
+
+```go
+// 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 通信。
+
+### 1）编解码协议实现
+
+下面是 seata 协议的定义：
+
+
+
+在 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
+ }
+}
+```
+
+### 2）Client 端实现
+
+再来看 client 端 `EventListener` 的实现 `RpcRemotingClient`：
+
+```go
+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)` 的逻辑将在下文中分析。
+
+### 3）Server 端 Transaction Coordinator 实现
+
+代码见 `DefaultCoordinator`：
+
+```go
+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)` 方法，将按照消息的类型码路由到具体的逻辑当中：
+
+```go
+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）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 来标识一个连接信息：
+
+```go
+type RpcContext struct {
+ Version string
+ TransactionServiceGroup string
+ ClientRole meta.TransactionRole
+ ApplicationId string
+ ClientId string
+ ResourceSets *model.Set
+ Session getty.Session
+}
+```
+
+当收到事务消息时，我们需要构造这样一个 RpcContext 供后续事务处理逻辑使用。所以，我们会构造下列 map 来缓存映射关系：
+
+```go
+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 通信模型的机制。
+
+### seata-golang 的未来
+
+seata-golang 从今年 4 月份开始开发，到 8 月份基本实现和 java 版 seata 1.2 协议的互通，对 mysql 数据库实现了 AT 模式（自动协调分布式事务的提交回滚），实现了 TCC 模式，TC 端使用 mysql 存储数据，使 TC 变成一个无状态应用支持高可用部署。下图展示了 AT 模式的原理：
+
+
+
+后续，还有许多工作可以做，比如：对注册中心的支持、对配置中心的支持、和 java 版 seata 1.4 的协议互通、其他数据库的支持、raft transaction coordinator 的实现等，希望对分布式事务问题感兴趣的开发者可以加入进来一起来打造一个完善的 golang 的分布式事务框架。如果你有任何疑问，欢迎钉钉扫码加入交流群【钉钉群号 33069364】：
+
+另外，欢迎对 dubbogo 感兴趣的朋友到 dubbogo 社区钉钉群（钉钉群号 31363295）沟通 dubbogo 技术问题。
+
+### 参考资料
+
+- seata 官方: https://seata.io
+- java 版 seata: https://github.com/seata/seata
+- seata-golang 项目地址: https://github.com/opentrx/seata-golang
+- seata-golang go 夜读 b 站分享: https://www.bilibili.com/video/BV1oz411e72T
+
+> 作者简介
+>
+> **刘晓敏** (GitHubID dk-lockdown)，目前就职于 h3c 成都分公司，擅长使用 Java/Go 语言，在云原生和微服务相关技术方向均有涉猎，目前专攻分布式事务。
+> **于雨**(（github @AlexStocks），dubbo-go 项目和社区负责人，一个有十多年服务端基础架构研发一线工作经验的程序员，陆续参与改进过 Muduo/Pika/Dubbo/Sentinel-go 等知名项目，目前在蚂蚁金服可信原生部从事容器编排和 service mesh 工作。
\ No newline at end of file
diff --git a/static/imgs/blog/dubbo-go/3.0-plan/p1.webp b/static/imgs/blog/dubbo-go/3.0-plan/p1.webp
new file mode 100644
index 0000000..0d256b6
Binary files /dev/null and b/static/imgs/blog/dubbo-go/3.0-plan/p1.webp differ
diff --git a/static/imgs/blog/dubbo-go/3.0-plan/p2.webp b/static/imgs/blog/dubbo-go/3.0-plan/p2.webp
new file mode 100644
index 0000000..ca0ce36
Binary files /dev/null and b/static/imgs/blog/dubbo-go/3.0-plan/p2.webp differ
diff --git a/static/imgs/blog/dubbo-go/3.0-plan/p3.webp b/static/imgs/blog/dubbo-go/3.0-plan/p3.webp
new file mode 100644
index 0000000..12df591
Binary files /dev/null and b/static/imgs/blog/dubbo-go/3.0-plan/p3.webp differ
diff --git a/static/imgs/blog/dubbo-go/3.0-plan/p4.webp b/static/imgs/blog/dubbo-go/3.0-plan/p4.webp
new file mode 100644
index 0000000..5e68554
Binary files /dev/null and b/static/imgs/blog/dubbo-go/3.0-plan/p4.webp differ
diff --git a/static/imgs/blog/dubbo-go/3.0-plan/p5.webp b/static/imgs/blog/dubbo-go/3.0-plan/p5.webp
new file mode 100644
index 0000000..479ad0a
Binary files /dev/null and b/static/imgs/blog/dubbo-go/3.0-plan/p5.webp differ
diff --git a/static/imgs/blog/dubbo-go/3.0-plan/p6.webp b/static/imgs/blog/dubbo-go/3.0-plan/p6.webp
new file mode 100644
index 0000000..c306c35
Binary files /dev/null and b/static/imgs/blog/dubbo-go/3.0-plan/p6.webp differ
diff --git a/static/imgs/blog/dubbo-go/3.0-plan/p7.webp b/static/imgs/blog/dubbo-go/3.0-plan/p7.webp
new file mode 100644
index 0000000..70216bc
Binary files /dev/null and b/static/imgs/blog/dubbo-go/3.0-plan/p7.webp differ
diff --git a/static/imgs/blog/dubbo-go/code2/p1.png b/static/imgs/blog/dubbo-go/code2/p1.png
new file mode 100644
index 0000000..f2a6d4b
Binary files /dev/null and b/static/imgs/blog/dubbo-go/code2/p1.png differ
diff --git a/static/imgs/blog/dubbo-go/code2/p2.png b/static/imgs/blog/dubbo-go/code2/p2.png
new file mode 100644
index 0000000..7522ba0
Binary files /dev/null and b/static/imgs/blog/dubbo-go/code2/p2.png differ
diff --git a/static/imgs/blog/dubbo-go/code2/p3.png b/static/imgs/blog/dubbo-go/code2/p3.png
new file mode 100644
index 0000000..eb48d7a
Binary files /dev/null and b/static/imgs/blog/dubbo-go/code2/p3.png differ
diff --git a/static/imgs/blog/dubbo-go/code2/p4.png b/static/imgs/blog/dubbo-go/code2/p4.png
new file mode 100644
index 0000000..27ac293
Binary files /dev/null and b/static/imgs/blog/dubbo-go/code2/p4.png differ
diff --git a/static/imgs/blog/dubbo-go/code2/p5.png b/static/imgs/blog/dubbo-go/code2/p5.png
new file mode 100644
index 0000000..5e18909
Binary files /dev/null and b/static/imgs/blog/dubbo-go/code2/p5.png differ
diff --git a/static/imgs/blog/dubbo-go/code2/p6.png b/static/imgs/blog/dubbo-go/code2/p6.png
new file mode 100644
index 0000000..5bb33ce
Binary files /dev/null and b/static/imgs/blog/dubbo-go/code2/p6.png differ
diff --git a/static/imgs/blog/dubbo-go/code2/p7.png b/static/imgs/blog/dubbo-go/code2/p7.png
new file mode 100644
index 0000000..4ba3728
Binary files /dev/null and b/static/imgs/blog/dubbo-go/code2/p7.png differ
diff --git a/static/imgs/blog/dubbo-go/code2/p8.png b/static/imgs/blog/dubbo-go/code2/p8.png
new file mode 100644
index 0000000..d35e906
Binary files /dev/null and b/static/imgs/blog/dubbo-go/code2/p8.png differ
diff --git a/static/imgs/blog/dubbo-go/code2/p9.png b/static/imgs/blog/dubbo-go/code2/p9.png
new file mode 100644
index 0000000..d079fb0
Binary files /dev/null and b/static/imgs/blog/dubbo-go/code2/p9.png differ
diff --git a/static/imgs/blog/dubbo-go/seata/p1.webp b/static/imgs/blog/dubbo-go/seata/p1.webp
new file mode 100644
index 0000000..88809ba
Binary files /dev/null and b/static/imgs/blog/dubbo-go/seata/p1.webp differ
diff --git a/static/imgs/blog/dubbo-go/seata/p2.webp b/static/imgs/blog/dubbo-go/seata/p2.webp
new file mode 100644
index 0000000..51c5e2a
Binary files /dev/null and b/static/imgs/blog/dubbo-go/seata/p2.webp differ
diff --git a/static/imgs/blog/dubbo-go/seata/p3.webp b/static/imgs/blog/dubbo-go/seata/p3.webp
new file mode 100644
index 0000000..74a43ae
Binary files /dev/null and b/static/imgs/blog/dubbo-go/seata/p3.webp differ