Posts

gRPC阅读（2）—— 客户端

启动客户端客户端的启动也是三部曲：初始化grpc.ClientConn 创建service对应的Client（比如codegen生成的GreeterClient）发起rpc调用第二步比较简单，只是把ClientConn作为GreeterClient的成员变量，重点分析建立连接和RPC调用初始化ClientConn 初始化ClientConn做了很多准备工作，包括但不限于：应用选项（DialOption）构建拦截器调用链（Interceptor）决定使用什么resolver（resolver.Builder）检查传输层凭证，比如TLS（TransportCredentials）解析自定义服务端配置（ServerConfig） … 但还有一些配置是在真正发起RPC调用的时候才会被设置和触发，比如重试限流器、RPC配置选择器、RPC负载均衡器等。 func NewClient(target string, opts ...DialOption) (conn *ClientConn, err error) { cc := &ClientConn{ target: target, conns: make(map[*addrConn]struct{}), dopts: defaultDialOptions(), } // 重试限流器 cc.retryThrottler.Store((*retryThrottler)(nil)) // 配置选择器，动态选择每个RPC的调用配置 cc.safeConfigSelector.UpdateConfigSelector(&defaultConfigSelector{nil}) cc.ctx, cc.cancel = context.WithCancel(context.Background()) // options ... // 确定使用哪个resolver（默认为dns） if err := cc.initParsedTargetAndResolverBuilder(); err != nil { return nil, err } // 内部使用的全局perTarget options for _, opt := range globalPerTargetDialOptions { opt.DialOptionForTarget(cc.parsedTarget.URL).apply(&cc.dopts) } // 初始化拦截器调用链 chainUnaryClientInterceptors(cc) chainStreamClientInterceptors(cc) // 验证安全传输，如TLS if err := cc.validateTransportCredentials(); err != nil { return nil, err } // 解析以json格式指定的配置 // 如负载均衡配置、per-RPC方法超时等 if cc.dopts.defaultServiceConfigRawJSON != nil { scpr := parseServiceConfig(*cc.dopts.defaultServiceConfigRawJSON, cc.dopts.maxCallAttempts) if scpr.Err != nil { return nil, fmt.Errorf("%s: %v", invalidDefaultServiceConfigErrPrefix, scpr.Err) } cc.dopts.defaultServiceConfig, _ = scpr.Config.(*ServiceConfig) } // keepalive对服务端探活 cc.mkp = cc.dopts.copts.KeepaliveParams // 获取authority，作为请求头中的:authority字段 if err = cc.initAuthority(); err != nil { return nil, err } // 注册channelz，用于监测grpc的运行 // 可通过http协议访问/grpc/channelz/v1查看grpc的状态 cc.channelzRegistration(target) channelz.Infof(logger, cc.channelz, "parsed dial target is: %#v", cc.parsedTarget) channelz.Infof(logger, cc.channelz, "Channel authority set to %q", cc.authority) // 连接状态管理器 cc.csMgr = newConnectivityStateManager(cc.ctx, cc.channelz) // 负载均衡器，动态选择每个RPC的子通道 cc.pickerWrapper = newPickerWrapper(cc.dopts.copts.StatsHandlers) // stats cc.metricsRecorderList = stats.NewMetricsRecorderList(cc.dopts.copts.StatsHandlers) cc.initIdleStateLocked() // Safe to call without the lock, since nothing else has a reference to cc. // idle状态管理 cc.idlenessMgr = idle.NewManager((*idler)(cc), cc.dopts.idleTimeout) return cc, nil } 这么一套下来可以看到，初始化ClientConn的时候并没有建立连接，所以猜测是在第一次发起RPC调用的时候才去尝试建立连接。还有一种验证方法是，把服务端关闭，尝试NewClient，是不会返回错误的。 ...

gRPC阅读（1）—— 服务端

gRPC介绍 gRPC 是一种由 Google 开发的高性能远程过程调用（RPC）框架，适用于分布式系统间的通信。它基于 HTTP/2 进行传输，使用 Protocol Buffers 进行序列化，提供跨平台的兼容性。gRPC 的核心理念是让客户端像调用本地函数一样调用远程服务，简化服务间的调用流程。通过编写与具体编程语言无关的 IDL （默认是 protobuf）来定义 RPC 方法，gRPC 框架就会生成语言相关的客户端/服务端代码。 HTTP/2介绍相比http1，具有更高的传输效率（多路复用：在同一个链连接上同时处理多个请求），更低的延迟（服务端推送，减少请求数量、简化header大小）、带宽利用率更高（头部压缩、数据流优先）、更安全（基于tls）。 http2具体特性有：帧、消息、流：帧是小通信数据单元；消息由一个或多个帧组成。例如请求的消息和响应的消息；一个连接中包含多个流，每个流包含多个帧。帧通过流id进行标识属于哪个流二进制分帧：每个消息由若干个帧组成，帧是最小传输单位，并且原来基于文本编码变成基于二进制，进一步减小帧大小压缩header 多路复用：即在同一连接中的多个stream的传输互不影响服务端推送流量控制和资源优先级：流量控制以有效利用多路复用机制，确保只有接收者使用的数据会被传输。优先级机制可以确保重要的资源被优先传输。启动服务端通过官方的 helloworld 例子可以看到，服务端的启动分为三步：创建gRPC的Server 将业务handler注册到Server 调用Server.Serve在端口上进行监听第一步没什么好说的，注意下第二步注册进去的东西： // 注册进去的ServiceDesc var Greeter_ServiceDesc = grpc.ServiceDesc{ ServiceName: "helloworld.Greeter", HandlerType: (*GreeterServer)(nil), Methods: []grpc.MethodDesc{ { MethodName: "SayHello", Handler: _Greeter_SayHello_Handler, }, }, Streams: []grpc.StreamDesc{}, Metadata: "helloworld/helloworld.proto", } // Method对应的handler func _Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(HelloRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(GreeterServer).SayHello(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: Greeter_SayHello_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(GreeterServer).SayHello(ctx, req.(*HelloRequest)) } return interceptor(ctx, in, info, handler) } 前两步比较简单，再来看第三步的如何建立连接并进行处理。类似标准库http的ListenAndServe，本质上就是创建一个死循环等待有新的连接到来，然后开新的goroutine去处理这个连接上的读写事件： ...

场景

短链接系统实现如何保证生成短链接不重复如何存储短链接用302（临时）还是301（永久）重定向 https://cloud.tencent.com/developer/article/1858351 https://blog.csdn.net/codejas/article/details/106102452 https://juejin.cn/post/7312353213348741132 秒杀使用redis（保证秒杀效率）的lua脚本（保证原子性）进行库存扣减，使用分布式事务的二阶段消息解决事务数据一致性。二阶段消息适用于无需回滚的这一类数据一致性问题，主要是为了保证第一阶段操作执行成功后，后续阶段一定能感知并执行。二阶段消息的回查操作，主要还是依赖事务中第一阶段使用的数据库，来保证第一阶段整体操作的原子性以及幂等。无论是请求执行lua脚本的服务端宕机，还是redis服务本身宕机，lua脚本都保证原子性，即写操作均无效库存扣减lua脚本： -- KEYS[1] 库存 -- KEYS[2] 事务当前操作 -- KEYS[3] 如果事务当前操作是回滚操作，则为回滚所对应的操作 local v = redis.call('GET', KEYS[1]) -- 库存 local e1 = redis.call('GET', KEYS[2]) -- 事务当前操作的状态 if v == false or v + ARGV[1] < 0 then -- 库存不足 return 'FAILURE' end if e1 ~= false then -- 当前状态不为空，幂等退出 return 'DUPLICATE' end -- 设置当前操作为已完成 redis.call('SET', KEYS[2], 'op', 'EX', ARGV[3]) if ARGV[2] ~= '' then local e2 = redis.call('GET', KEYS[3]) if e2 == false then -- 如果是回滚操作，将回滚对应操作的状态设置为已回滚 redis.call('SET', KEYS[3], 'rollback', 'EX', ARGV[3]) return end end -- 库存扣减 redis.call('INCRBY', KEYS[1], ARGV[1]) 回查lua脚本： local v = redis.call('GET', KEYS[1]) -- 扣减库存操作的状态 if v == false then -- 为空则直接回滚 redis.call('SET', KEYS[1], 'rollback', 'EX', ARGV[1]) v = 'rollback' end -- 如果阶段1是回滚，直接返回事务失败 if v == 'rollback' then return 'FAILURE' end -- 如果不是回滚，说明事务成功以下是可能出现的各个场景。 ...

杂

clickhouse ClickHouse 的高性能主要来自以下几个方面的设计特点： 1. 列式存储：ClickHouse采用列式存储（Columnar Storage），在查询时可以只读所需的列，而不是整个行。这极大减少了磁盘I/O，尤其适合分析型查询。 2. 数据压缩：列式存储便于数据压缩，ClickHouse内置了多种压缩算法（如LZ4、ZSTD等），根据数据特征选择最佳压缩方式，减少了存储空间和I/O开销。 3. 向量化引擎：ClickHouse的数据处理是基于向量化引擎的。向量化处理数据意味着每次处理一个数据块，而不是单条数据，充分利用了CPU的指令集，极大提升了数据处理速度。 4. 多线程并行查询：ClickHouse支持多线程并行处理，分区并行化查询，充分利用多核CPU的性能。在一个查询中可以调度多个线程同时工作，使得复杂查询能高效完成。 5. 内存管理：ClickHouse对内存管理进行了优化，尽可能减少内存分配和释放操作。比如查询时，通常不会频繁分配和释放内存，而是通过重用内存块来降低内存碎片。 6. 数据分区和分片：ClickHouse支持对数据进行水平分区和分片，能够处理海量数据。它可以将不同分片的数据分布到不同节点上，并行处理分片数据，提高了系统的扩展性和查询性能。 7. 基于MergeTree的表引擎：MergeTree 是ClickHouse的核心存储引擎。它支持数据的自动排序和索引，查询时可以快速定位目标数据块，避免了全表扫描。 8. 向量化索引：ClickHouse会为列数据创建稀疏索引，通过向量化索引来过滤数据块，进一步加速查询。辅助索引的存在能够快速定位相关数据块，尤其适用于高基数的列。 9. 数据延迟写入和异步物化视图：ClickHouse的写入延迟较低，因为它将数据先写入内存中的“insert buffer”缓冲区，之后才会批量写入磁盘。异步物化视图会在后台进行处理，不影响查询性能。以上技术特性使ClickHouse在处理大数据量的复杂分析查询时，性能极为优越，能够快速响应实时查询需求。建表时order by的作用：确定数据存储顺序：ORDER BY字段决定了表内数据的存储顺序，ClickHouse会按照该字段对数据块进行排序。自动生成稀疏索引：ORDER BY字段会自动创建稀疏索引，帮助ClickHouse跳过不满足查询条件的数据块，从而加速查询。提高查询效率：尤其对于范围查询、过滤、分组等，选择合适的ORDER BY字段能够减少扫描量并加快响应速度。名词解释稀疏索引：在ClickHouse中，数据组织为：分区（partition）->数据块（parts）->数据段（granules）。稀疏索引记录每个数据段的特征（比如最大值、最小值），查询的时候用来快速跳过某些不需要加载的数据段。辅助索引：稀疏索引不适用于加速高基数列的查询。辅助索引有bloom filter、minmax等

golang sync包源码阅读

前言 sync包提供了常见的并发编程工具，比如最常见的Mutex、WaitGroup等。这些工具都非常简洁，几乎0学习成本。本篇将从源码角度简单看看这些工具的实现原理，以在未来有需求的时候，理解甚至是手动实现功能更强大的，更复杂的并发编程工具。 sync.Mutex sync.Mutex是golang中的互斥锁，但是注意它仅仅具有互斥访问的功能，没有其他功能，比如不支持可重入、不可自定义公平/非公平。公平性对于公平性，Mutex采取了综合两者的做法： normal mode（非公平模式，利于高效率运行）：锁释放时，优先让同时新来尝试获取锁的线程获取到锁，而不是等待队列中的线程，运行成本低，只需数次CAS就能获取到锁。这是默认的模式 starvation mode（公平模式，避免高并发下线程饿死）：锁释放时，优先让等待队列的线程获取到锁，而不是新来的线程。当等待队列队头线程等待超过1ms进入公平模式如果当前为公平模式，那么当等待队列唯一的队头线程获取到锁，或者队头线程等待时间不足1ms，又会自动回到非公平模式。可重入性在开始源码之前，关于为什么golang的官方互斥锁不考虑支持可重入性我想简单讨论下。Russ Cox在讨论里核心观点在于：互斥锁的目的是保护程序的不变性（即invariant，关于什么是程序的不变性可以参考这篇）。因此当线程获取到互斥锁以及释放锁的那一刻，程序都应该是invariant的，在持有锁的期间，程序可以随便破坏invariant，只要保证释放锁的那一刻恢复了invariant即可。从这个观点来说，如果锁是可重入的，就会有这样的情况发生： func G() { mu.Lock() // 破坏invariant ... F() // 恢复invariant ... mu.Unlock() } func F() { mu.Lock() // 此时持有锁，程序应该是invariant的 // 继续执行下去可能会导致bug，因为F认为持有锁的那一刻程序是invariant的 // 但F不知道invariant已经被G破坏 ... mu.Unlock() } 也就是说，Russ Cox给互斥锁功能上的定义是保持程序的invariant，因此可重入锁的想法就是错的。但也有别的观点认为他对互斥锁的定义是错的，互斥锁本身就是为了避免多线程访问修改变量，invariant是开发者的责任，与你用不用互斥锁无关，互斥锁只是帮助你实现invariant的，并且出于编程上的方便，可重入锁可以make your life easier！况且很多语言其实都支持可重入锁。另外关于invariant，本人也不认为是互斥锁的责任，比如在单线程的程序中，你需要维护(a==b)==true这个invariant，而且由于单线程你根本不需要锁，那么只要你会改变a或者改变b，就肯定会有某些时刻会出现invariant被破坏的情况，但这些情况一般是函数内部的瞬时发生的，而函数执行前后都是保持invariant的就没问题，可以看到与锁无关。因此锁只是个实现invariant的工具之一，它只需要关注底层并发的事情，不需要给他下“保持程序的invariant”这样的高层抽象定义。后面我们会利用sync包现有的工具，尝试实现一个可重入锁。源码 Mutex结构体： const ( mutexLocked = 1 << iota // 锁是否被线程持有 mutexWoken // 是否有被唤醒的线程正在等待获取锁 mutexStarving // 是否处于饥饿模式 ) type Mutex struct { state int32 // 低三位分别对应上述三个状态，高位记录等待队列的线程数 sema uint32 // 给底层同步原语使用的信号量 } Lock方法： func (m *Mutex) Lock() { // Fast path: 使用CAS快速加锁 if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) { if race.Enabled { race.Acquire(unsafe.Pointer(m)) } return } // Slow path: CAS加锁失败，说明锁被其他线程占有，当前应该被阻塞 m.lockSlow() } lockSlow方法： ...