后台 on 子恒的博客

[后台]设计合理的幂等方案

Tue, 09 Apr 2024 16:07:23 +0000

幂等的作用

幂等的作用：保证可重入性，防止重复操作导致脏数据出现。使用场景可能有

前端防抖
接口超时重试
消息重试
只能调一次接口，如一个用户只能领一次券

我们的upsert、redis分布式锁、version乐观锁、for update、唯一索引、状态机其实都有幂等的功能，会在请求重复的时候报错

分布式锁：一般用来防止并发操作，可以处理重复操作的问题。可以加在请求处理的最开始
乐观锁：锁住读实体~写实体期间，这期间的任何其他操作都会失败。但自己也可能失败，要处理好失败的回滚逻辑

方案	支持维度	性能	幂等期	判断幂等时机	使用场景	备注
分布式锁	任意	高	加锁期间	请求开始	操作实体
乐观锁	实体id	低	读-写db期间	写db时	修改实体db行	失败需要回滚前面的操作
insert+uk	任意	低	写db之后	写db时	创建实体	失败需要回滚前面的操作

幂等键设计

结论：一定要用有业务语义的幂等键！最好的幂等键组合是entity_id+idem_key的组合

接口维度的幂等：幂等的控制交给上游，由上游保证自己的请求是可以幂等/不被幂等的。比如上游直接传一个md5sum(req)作为幂等键进来（其他常用的包括req里的核心参数、reqid、时间戳、消息id）。我们检测这个幂等键是否存在：

幂等键已存在：直接幂等。问题：下游可能传错了，不幂等的也结果被幂等，比如批量请求、同一个请求里发起多次请求等。幂等应该自己来控制
幂等键不存在：不幂等，这个不会出错

为了避免上面问题的出现，我们可以结合数据库的实体uk做判断。
数据维度的幂等：采用数据库的带业务语义的uniq key+幂等键联合判断。假设uk就是实体的id，幂等键是业务传过来的自定义值。我们去查找幂等键

uk已存在，幂等键不存在：说明用户希望再次操作同一个数据实体，不幂等
uk不存在，幂等键存在：说明用户希望再次操作其他数据实体，不幂等
uk、幂等键都存在：直接幂等
uk、幂等建都不存在：不幂等

// 这个例子实现了一个幂等键
// 业务场景：假设我们要设计一个领奖接口，每个用户只能领一次奖品。
// seq: 如果此奖励一个user_id可以重复领取多次，需要用seq标识唯一
// 用户id、奖励id、和seq表达一次幂等
func getIdemKey(userId string, prizeId string, seq string) (uniqueId string){
   return fmt.Sprintf("%v_%v_%v", userId, prizeId, seq)
}

实现

实现方法：

令牌发放：服务端/客户端生成一个token（幂等键），给客户端用，客户端带着token前来请求，token是一次性的，用过就直接幂等
幂等表：mysql的数据表里加一列幂等键，这个表一般是操作流水表，用来记操作。每次来请求的时候从流水表查幂等键存不存在，存在则直接幂等
实体表uk：实体表加一列create_idem_key作为uk，幂等的时候直接报错uk冲突。缺点是只能做实体创建的幂等，不能做实体操作的幂等
分布式锁：redis里存幂等key，用setnx+幂等key+超时时间控制

我们举例一个轻量级的幂等应用：

[后台]服务端高性能架构之道（系统和服务篇）

Tue, 28 Sep 2021 21:13:08 +0800

如果你在服务端的工区，常常会听到同学们激烈的讨论，包括能不能扛得住xx流量？能不能P99达到x毫秒？某操作能不能立即生效？某服务CPU飙升了，某服务OOM了，某服务超时率暴涨了？
这些灵魂的质问，其实就是在保障服务端的高并发、高性能、高可用、高一致性等等，是我们服务端同学必备的扎实基本功。

克服系统瓶颈

服务端的代码都跑在各种版本的Linux之上，所以高性能的第一步要和操作系统打交道。我们的服务需要通过操作系统进行I/O、CPU、内存等等设备的使用，同时在使用各种系统调用时避免各种资源的开销过大。

零拷贝

认识零拷贝之前，我们先要对Linux系统I/O机制有一定的了解。当我们执行一个write(2)或者read(2)的时候（或者recv和send），什么时候操作系统会执行读写操作？什么时候又最终会落到磁盘上？
以一个简单的echo服务器为例，我们模拟下每天都在发生的请求和回包：

