未分类

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 1. 引言 在萌叔的文章玩转consul(3)–大规模部署的性能开销定量分析 中，探讨了consul支持大规模集群可能出现性能瓶颈。引出可以通过拆分consul集群或者逻辑切分DC的方法，来降低consul的压力。本文将展开说明一下。 2. 解释 2.1 单个DC的情况 假设服务A依赖服务B(调用服务B), 服务A和服务B各有200个实例, 服务B注册到consul集群上， 服务A通过consul集群发现服务B。 如果服务B做滚动发布，考虑最坏的情况，每次发布一个实例， consul总共发出的通知数为 200 * 200 = 40,000 2.2 2个DC的情况 拆分成2个DC后，单个DC中，服务A和服务B各有100个实例，限制不允许跨DC调用。如果服务B做滚动发布，考虑最坏的情况，DC1和DC2同时发布，每次发布一个实例，consul总共发出的通知数为 100 * 100 + 100 * 100 = 20,000 3. 总结 当服务的规模很大时，即使拆分成2个DC，服务整体的高可用能力不会受到影响。对比2.1和2.2，consul集群的压力减少了一半。另外咱们之前提过，单个DC中consul集群(server)的规模不宜过大，否则会影响日志复制、还有选主的速度。拆分到2个DC后，原有的consul集群也被拆分成了2个集群，单个DC中consul集群的规模下降了，这样还给我们留出扩容的余地。 后记 2022年1月29日 用单元化的思想（见参考资料1、2）来解决注册中心压力大的问题与我文章中提到的，划分逻辑DC的思想不谋而合。 参考资料 1.单元化架构解决了什么问题 2.单元化介绍 请我喝瓶饮料

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 1. 前言 读者们可能使用过 fullstorydev/grpcurl 来对grpc服务进行调用调试。 假定有一个简单的grpc的服务 main.go package main import ( "context" "log" "net" "google.golang.org/grpc" pb "google.golang.org/grpc/examples/helloworld/helloworld" "google.golang.org/grpc/reflection" ) const ( port = ":50051" ) // server is used to implement helloworld.GreeterServer. type server struct { pb.UnimplementedGreeterServer } // SayHello implements helloworld.GreeterServer func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) { log.Printf("Received: %v", in.GetName()) return &pb.HelloReply{Message: "Hello " + in.GetName()}, nil } func main() { lis, err := net.Listen("tcp", port) if err != nil { log.Fatalf("failed to listen: %v", err) } s := grpc.NewServer() helloService := &server{} pb.RegisterGreeterServer(s, helloService) // 注册反射服务 // Register reflection service on gRPC server. reflection.Register(s) if err := s.Serve(lis); err != nil { log.Fatalf("failed to serve: %v", err) } } 使用grpcurl来请求grpc服务 ...

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 1. 引言 假如你正在使用go mod管理某个项目的代码库依赖 case1: 某个项目的tag是超过v1版本的，你该怎么定义module case2: 某个项目要同时进行2个版本的开发，v2和v3 你该怎么定义module 2. 问题 为了验证效果，萌叔创建了项目 vearne/mod-multi-version 这个代码库只有一个文件 package mmv import "fmt" const Versoin = "v1.0.0" func PrintVersion(){ fmt.Println("version:", Versoin) } 其它项目使用vearne/mod-multi-version package main import ( "github.com/vearne/mod-multi-version" ) func main() { mmv.PrintVersion() } 代码库的tag列表中包含以下tag v1.0.0 v1.0.2 v2.0.1-Alpha v2.0.5 v2.0.6 v3.0.0 v3.0.1 但如果你试图拉取v2.0.1-Alpha ╰─$ go get github.com/vearne/mod-multi-version@v2.0.1-Alpha go get: github.com/vearne/mod-multi-version@v2.0.1-Alpha: invalid version: module contains a go.mod file, so major version must be compatible: should be v0 or v1, not v2 错误提示主版本号只能是v0或者v1 ...

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 1. 引言 kt-connect是阿里开源的k8s的调试工具，它的作用类似于VPN，能够打通k8s集群和本地的网络。 传送门: alibaba/kt-connect 它有3种模式 Connect 本地网络直接访问k8s集群网络 Exchange 转发集群流量到本地 Service Mesh 支持 另外它提供了一个Dashboard可以查看k8s集群内的所有可访问的service资源以及正在进行调试的Connect和Exchange数量, 用处不大。 2. 使用介绍 这里萌叔只简单介绍Connect和Exchange2种模式，更详细的使用说明见参考资料1 2.1 Connect模式 sudo ktctl -i ik8share/kt-connect-shadow:stable connect 注意: kt-connect 依赖sshuttle, 且运行时必须拥有root权限。另外sshuttle 又依赖了iptables(linux操作系统), ptctl(macOS) -i 参数指定镜像的地址 这里ik8share/kt-connect-shadow:stable是镜像的名字，阿里默认提供的镜像地址rdc-incubator/kt-connect-shadow在萌叔的测试k8s集群中无法正常拉取。这里提供了一个docker hub的镜像地址。 2.2 Exchange模式 sudo ktctl -n test -i ik8share/kt-connect-shadow:stable exchange dm-backend-v0-0-1 --expose 3000 注意: 这里的dm-backend-v0-0-1 是k8s集群中Deployment资源的名称。 该命令会将所有发往dm-backend-v0-0-1所属Pod的3000端口的请求都转发到本地的3000端口上。 总结：Connect和Exchange 模式都是单向的，一个是从集群外部到集群内部，一个是从集群内部到集群外部。 3. 原理和实现 kt-connect设计巧妙，且最大限度的避免了重复发明轮子，值得称赞。 3.1 Connect模式 我们先来看看Connect模式要达到的目标 假定 Namespace: test Service: sv-backend-v0-0-1 Pod的IP: 172.20.1.29 ...

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 1. 前言 KCP协议的很多东西都是脱胎于TCP协议，所以他们在思想和实现上是完全相通的。xtaci/kcp-go 包含FEC也不过4000多行代码，skywind3000/kcp 主要是C/C++的代码，也就2000多行，萌叔建议大家都去阅读下源码。 慢启动、拥塞避免、拥塞发生、快速重传，这些概念都非常唬人，但看完代码你会发现不过尔尔。 在开始正式的文章之前，萌叔打算问几个问题？ 流控是为了保护谁？ 在实现中如何体现？ 2. 流控是为了保护谁？ TCP是全双工，这里简化一下，我们只看半双工的情况 Sender发送数据给Receiver 1) Sender中的应用程序把数据写入到本机的发送缓冲区 2) 数据从发送缓冲区写入到链路中，链路可能是由实际的光缆、电缆、多个路由器节点组成。 3）数据从链路转交到Receiver的接收缓冲区 4）数据从接收缓冲区交给Receiver的应用程序 发送缓冲区大小是有限的，它必须被保护起来 Sender和Receiver之间的链路的收发能力也是有限的，且是与网络中的其它节点共享的，因此Link也必须受到保护 接收缓冲区大小也是受限的，它也应该受到保护 3. 在实现中如何体现？ 在实际实现中每一个需要保护的点，都有与之对应的参数，先上结论 3.1 使用发送端的发送窗口(snd_wnd)保护本机的发送缓冲区 3.2 使用拥塞窗口(cwnd)来保护发送端与接收端之间的链路 cwnd是动态变化的值, 算法与TCP协议基本相同 3.3 使用接收端的接收窗口(rmt_wnd, 表示接收窗口的空闲大小）保护接收端的接收缓冲区 rmt_wnd对应KCP协议的wnd, 由接收端汇报 回顾一下KCP协议 0 4 5 6 8 (BYTE) +---------------+---+---+-------+ | conv |cmd|frg| wnd | +---------------+---+---+-------+ 8 | ts | sn | +---------------+---------------+ 16 | una | len | +---------------+---------------+ 24 | | | DATA (optional) | | | +-------------------------------+ 4. 分析一次完整的写入动作 4.1 发送窗口和接收窗口对写入的影响 在KCP中，数据被拆分成Segment后 ...

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 1. 创建协程 创建协程示例 n.waitGroup.Wrap(func() { // do something }) util.WaitGroupWrapper package util import ( "sync" ) type WaitGroupWrapper struct { sync.WaitGroup } func (w *WaitGroupWrapper) Wrap(cb func()) { w.Add(1) go func() { cb() w.Done() }() } 2. 协程池 用于从Channel扫描数据到Client func (n *NSQD) queueScanLoop() { ... for { select { case <-workTicker.C: if len(channels) == 0 { continue } case <-refreshTicker.C: channels = n.channels() // 协程池中的协程的数量是动态变化的 // 理想数量是与channel的数量保持一致 n.resizePool(len(channels), workCh, responseCh, closeCh) continue case <-n.exitChan: goto exit } ... } // resizePool adjusts the size of the pool of queueScanWorker goroutines // // 1 <= pool <= min(num * 0.25, QueueScanWorkerPoolMax) // func (n *NSQD) resizePool(num int, workCh chan *Channel, responseCh chan bool, closeCh chan int) { idealPoolSize := int(float64(num) * 0.25) if idealPoolSize < 1 { idealPoolSize = 1 } else if idealPoolSize > n.getOpts().QueueScanWorkerPoolMax { idealPoolSize = n.getOpts().QueueScanWorkerPoolMax } for { if idealPoolSize == n.poolSize { break } else if idealPoolSize < n.poolSize { // contract closeCh <- 1 n.poolSize-- } else { // expand n.waitGroup.Wrap(func() { n.queueScanWorker(workCh, responseCh, closeCh) }) n.poolSize++ } } } 3. 通讯协议 nsq中数据被分为2类指令型数据，消息型数据, 2种数据类型，格式不相同 ...

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 引言 提到GRPC大家想到的就是 HTTP2 + protobuf。HTTP2能够实现多路复用, protobuf提高数据传输的压缩率。但其实GRPC还可以跟Thrift、JSON等编码方式进行整合。 gRPC lets you use encoders other than Protobuf. It has no dependency on Protobuf and was specially made to work with a wide variety of environments. We can see that with a little extra boilerplate, we can use any encoder we want. While this post only covered JSON, gRPC is compatible with Thrift, Avro, Flatbuffers, Cap’n Proto, and even raw bytes! gRPC lets you be in control of how your data is handled. (We still recommend Protobuf though due to strong backwards compatibility, type checking, and performance it gives you.) ...

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 1. 引子 公司内部有简单的搜索引擎，使用ES搭建。 前两天测试人员问我，为什么同一个查询条件，同一条数据，多次查询。score会发生变化。经过验证，确实存在这种问题，那么这种情况到底怎么产生的呢？ 2. 例子 来造个例子 2.1 创建index curl -XPUT -H "Content-Type: application/json" dev1:9200/test -d ' { "settings": { "index.number_of_replicas": "2", "index.number_of_shards": "1" }, "mappings": { "_default_": { "dynamic_templates": [], "properties": { "brand": { "type": "keyword" } } } } } 2.2 写入数据 第1次执行 insert1.py import requests for i in range(500): url = "http://dev1:9200/test/car/%d" % (i) res = requests.put(url, json={"brand":"buick", "age":i}) print(i, res.status_code) 第2次执行 insert2.py ...

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 1. 前言 笔者连续2篇文章，探讨如何开发一个gin的timeout middleware，但是"百密一疏"啊。仍然考虑的不够周全。 笔者的文章 gin的timeout middleware实现(续) 中实现的程序有2个问题 func Timeout(t time.Duration) gin.HandlerFunc { return func(c *gin.Context) { // sync.Pool buffer := buffpool.GetBuff() blw := &SimplebodyWriter{body: buffer, ResponseWriter: c.Writer} c.Writer = blw // wrap the request context with a timeout ctx, cancel := context.WithTimeout(c.Request.Context(), t) c.Request = c.Request.WithContext(ctx) finish := make(chan struct{}) // 子协程 // ****************注意*************** // 创建的子协程没有recover，存在程序崩溃的风险 go func() { c.Next() finish <- struct{}{} }() // ****************注意*************** select { case <-ctx.Done(): // ****************注意*************** // 子协程和父协程存在同时修改Header的风险 // 由于Header是个map，可能诱发 // fatal error: concurrent map read and map write c.Writer.WriteHeader(http.StatusGatewayTimeout) // ****************注意*************** c.Abort() // 超时发生, 通知子协程退出 cancel() // 如果超时的话，buffer无法主动清除，只能等待GC回收 case <-finish: // 结果只会在主协程中被写入 blw.ResponseWriter.Write(buffer.Bytes()) buffpool.PutBuff(buffer) } } } 2. 解决 main.go ...

版权声明 本站原创文章 由 萌叔 发表 转载请注明 萌叔 | http://vearne.cc 前言 关于Golang中的锁的实现和源码早就文章阐述过了，这里我从实验的角度让大家看看对协程数量对任务执行速度，以及CPU开销的影响 运行环境 CPU: Intel(R) Xeon(R) CPU E5-2650 v4 @ 2.20GHz 8核 内存:16G GO version 1.10 协程数量分别是1, 2, 5, 10, 100, 200下完成1000000000次计数所花 总时间 CPU消耗 CPU峰值 分析 观察图可以发现， 协程数量是1的时候运行总时间反而最短，只耗时2.8秒，协程数量10时，运行总时间最长，耗时21.6秒 协程数量是100时，CPU峰值最大，为530%，协程数量是1是，CPU峰值最小，不到100% 代码 package main import ( "fmt" "sync" "time" ) type LockBox struct{ sync.Mutex Value int } func deal(wpg *sync.WaitGroup, bp *LockBox, count int){ for i:=0;i< count;i++{ bp.Lock() bp.Value++ bp.Unlock() } wpg.Done() } func main() { timeStart := time.Now() //workerCount := 1 //workerCount := 2 workerCount := 200 taskCount := 1000000000 var wg sync.WaitGroup var box LockBox for i:=0;i<workerCount;i++{ wg.Add(1) go deal(&wg, &box, taskCount/workerCount) } wg.Wait() fmt.Println("cost", time.Since(timeStart)) } 请我喝瓶饮料