sockfd = socket(...); 					//打开socket
buffer = new buffer(...); 				//创建buffer 
while((clientfd = accept(socketfd...)){	// 接收一个请求
	read(clientfd, buffer, ...);        //从文件内容读到buffer中 
	write(clientfd, buffer, ...);       //将buffer中的内容发送到网络
}

看一下这段代码的拷贝流程（下图）：

数据包到达网卡，网卡进行DMA操作，把网卡寄存器的数据拷贝到内核缓冲区
CPU把内核缓冲区的数据拷贝到用户空间的缓冲区
用户空间处理buffer中的数据（此处不处理）
CPU把用户空间的缓冲区的数据拷贝到内核缓冲区
网卡进行DMA操作，把内核缓冲区的数据拷贝到网卡寄存器，发送出去

整个过程触发了4次拷贝（2次CPU，2次DMA），2次系统调用（对应4次上下文切换）
（注：DMA(Direct Memory Access)， I/O 设备直接访问内存的一个通道，可以完成数据拷贝，使得CPU 不再参与任何拷贝相关的事情，现在几乎所有的设备都有DMA）

使用mmap

mmap可以把用户空间的内存地址映射到内核空间，这样对用户空间的数据操作可以反映到内核空间，省去了用户空间的一次拷贝：

应用调用mmap，和内核共享缓冲区（只需一次）
数据包到达网卡，网卡进行DMA操作，把网卡寄存器的数据拷贝到内核缓冲区
CPU把接收到的内核缓冲区的数据拷贝到发送的内核缓冲区
网卡进行DMA操作，把内核缓冲区的数据拷贝到网卡寄存器，发送出去

整个过程触发了3次拷贝（1次CPU，2次DMA），2次系统调用（对应4次上下文切换）

使用sendfile/splice

Linux 内核版本 2.1 中实现了一个函数sendfile(2)：

他把read(2)和write(2)合二为一，成为一次系统调用，实现了把一个文件读取并写到另一个文件的语义
系统调用中不再切换回用户态，而是在内核空间中直接把数据拷贝过去（2.4 之后这一步支持了DMA拷贝，实现了CPU零拷贝）

我门看下使用sendfile之后的流程：

整个过程触发了3次拷贝（0次CPU，3次DMA），1次系统调用（对应2次上下文切换）

Linux 内核版本 2.6 中实现了一个函数splice(2)，类似sendfile，但是接收/发送方必须有一个文件是管道，通过管道的方式连接发送方和接收方的内核缓冲区，不再需要拷贝（0次CPU，2次DMA，1次系统调用）

transferTo（内部调用sendfile）的性能对比：

对于我们的启发

零拷贝能带来显著的性能提升，目前kafka，nginx默认都开启了零拷贝（大文件传输可以提升60%以上）
部分场景对时效性或者拷贝次数有严格的要求时（比如数据库、消息队列的实现），可以考虑用mmap或者直接I/O，配合自己实现的缓存替代操作系统的缓存方案
拷贝很可能是CPU消耗的主要原因，比如业务代码中的大结构体复制，所以我们要谨慎控制复制操作，尽量使用指针或者引用类型

无锁

多线程、多协程、多机器、多地部署是我们服务端实现高并发和强容灾的必备解决方案，这些方案都有一个共性，把数据或者过程分而治之。问题在于，几乎所有的并发场景都会涉及到数据竞争，涉及到共享数据的地方就会涉及到锁，协程有锁，线程有锁，多机部署的服务有分布式锁。
服务中的锁会带来很多问题，随着并发数量的加大，会带来更大的上下文切换、用户态切换的开销，出现CPU飙升且都在做一些无用功的现象，也会导致性能快速下降，甚至还不如单线程模型的效率高。除此以外，各种锁还会带来很高的复杂度，和并发的复杂度相叠加，非常容易出现死锁和各种并发问题。

因此，我们使用锁一定是去解决某种问题而去用的，能无锁就无锁，能轻量级就轻量级。

无锁的替代方案

单线程

最简单的方案就是单线程reactor模式，redis、nginx都用了这种方式来避免加锁带来的损耗和复杂性，适用于功能简单的场景。

如图，redis的单线程模型有这么几个部分：

[Redis]Redis 应用篇

Thu, 24 Sep 2020 15:57:27 +0800

分布式锁

简单加锁

// 思想：利用setnx检测有没有set过，如果set过就表示没有抢到锁
> setnx locker true
OK
// ... do somthing ...
> del locker

处理set之后进程崩溃的死锁问题

// 思想：给锁加上过期时间，即使set之后进程挂掉，也不会死锁
> setnx locker true
OK
> expire locker 5
// ... do somthing ...
> del locker

处理非原子性问题

setnx之后，expire之前，进程挂了，也会死锁。怎么处理这种情况？

使用redis事务吗？事务里没有if else，要么全部执行，要么全部不执行。需求是setnx成功才执行expire，有依赖关系，没法用事务
使用新的原子命令，如下

> set locker true ex5 nx
OK
// ... do somthing ...
> del locker

处理超时问题

上面的方案设定了超时时间。但是如果少数操作的时间超过了超时时间怎么办？有两个问题：

[后台]负载均衡（三）限流篇

Mon, 27 Jul 2020 17:28:37 +0800

限流

限流能力是高并发系统中，对于服务提供方的一种保护手段。通过限流功能，我们可以通过控制QPS的方式，以避免被瞬时的流量高峰冲垮，从而保障系统的高可用性。

考虑的问题

完成一个限流系统，我们可以结合场景的需要做下面的考虑

多规则匹配：是否会存在有多重规则的限流？比如有的规则限制每天1000次，有的规则限制每分钟1次？是同时生效还是优先生效某个？
资源类型：能限流什么？QPS，连接数，并发数
全局限流/单机限流：多个服务的实例共享一个全局的流量限额，比如所有机器共享1000QPS。或者单个实例的限流，比如被调限定每台机器不超过1000QPS
限流阈值：单位时间内的最大配额数。是按照每秒种一次，还是按照每分钟60次？
限流处理：客户端如何处理超出限额的请求？超额后直接拒绝，还是超额后进行排队？

抽象出一个方案
接口级别限流：每个接口分配一个appid和key，各自计算各自的配额
多维度限流：支持每秒N次、每分钟N次、每天N次等维度
匀速防刷：假设配置了每分钟60次，依然可能出现第一秒访问了60次用光了配额。匀速防刷可以匀速消耗配额，解决这个问题
多级限流：支持不同的限流规则，并有采用的优先级，采用优先级最高的方案进行限流

限流算法

固定窗口

固定窗口是在一段时间内可以限制访问次数的方法。

将时间划分为多个窗口
在每个窗口内每有一次请求就将计数器加一
如果计数器超过了限制数量,则本窗口内所有的请求都被丢弃。当时间到达下一个窗口时,计数器重置

这样有一定的限流效果，但是限制住的流量可能是有毛刺的。比如1000次/分钟，可能00:59的时候有1000流量，01:00的时候也有1000流量，这样这两秒内就有2000流量！
具体实现：用一个变量C标记访问次数，一个事件定时过期，并在过期时把变量C清零：

type FixedWindowCounter struct {
	TimeSlice  time.Duration
	NowCount   int32
	AllowCount int32
}

func (p *FixedWindowCounter) Take() bool {
	once.Do(func() {
		go func() {
			for {
				select {
				case <-time.After(p.TimeSlice):
					atomic.StoreInt32(&p.NowCount, 0)
				}
			}
		}()
	})

	nowCount := atomic.LoadInt32(&p.NowCount)
	if nowCount >= p.AllowCount {
		return false
	}
	if !atomic.CompareAndSwapInt32(&p.NowCount, nowCount, nowCount+1) {
		return false
	}
	return true
}

滑动窗口

滑动窗口把一个固定窗口划分为多个小区间，以固定的窗口大小在区间上滑动，在一定程度上缓解了固定窗口的毛刺问题（我们假设某一区间qps非常高）。当然性能损耗也比较大。

[后台]负载均衡（二）能力篇

Thu, 21 May 2020 17:18:55 +0800

名字服务

基础设计

名字服务考虑的基本设计

客户端发现：
服务提供者的实例在启动时或者位置信息发生变化时会向服务注册表注册自身，在停止时会向服务注册表注销自身，如果服务提供者的实例发生故障，在一段时间内不发送心跳之后，也会被服务注册表注销。
服务消费者的实例会向服务注册表查询服务提供者的位置信息，然后通过这些位置信息直接向服务提供者发起请求。

服务端发现：
第一步与客户端发现相同。
服务消费者不直接向服务注册表查询，也不直接向服务提供者发起请求，而是将对服务提供者的请求发往一个中央路由器或者负载均衡器，中央路由器或者负载均衡器查询服务注册表获取服务提供者的位置信息，并将请求转发给服务提供者。

这两种模式各有利弊，客户端发现模式的优势是，服务消费者向服务提供者发起请求时比服务端发现模式少了一次网络跳转，劣势是服务消费者需要内置特定的服务发现客户端和服务发现逻辑；
服务端发现模式的优势是服务消费者无需内置特定的服务发现客户端和服务发现逻辑，劣势是多了一次网络跳转，并且需要基础设施环境提供中央路由机制或者负载均衡机制。目前客户端发现模式应用的多一些，因为这种模式的对基础设施环境没有特殊的要求，和基础设施环境也没有过多的耦合性。
主调调用被调时，根据被调的名字从服务注册中心获取服务实例列表，包括节点ip、端口、权重、地理位置等；一般采取分钟级别的定时任务去拉取，本地做缓存，异步更新。

实现方式

DNS，传播速度太慢，没法发现端口。SkyDNS解决了这个问题，在k8s里大量使用
zookeeper或者etcd，如SmartStack，能保证强一致，但是要做很多开发
Eureka。Netflix的java生态里的优秀方案
Consul，提供服务配置、服务的内存和磁盘监测等

服务注册信息

IP和端口

一个服务端要接入名字服务，必须要先提供自己的IP和端口信息。
IP的获取方法：

通过遍历网卡的方式去获取，找到第一个不为本地环回地址的 IP 地址。dubbo就是这种方法
指定网卡名interfaceName，来获取IP
直接与服务注册中心建立 socket 连接，然后通过socket.getLocalAddress() 这种方式来获取本机的 IP
端口的获取方法：
一般的RPC服务或者Web服务监听的端口都在配置中写好，可以直接获取上报。

扩展设计

除了IP和端口，可能还有一些常用的服务信息需要注册上来，提供更高级的功能：
1.支持TLS：想知道某个 HTTP 服务是否开启了 TLS。
2.权重：对相同服务下的不同节点设置不同的权重，进行流量调度。
3.环境分配：将服务分成预发环境和生产环境，方便进行AB Test功能。
4.机房：不同机房的服务注册时加上机房的标签，以实现同机房优先的路由规则。

无损注册/下线

虽然服务注册一般发生在服务的启动阶段，但是细分的话，服务注册应该在服务已经完全启动成功，并准备对外提供服务之后才能进行注册。
1.有些 RPC 框架自身提供了方法来判断服务是否已经启动完成，如 Thrift ，我们可以通过 Server.isServing() 来判断。
2.有一些 RPC 框架本身没有提供服务是否启动完成的方式，这时我们可以通过检测端口是否已经处于监听状态来判断。
3.而对于 HTTP 服务，服务是否启动完毕也可以通过端口是否处于监听状态来判断。
下线也是一样的，可以注册服务下线的回调，或者监听服务下线的信号，或者做健康检查

健康检查

客户端主动心跳上报健康：

客户端每隔一定时间主动发送“心跳”的方式来向服务端表明自己的服务状态正常，心跳可以是 TCP 的形式，也可以是 HTTP 的形式。
也可以通过维持客户端和服务端的一个 socket 长连接自己实现一个客户端心跳的方式。
客户端的健康检查只能表明网络可达，不能代表服务可用。服务端的健康检查可以准确获得服务的健康状态：
服务端调用服务发布者某个 HTTP 接口来完成健康检查。
对于没有提供 HTTP 服务的 RPC 应用，服务端调用服务发布者的接口来完成健康检查。
可以通过执行某个脚本的形式来进行综合检查，覆盖多个场景。
服务端检查也有问题，一个是调用服务的方式不通用，需要额外实现，还有就是服务注册中心可能和服务的网络不互通。

节点变化通知

当服务有节点退出或新的节点加入时，订阅者如何及时收到通知？经典的push和pull的问题

[后台]负载均衡（一）算法篇

Wed, 06 May 2020 20:24:30 +0800

当单机的访问压力很大时，就需要引入集群。集群一个很重要的事情就是把请求均匀地分配在各个机器上，这就是负载均衡的雏形。
有基于MAC地址的二层负载均衡和基于IP地址的三层负载均衡。二层负载均衡会通过一个虚拟MAC地址接收请求，然后再分配到真实的MAC地址；三层负载均衡会通过一个虚拟IP地址接收请求，然后再分配到真实的IP地址；
四层通过虚拟IP+端口接收请求，然后再分配到真实的服务器（比如LVS，F5）；七层通过虚拟的URL或主机名接收请求，然后再分配到真实的服务器（Haproxy和Nginx）。
四层和七层是最常见的负载均衡模型。

**四层：**以常见的TCP为例，负载均衡设备在接收到第一个来自客户端的SYN请求时，通过负载均衡算法选择服务器，并对报文中目标IP地址进行修改（改为后端服务器IP），直接转发给该服务器。TCP的连接建立，即三次握手是客户端和服务器直接建立的，负载均衡设备只是起到一个类似路由器的转发动作。在某些部署情况下，为保证服务器回包可以正确返回给负载均衡设备，在转发报文的同时可能还会对报文原来的源地址进行修改。
**七层：**以常见的TCP为例，负载均衡设备如果要根据真正的应用层内容再选择服务器，只能先代理最终的服务器和客户端建立连接(三次握手)后，才可能接受到客户端发送的真正应用层内容的报文，然后再根据该报文中的特定字段，再加上设置的负载均衡算法，选择内部某台服务器。负载均衡设备在这种情况下，更类似于一个代理服务器。负载均衡和前端的客户端以及后端的服务器会分别建立TCP连接。所以从这个技术原理上来看，七层负载均衡明显的对负载均衡设备的要求更高，处理七层的能力也必然会低于四层模式的部署方式。
参考资料：四层和七层负载均衡的区别

nginx用的负载均衡算法

Nginx可以作为HTTP反向代理，把访问本机的HTTP请求，均分到后端集群的若干台服务器上。负载均衡的核心就是负载均衡所使用的平衡算法，适用于各种场景。
Nginx的负载均衡算法
Nginx目前提供的负载均衡模块：
ngx_http_upstream_round_robin，加权轮询，可均分请求，是默认的HTTP负载均衡算法，集成在框架中。
ngx_http_upstream_ip_hash_module，IP哈希，可保持会话。
ngx_http_upstream_least_conn_module，最少连接数，可均分连接。适用于链接数体现资源的服务，比如FTP。
ngx_http_upstream_hash_module，一致性哈希，可减少缓存数据的失效。

随机访问

在介绍nginx的模式前，先介绍下普通的负载均衡方法。假设有7个请求，我们给A、B、C三个节点分别4、2、1的权重。最朴素的负载均衡方式有下面几种：

完全轮询：访问完A去访问B，访问完B去访问C，再去访问A。缺点是没有权重，不能根据负载调节。
列表轮询：构造一个数组[A, A, A, A, B, B, C]，每次pop出去一个访问。缺点是pop出去的元素太随机，可能一次集中访问A ，而且占用内存太大，对于几万的权重范围不合适。
随机数：我们按照A、B、C的权重划分好区间，A（0、1、2、3），B（4、5），C（6），然后取一个随机数，模余7，看看最后的结果在哪个区间内，就取哪个节点。缺点是完全随机，无法避免集中访问。

加权轮询

假设有7个请求，我们给A、B、C三个节点分别4、2、1的权重。如果随机按照概率来选，那么很可能出现连续四个请求都在A上面的情况，这样只能保证结果看起来均衡，但是时间段内不均衡。Nginx采用了一种平滑的加权平均算法来选取节点（Weighted Round Robin）。
先引入三个概念，都用来描述服务器节点的权重：

$W$ : weight 我们指定的权重，就是上面例子中的4、2、1。
$W_{ew}$: effective_weight 有效权重，初始值为$W$。用来对故障节点降权。
如果通信中有错误产生，就减小effective_weight。（故障降权）
此后有新的请求过来时，再逐步增加effective_weight，最终又恢复到weight。（自动恢复）
$W_{cw}$ : current_weight 当前真实权重，每次都会选到最大的真实权重的节点去请求

真实权重$W_{cw}$计算方式：

初始化：$W_{cw}$ 起始值为0
获得实时权重：请求到来后，给每个节点的真实权重加上有效权重，即$每个节点 W_{cw} = W_{cw} + W_{ew}$
选出最大权重：选择真实权重最大的节点最为本次请求的目标
回避刚选的节点：最选择的节点的实时权重减去所有节点（包括自己）的有效权重和。即$选中节点 W_{cw} = W_{cw} - (W_{ew1} + W_{ew2} + … + W_{ewn})$

来看一个具体的例子：
假设A、B、C三个节点的权重分别为4、2、1。

请求序号	请求后的current_weight	选择的节点	选择后的current_weight
未请求	{0,0,0}	/	/
1	{4,2,1}	A	{-3,2,1}
2	{1,4,2}	B	{1,-3,2}
3	{5,-1,3}	A	{-2,-1,3}
4	{2,1,4}	C	{2,1,-3}
5	{6,3,-2}	A	{-1,3,-2}
6	{3,5,-1}	B	{3,-2,-1}
7	{7,0,0}	A	{0,0,0}
三个结论：每个节点被选中的次数是符合权重的；A没有被连续选取；七次之后权重会归零，是一个循环。

static ngx_http_upstream_rr_peer_t *ngx_http_upstream_get_peer(ngx_http_upstream_rr_peer_data_t *rrp)
{
    time_t                        now;
    uintptr_t                     m;
    ngx_int_t                     total;
    ngx_uint_t                    i, n, p;
    ngx_http_upstream_rr_peer_t  *peer, *best;

    now = ngx_time();
    best = NULL;
    total = 0;
    ...

    /* 遍历集群中的所有后端 */
    for (peer = rrp->peers->peer, i = 0;
         peer;
         peer = peer->next, i++)
    {

        n = i / (8 * sizeof(uintptr_t));
        m = (uintptr_t) 1 << i % (8 * sizeof(uintptr_t));

        /* 检查该后端服务器在位图中对应的位，为1时表示不可用 */
        if (rrp->tried[n] & m)
            continue;
       
        /* 永久不可用的标志 */
        if (peer->down) 
            continue;
        
       /* 在一段时间内，如果此后端服务器的失败次数，超过了允许的最大值，那么不允许使用此后端了 */
       if (peer->max_fails
            && peer->fails >= peer->max_fails
            && now - peer->checked <= peer->fail_timeout)
            continue;
        
        peer->current_weight += peer->effective_weight; /* 对每个后端，增加其当前权重 */
        total += peer->effective_weight; /* 累加所有后端的有效权重 */

        /* 如果之前此后端发生了失败，会减小其effective_weight来降低它的权重。          
         * 此后在选取后端的过程中，又通过增加其effective_weight来恢复它的权重。          
         */        
        if (peer->effective_weight < peer->weight) 
            peer->effective_weight++;
        
        /* 选取当前权重最大者，作为本次选定的后端 */
        if (best == NULL || peer->current_weight > best->current_weight) {
            best = peer;
            p = i;
        }
    }

    if (best == NULL) /* 没有可用的后端 */
        return NULL;
    
    rrp->current = best; /* 保存本次选定的后端 */

    n = p / (8 * sizeof(uintptr_t));
    m = (uintptr_t) 1 << p % (8 * sizeof(uintptr_t));

    /* 对于本次请求，如果之后需要再次选取后端，不能再选取这个后端了 */    
    rrp->tried[n] |= m;

    best->current_weight -= total; /* 选定后端后，需要降低其当前权重 */  
    /* 更新checked时间 */
    if (now - best->checked > best->fail_timeout)
        best->checked = now;
    
    return best;
}

ip_hash

ip_hash是基于客户端IP的哈希值来选择服务器。同一个客户端的请求，都会发往同一台后端，除非该后端不可用了。ip_hash能够达到保持会话的效果。
和随机加权哈希一样，ip哈希借鉴了权重分段的思想，先算出哈希值, 然后模余total_weight，得到初始权重W [0, total_weight)，开始遍历节点。如果哈希值小于当前节点的权重，就选择当前节点；如果哈希值大于等于当前节点的权重，就减去当前节点的权重，再去尝试下个节点。我们还是讨论A，B，C三个节点的情况，假设权重分别为2、4、1 。
权重的和是7，那么哈希值应该是0~6 （因为模了哈希值）
6的情况：6 >= 2 , 所以不选A，减去2。 6-2 >= 4，所以不选B，减去4。2-4 < 1，所以选C
5的情况：5 >= 2 , 所以不选A，减去2。 5-2 < 4，所以选B
4的情况：4 >= 2 , 所以不选A，减去2。 4-2 < 4，所以选B 。哈希值为3、2的情况也一样。
1、0的情况：1 < 2，所以选A
可以看到最后四个选B，两个选A，一个选C，还是均衡的。

[后台]RocketMQ的架构和设计

Wed, 04 Sep 2019 19:00:56 +0800

主要整理文献：
RocketMQ部署架构和技术架构 - Github
RocketMQ关键机制的设计原理 - Github
RocketMQ 原理简介 - 淘宝消息中间件项目组

设计理念和部署

消息队列需要解决的问题

发布/订阅 最基础的需求，可以做解耦&聚合，如果用Redis做，不够可靠
支持优先级队列、延时队列
顺序消费，rockmq严格有序
支持消息过滤，Producer和consumer共同过滤
持久化 内存缓存+文件
异常恢复
broker crash，os crash，掉电 —保证消息不丢，或者丢失少量数据（依赖刷盘方式是同步还是异步）
磁盘损坏，机器永久损坏 —通过异步复制，可保证99%的消息不丢
实时性 RocketMQ使用长轮询Pull方式，可保证消息非常实时，消息实时性不低于Push。
At least Once 和 Exactly Only Once，至少消费一次且只消费一次
broker的buffer容量问题。RocketMQ 的内存Buffer持久化在硬盘，抽象成一个无限长度的队列，不管有多少数据进来都能装得下，当然也会定时清理。
回溯消费 一般是按照时间维度，例如由于 Consumer 系统故障，恢复后需要重新消费 1 小时前的数据，那么 Broker 要提供一种机制，可以按照时间维度来回退消费进度。
RocketMQ 支持按照时间回溯消费，时间维度精确到毫秒，可以向前回溯，也可以向后回溯。
消息堆积 消息堆积在内存Buffer，一旦超过内存Buffer，可以根据一定的丢弃策略来丢弃消息，对性能影响不大，但是不能堆积太多
消息堆积到持久化存储系统中，例如DB，KV存储，文件记录形式。当消息不能在内存Cache命中时，要不可避免的访问磁盘，会产生大量读IO，读IO的吞吐量直接决定了消息堆积后的访问能力。
消息重试 消息重试有两种原因，一种是消息本身处理失败，如编码有问题等，重试永远不会成功。另一部分是处理消息依赖的下游服务暂时不可用，一段时间重试后可以成功。所以可以消极重试，逐步重试增大等待重试间隔。

RockMQ 模块

Name Server ：NameServer是一个非常简单的Topic路由注册中心，其角色类似Dubbo中的zookeeper，支持Broker的动态注册与发现。
(1) 路由管理
Broker管理：NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；
路由信息管理：每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，找到对应topic的路由信息，从而进行消息的投递和消费。
(2) 无状态：NameServer通常也是集群的方式部署，各实例间相互不进行信息通讯。它是一个几乎无状态的结点，他们之间互不通信。Broker是向每一台NameServer注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，Broker仍然可以向其它NameServer同步其路由信息，Producer,Consumer仍然可以动态感知Broker的路由的信息。
(3) 随机选择：客户端连接时，会随机选择。
(4) 长连接：Broker向所有的NameServer结点建立长连接，注册Topic信息。Producer和Consumer也是长连接。
Broker：处理消息存储，转发等处理的服务器。
(0) 分Group：Broker以group分开，每个group只允许一个master，若干个slave。
(1) 读写分离：只有master才能进行写入操作，slave不允许。
(2) 主从同步：slave从master中同步数据。同步策略取决于master的配置，可以采用同步双写，异步复制两种。
(3) 默认消费：在默认情况下，消费者都从master消费，只有master挂掉或者产生消息堆积了才从slave消费。
Broker有下面几个重要的子模块：
(1) Remoting Module：整个Broker的实体，负责处理来自clients端的请求。
(2) Client Manager：负责管理客户端(Producer/Consumer)和维护Consumer的Topic订阅信息
(3) Store Service：提供方便简单的API接口处理消息存储到物理硬盘和查询功能。
(4) HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。
(5) Index Service：根据特定的Message key对投递到Broker的消息进行索引服务，以提供消息的快速查询。

[Linux]谈一谈并行Counting

Fri, 26 Jul 2019 16:49:50 +0800

简单的并行计数

在一个简单的多线程计数程序中，我们假设要每个线程去把sum的值多加100m，同时进行。代码如下：

#include <pthread.h>
#include <stdio.h>

#if 0
  #define ADD_P(x) __sync_fetch_and_add((x), 1)
#else
  #define ADD_P(x) (++(*x))
#endif
#define TC 8

void *thgo(void *arg){
  long i = 1000*1000*100;
  while(i-- > 0){ADD_P((long *)arg);};
  pthread_t me = pthread_self();
  printf("thread sum: %ld tid: %lu \n",  *(long *)arg, (unsigned long)me);
}

int main (){
  long sum = 0;
  pthread_t ths[TC];
  // threads
  for (int i = 0; i < TC; ++i){
      pthread_create(&ths[i], NULL, thgo, &sum);
  }
  // main thread
  thgo(&sum);
  // join
  for (int i = 0; i < TC; ++i){
      pthread_join(ths[i], NULL);
  }
  printf("all final sum : %ld\n", sum);
  return 0;
}

如果使用一般的计数，会出现严重的数据踩踏问题，导致结果只能取得一定近似的值：

[后台]分布式基础概念

Thu, 09 Nov 2017 22:22:02 +0800

CAP

Brewer’s CAP Theorem, 2009

CAP理论是分布式系统的基石，他指出了分布式系统的以下特性：

Consistency 强一致性(返回的数据时刻一致)
Availability 高可用性(服务一直可用，响应时间正常)
Tolerance of network Partition 分区容错性(保证有机器宕机服务依然正常)

CAP理论表明，一个分布式系统不可能同时满足一致性，可用性和分区容错性这三个需求，
最多只能同时满足两个。证明可以看上面的链接。

所以架构师往往要做出牺牲某一特性的选择：

CP：金融行业的数据库，价格昂贵，符合ACID
CA：传统关系型数据库，用于小型系统或单机系统
AP：key-value数据库，现代UGC服务基本都是这种架构，用最终一致性来换取高可用和分区容错性。
电商：牺牲少量的可用性和一致性，比较平衡，符合BASE

ACID

老生常谈的数据库事务的特性：

原子性(Atomicity) 事务中有失败，立即回滚到执行前。没有失败，全部成功

一致性(Consistency) 事务执行后数据的约束没有被破坏

隔离性(Isolation） 事务之间不交叉进行

持久性(Durability） 事务完成，永久储存

BASE

可伸缩性最佳实践：来自eBay的经验

BASE理论是对CAP理论的延伸，核心思想是即使无法做到强一致性（CAP的Consistency），但应用可以采用适合的方式达到最终一致性，来保证系统的容量和可用性。

Basically Availble （基本可用）

基本可用是指系统只保证核心可用，在访问量突增时采用有损服务的策略，让部分用户使用降级的服务。
什么是基本可用的服务？以秒杀为例：
逻辑有损：秒杀时只执行重要逻辑，加载资源可以先使用预加载的或者小图等
业务有损：秒杀需要会员、抢秒杀资格
流程有损：比如秒杀时前段丢掉大部分请求，从后端少量请求中选取中奖的请求处理

Soft-state （软状态/柔性事务）

“Soft state” 可以理解为"无连接"的, 而 “Hard state” 是"面向连接"的。换句话说，软状态的数据库可能存在很多中间状态，不同节点到达最终统一的状态前会有延迟（如异步复制）。

Eventual Consistency （最终一致性）

并非时刻保持一致，所有复制节点在某段时间后保持一致。
最终一致性是弱一致性的一种特例：
Step 1. A首先write了一个值到存储系统
Step 2. 存储系统保证在A,B,C后续读取之前没有其它写操作更新同样的值
Step 3. 最终所有的读取操作都会读取到最A写入的最新值

从A写入到读取操作读取成功有一定的时间，即不一致性窗口。如果没有失败发生的话，“不一致性窗口”的大小依赖于以下的几个因素：交互延迟，系统的负载，以及备机slave的个数。最终一致性方面最出名的系统可以说是DNS系统，当更新一个域名的IP以后，根据配置策略以及缓存控制策略的不同，最终所有的客户都会看到最新的值。

在下一篇文章的中，会表明只要集群$V_r + V_w \leq N$，即此时读取和写入操作是不重叠的，就能保证最终一致性。这个时候不一致性的窗口依赖于存储系统的异步实现方式，不一致性的窗口大小就等于从更新开始到所有的节点都异步更新完成之间的时间。

[后台]用select写一个HTTP代理

Thu, 12 Jan 2017 11:23:44 +0800

函数说明

18.3. select — Waiting for I/O completion — Python 3.6.0 documentation

select.select(rlist, wlist, xlist[, timeout])

Unix select()系统调用的一个简单接口。前三个参数是文件描述符（int类型）的序列。

1
2
3

    rlist：等待准备读取
    wlist：等待准备写作
    xlist：等待一个异常

参数说明

在Unix上序列可以为空，但在Windows则不可以。可选的timeout参数将超时指定为浮点数（以秒为单位）。当省略timeout参数时，该函数阻塞，直到至少一个文件描述符已准备好。该值为0时表示轮询从不停止。

返回值

准备好的对象列表的三元组（前三个参数的子集）。当超时但没有文件描述符就绪时，返回三个空列表。

序列对象

序列中可用的对象类型包括Python文件对象（例如sys.stdin或由open()或os.popen()返回的对象），由socket.socket()返回的socket对象。你也可以自己定义一个类，只要它有一个合适的fileno()方法（返回一个真正的文件描述符，而不只是一个随机整数）。

注意：
Windows上的文件对象不可接受，但socket可以。在Windows上，底层的select()函数由WinSock库提供，不处理不是源自WinSock的文件描述符。

实现HTTP代理

代理由两个部分组成：客户端-代理的socket，和代理-Web服务器的socket。
因为代理像两边接受数据时调用recv方法都会发生阻塞，
所以，在这个样例中，事件循环由两个事件组成：客户端-代理链接可以继续接受数据了，和代理-Web服务器链接可以接受数据了（指都是代理端）。

    import select
    import socket
    import _thread
    
    BUFF_SIZE = 1024
    
    
    # get http head data from a connection
    def get_post(conn):
        res = b''
        while True:
            res += conn.recv(BUFF_SIZE)
            if b'\r\n\r\n' in res:
                break
        return res
    
    
    class Proxy:
        # init. in_conn: browser-proxy connection, out_conn: proxy-web_server connection
        def __init__(self, conn, timeout):
            self.in_conn = conn
            self.out_conn = None
            self.timeout = timeout
    
            self.run()
    
        # main function
        def run(self):
            action, url, http_ver, req, headers = self.__parse_header()
            # https
            if action in ["CONNECT"]:
                self.in_conn.send(b'HTTP/1.1 200 Connection established\r\nProxy-agent: God Proxy\r\n\r\n')
                self.__connect(headers)
            # http
            else:
                self.__connect(headers)
                self.out_conn.send(req)
            self.event_loop()
            # close connection
            self.in_conn.close()
            self.out_conn.close()
    
        # event loop
        def event_loop(self):
            count = 0
            inputs = [self.in_conn, self.out_conn]
            outputs = []
            while True:
                read_list, _, exception_list = select.select(inputs, outputs, inputs, self.timeout)
                count += 1
                for future_object in read_list:
                    # once could read, receive data
                    data = future_object.recv(1024)
                    if data:
                        # read in_conn's data, send to out_conn or out_conn to in_conn
                        if future_object is self.in_conn:
                            # print(count, 'out conn', data[:10], '...', data[-10:])
                            self.out_conn.sendall(data)
                        else:
                            # print(count, 'in conn', data[:10], '...', data[-10:])
                            self.in_conn.sendall(data)
                if exception_list:
                    break
                # time out
                if count > 100:
                    break
    
        # parse host from url and setting up out_conn socket
        def __connect(self, headers):
            host_list = headers['Host'].split(':')
            host = host_list[0]
            if host_list[-1].isdigit():
                port = int(host_list[1])
            else:
                port = 80
            socket_info = socket.getaddrinfo(host, port)[0]
            self.out_conn = socket.socket(socket_info[0])
            self.out_conn.connect(socket_info[4])
            print('WEB SITE: ', host, socket_info[4])
    
        # get headers from in_conn and parse
        def __parse_header(self):
            req = get_post(self.in_conn)
            str_req = req[:req.find(b'\r\n\r\n')].decode('utf8')
            headers = {}
            lines = str_req.split('\r\n')
            for line in lines[1:]:
                line = line.split(': ')
                headers[line[0]] = line[1]
            action, url, http_ver = lines[0].split()
            return action, url, http_ver, req, headers
    
    
    def main():
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    
        server_address = ('0.0.0.0', 8899)
        server.bind(server_address)
    
        server.listen(0)
        # 1 connection to 1 thread
        while True:
            conn, addr = server.accept()
            _thread.start_new_thread(Proxy, (conn, 3))
    
    if __name__ == '__main__':
        main()

[后台]在Python中使用epoll

Sat, 07 Jan 2017 16:07:35 +0800

学期初写代理的时候发现阻塞式socket并不能满足需要，多线程也不是一个很好的解决方案。这篇文章系统地介绍了epoll是什么以及如何使用。

英文好的同学可以直接看
How To Use Linux epoll with Python
本博客的大部分内容都是基于这篇文章的。

什么是epoll

（维基百科）
于Linux 2.5.44首度登场的epoll是Linux内核的可扩展I/O事件通知机制。它设计目的只在替换既有POSIX select(2)与poll(2)系统函数，让需要大量操作文件描述符的程序得以发挥更优异的性能(举例来说：旧有的系统函数所花费的时间复杂度为O(n)，epoll则耗时O(1))。epoll与FreeBSD的kqueue类似，底层都是由可配置的操作系统内核对象建构而成，并以文件描述符(file descriptor)的形式呈现于用户空间。

为什么使用epoll

首先理解同步异步的概念：
怎样理解阻塞非阻塞与同步异步的区别？- 知乎

同步与异步
同步和异步关注的是消息通信机制 (synchronous communication/ asynchronous communication)
所谓同步，就是在发出一个调用时，在没有得到结果之前，该调用就不返回。但是一旦调用返回，就得到返回值了。
换句话说，就是由调用者主动等待这个调用的结果。
而异步则是相反，调用在发出之后，这个调用就直接返回了，所以没有返回结果。换句话说，当一个异步过程调用发出后，调用者不会立刻得到结果。而是在调用发出后，被调用者通过状态、通知来通知调用者，或通过回调函数处理这个调用。
阻塞与非阻塞
阻塞和非阻塞关注的是程序在等待调用结果（消息，返回值）时的状态.
阻塞调用是指调用结果返回之前，当前线程会被挂起。调用线程只有在得到结果之后才会返回。
非阻塞调用指在不能立刻得到结果之前，该调用不会阻塞当前线程。

阻塞式socket：
阻塞式socket使用单线程进行通信。主线程创建一个新的socket线程并侦听该socket，它一次性接受所有连接，然后让这个线程与客户端进行交互。因为创建的线程都只和一个客户端通信，所以阻塞并不妨碍其他线程工作
虽然多线程的阻塞式通信编码简单，但是在多线程共享资源时非常困难，在多核下表现也很差

非阻塞的socket：
于是很多替代并发阻塞式socket的方法出现了，例如基于事件系统设计的epoll，它并不会阻塞线程，它可以使用一个线程实现，大大节约了资源.

可见epoll是一个同步的非阻塞解决方案。

一般步骤

使用python中的epoll一般有如下步骤：

创建epoll对象
告诉epoll对象监视特定socket上的特定事件
询问epoll对象哪些socket可能有自上次查询以来的指定事件
对这些socket执行一些操作
告诉epoll对象修改要监视的socket和事件的列表
重复步骤3到5直到完成
销毁epoll对象

代码实例

epoll设有水平触发和边沿触发两种模式。

在边沿触发操作模式下，socket发生读或写事件后，调用epoll.poll()只会给socket返回一次事件。调用程序必须处理与该事件相关的所有数据。当事件的数据为空时，对socket的继续操作将导致异常。
在水平触发操作模式中，对epoll.poll()的重复调用将导致注册事件的重复通知，直到与该事件相关联的所有数据都已被处理。

边沿触发往往被用于程序员不想使用系统的事件管理机制时。
下面的示例是一个水平触发的典型样例。
需要注意的是，event & select.EPOLLIN 表示事件的掩码，当掩码值为0时对应的事件发生。

    import socket, select

    EOL1 = b'\n\n'
    EOL2 = b'\n\r\n'
    response  = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n'
    response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n'
    response += b'Hello, world!'

    serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    serversocket.bind(('0.0.0.0', 8080))
    serversocket.listen(1)

    # 由于socket在默认情况下阻塞，因此使用非阻塞（异步）模式
    serversocket.setblocking(0) 

    # 创建一个epoll对象
    # 注册socket的读事件为偏好(读事件会在socket建立链接时发生)
    epoll = select.epoll()
    epoll.register(serversocket.fileno(), select.EPOLLIN)

    try:
       # connections字典将文件描述符（int类型）映射到它们对应的网络连接对象。
       connections = {}; requests = {}; responses = {}
       while True:
          # 查询epoll对象并获取等待时间内发生的事件列表。参数“1”表示可以等待一秒钟。
          events = epoll.poll(1)    
          for fileno, event in events:  # events 是(文件描述符，事件代码）二元组的数组
             # 建立一个新的链接
             if fileno == serversocket.fileno():    
                connection, address = serversocket.accept()
                connection.setblocking(0)   # 设置为非阻塞模式
                epoll.register(connection.fileno(), select.EPOLLIN) # 注册读事件（EPOLLIN）
                connections[connection.fileno()] = connection
                requests[connection.fileno()] = b''
                responses[connection.fileno()] = response
             # 当服务端该收数据（客户端已经发出）
             elif event & select.EPOLLIN:   
                requests[fileno] += connections[fileno].recv(1024)
                if EOL1 in requests[fileno] or EOL2 in requests[fileno]:
                   # 一旦接收到完整的请求，则取消注册读事件并注册写事件（EPOLLOUT）
                   # 将响应数据发送回客户端时可能会发生写事件
                   epoll.modify(fileno, select.EPOLLOUT)    
                   # 打印完整的请求(尽管与客户端的通信是交错的，但是该数据可以被组合并作为整个消息处理)
                   print('-'*40 + '\n' + requests[fileno].decode()[:-2]) 
             # 当服务端该发数据（客户端可以收）
             elif event & select.EPOLLOUT:  
                byteswritten = connections[fileno].send(responses[fileno]) # 一次一次发送响应数据
                responses[fileno] = responses[fileno][byteswritten:]
                if len(responses[fileno]) == 0: # 直到完整响应已传送到OS准备进行传输
                   epoll.modify(fileno, 0)  # 发送完整的响应后，禁止读写事件
                   connections[fileno].shutdown(socket.SHUT_RDWR)   # 关闭连接
             # HUP事件表示客户端socket已经断开连接(不需要注册, 默认发送)
             elif event & select.EPOLLHUP:  
                epoll.unregister(fileno)
                connections[fileno].close()
                del connections[fileno]
    finally:
       # Python将在程序结束时自动关闭socket连接,不需显式关闭
       epoll.unregister(serversocket.fileno())
       epoll.close()
       serversocket.close()