前言

上一篇文章介绍了Pulsar的相关理论知识，今天就来总结下实践过程中Pulsar的相关注意事项吧

Pulsar小总结

基础架构

Apache Pulsar 当然也是一款MQ，它是Pub/Sub型的消息系统，但是从架构设计上来说与前文介绍的kafka是完全不同的，Pulsar在结构上将计算与存储完全分离。先看下图来初步认识下Pulsar的整体架构：

首先来总结一下Pulsar较其他消息队列的优势，为什么在众多MQ当中，我们选择了Pulsar：

1、Pulsar支持多租户，按Namespace进行级别进行资源隔离，搭建公司级集群更加方便；
2、计算与存储隔离，无状态broker可以更好的动态扩所容，天赋云原生；
3、消费模式同时支持流模式（kafka）和队列模式（rabbitmq），选择性更强；
4、读写分离，支持百万级topic(tail-read And catch-up-read)，更低的端到端时延；
5、无Rebalance；
6、支持消息空洞，处理并发消费模式更加便捷；

先来回顾一下上文中的两个核心概念 ：订阅模式 与 Ack模式

–

接下来对每一种模式进行细说：

第一种 - 独占模式（Exclusive）
一个 Subscription 只能与一个 Consumer 关联，只有这个 Consumer 可以接收到 Topic 的全部消息，如果该 Consumer 出现故障了就会停止消费。
Exclusive 订阅模式下，同一个 Subscription 里只有一个 Consumer 能消费 Topic，如果多个 Consumer 订阅则会报错，，也就是说一个消费者处理主题的所有分区，适用于全局有序且对性能要求不高的消费场景，如下图所示：

这个是Consumer B启动时会报错，只有ConsumerA能收到消息。

注意：Pulsar默认的订阅类型为 Exclusive。

第二种 - 灾备模式（Failover）
当存在多个 consumer 时，将会按字典顺序排序，第一个 consumer 被初始化为唯一接受消息的消费者。当第一个 consumer 断开时，所有的消息（未被确认和后续进入的）将会被分发给队列中的下一个 consumer。说到这里其实会有一个误区，常常会认为该模式是上述Exclusive模式上增加了一个backup机制，其实不然，这两个模式并无任何关联，请听分解：

Failover订阅模式，可以允许多个消费者附加到一个订阅上。消费者的队列分配策略根据主题类型有所区别：

如果是分区主题(一个主题拥有多个队列的主题)，Broker服务端会根据消费者优先级和消费者名称字典顺序进行排序，然后Broker会将主题中的分区平均分配给高优先级的消费者，低优先级消费者会成为分区的备消费者。
如果是非分区主题，Broker按照消费者订阅主题的顺序选择主消费者，其他的成为备消费者。
非分区主题的订阅示例说明如下：

对于分区主题的订阅图解说明如下：

所以你能看到这种消费模式接近于kafka的消费模式，并且有了backup机制，并且还没有rebalance！为什么没有rebalance？因为无状态broker端做了hash映射，已推送但是没有ack数据按原hash方式，新来数据按新hash方式；

第三种 - 共享模式（Shared）
消息通过 round robin 轮询机制（也可以自定义）分发给不同的消费者，并且每个消息仅会被分发给一个消费者。当消费者断开连接，所有被发送给他，但没有被确认的消息将被重新安排，分发给其它存活的消费者。

在Shared订阅模式下，多个消费者可以附加到同一个订阅，消息以循环分发的方式轮流发送给各个消费者，并且任何给定的消息只会传递给一个消费者，当一个消费者断开连接时，所有发送给它并未被确认的消息将重新调度，再发送给其他消费者。

Shared模式的订阅图解如下所示：

上图中的ConsumerA、ConsumerB、ConsumerC都会参与消息消费。

Shared模式与Failover模式的主要差别是Shared模式并不和消费者绑定队列，即Shared模式将所有分区的消息当成一个整体来看，

使用Shared模式需要的几个注意事项：

无法保证消息的顺序性
Shared模式不能使用累积确认机制。
这种模式的最大缺点是不能启用累积确认机制，消息确认效率会降低，但其优势也比较明显，在解决单个队列积压方面，能充分所有消费者的处理能力。我们的VDC引擎调度就是按照该模式构建一个支持优先级的调度中心。

第四种 - KEY 共享模式（Key_Shared）
在Key_Shared模式中，多个消费者可以附加到同一个订阅。具有相同Key的消息会分发给同一个消费者。

Key_Shared模式的消息分发机制如下所示：

Pulsar提供了Sticky（粘性）、Auto-split Hash Range（自动分割哈希范围）、Auto-split Consistent Hashing（自动分割一致性哈希）这三种选择算法。

选择消费者的基本过程如下所示：

将分片Key传递到一个哈希函数，生成一个哈希值
将Key的哈希值传入到对应的分片算法中，从而选择出一个消费者。
注意：
Key_Shared 本身在使用上存在一定的限制条件，由于其工程实现复杂度较高，在社区版本迭代中，不断有对 Key_Shared 的功能进行改进以及优化，整体稳定性相较 Exclusive,Failover 和 Shared 这三种订阅类型偏弱。如果上述三种订阅类型能满足业务需要，可以优先选用上述三种订阅类型。
专业集群可以保证相同 KEY 的消息按顺序投递；虚拟集群无法保障消息投递顺序。
当一个新的消费者加入或者一个消费者退出时，分配算法都将会重新计算消息到消费者的映射(选择)。

接下来分别介绍这三种分配算法底层的工作机制。

订阅模式如何选择
平时的工作当中，一般正常情况下选择Shared模式即可，每一个Consumer都会消费同一个Partition中的数据，每一条消息只会发送给一个Consumer，例如VDC的调度数据消费就是采用的这种模式，这种模式下消息是无序的，Consumer可以利用协程池进行消息的并发出，然后对批量消息进行单条数据的进行Ack即可，因为Broker上的Cursor会记录每一个consumer在该partition上的消费进度以及消息空洞信息，对失败数据会进行重发；
如果需要让相同 Key 的消息分给同一个消费者，或者说是需要保证消息的顺序性，这个时候 Shared 订阅模式无法满足需求了。有两种方式可供选择：Key_Shared订阅模式 或者 通过多分区主题 + Failover 订阅模式实现，如果同时满足Key 数量多且每个 Key 的消息分布相对均匀并且消费处理速度快，无消息堆积两种条件，那么就推荐使用Key_shared模式，如果上述两个条件一个不满足，则推荐使用【多分区主题 + Failover 订阅】
如果对顺序要求严格并且对性能要求不高，也可以选用Exclusive模式， 在这种模式下需要定期对服务做好严格的健康检查；

前言

以前做ERP系统，QPS这个概念离我着实有些遥远，毕竟ERP系统主要还是公司内部特定业务的一些使用，QPS并不会很高，也就不会出现流量洪峰把服务打跪的情况，但是目前本人在做的系统QPS已经接近百万级别，如何防止流量高峰把服务打挂就是要关注的重点了，当然防止服务雪崩其实有很多方法，比如Lb、Cache、扩容等等，但是以上这些操作其实都离不开一个主题那就是资源，如果不考虑资源问题这些都是很好的方法，但是如果当我们的资源已经匮乏了，这时就需要下面这种保护服务的方法了— 《限流器》,限流是微服务中必不缺少的一个环节，可以很好的起到保护下游服务，防止服务过载等作用。当然限流从功能上来说也分两种，一种是框架层的限流，像springcloud中的histrix亦或者我现在使用的polaris等等，框架层限流通过半开、全开、熔断的方式进行实现，另外一种就是今天我们要讨论的业务限流，接下来就介绍下本人在项目实战中是如何进行限流的，其实使用非常简单，但我们要知其然还要知其所以然～

概念

从目前市面上的主流限流方式来看，大致可以分为漏桶和令牌桶两种限流方式，下面先对这两类限流算法进行一下简单介绍。

漏桶

什么是漏桶了？举个通俗点的例子吧，本人开个一间澡堂子，澡堂子有一个等待大厅可以容纳10个人，晚上6点来了8个人洗澡，这时候我会安排这8个人在等待大厅等候，然后每隔一分钟就放一个人进去洗澡，当我放了3个人进去后，这时等待大厅还有5个人，此时又来了6个人要洗澡，但是现在等待大厅最多能容纳5个人，那多出来的那个人咋办？两个选择要么等着要么就先行离开待会再来，这样其实就是一个简易的漏桶系统了；
其实总结下来漏桶原则就是以任意速率接受请求，然后以恒定速率放行请求，如下图所示：

优点：无法击垮服务、能很好的保证服务的健康；
缺点：因为当流出速度固定，大规模持续突发量，无法多余处理，浪费网络带宽；

令牌桶

那什么是令牌桶了？还是上面的澡堂子场景，晚上六点来了100号大汉来洗澡，那本人就站在门口接待并且每隔1分钟放10个大汉进去等待大厅，进入大厅等待的大汉自行决定要不要立刻去洗澡，可以全部人一次性全部进去洗澡，也可以慢慢悠悠坐会儿再清洗，而我则根据等待大厅的实际空位情况进行人员放行。如此便是一个简易的令牌桶系统了；
其实总结下来令牌桶就是以恒定速率接受请求，然后以任意速率放行请求，当然上限就是桶大小，如下图：

优点：支持大的并发，有效利用网络带宽，可合理处理流量激增的场景；
缺点：如果设计不合理在流量突增情况下限流效果不明显，还是有可能出现压垮服务的情况（下面会解释这种情况）；

令牌桶的原理

其实上面两种算法孰优孰劣，这里我不发表评论，因为我一贯的思想是没有最有的算法只有最优的选择，我们需要根据实际的场景进行最佳的选择，因为本人的项目是需要支持大的并发以及一些流量突增场景的，所以就选择了令牌桶算法；

一般在网上搜索令牌桶的时候，都会有一张这样的原理图：

通常看到这张图，我的第一反应是需要两个核心组件 : 一个 定时器 和两个 阻塞队列。定时器 恒定速率往令牌桶（阻塞队列2） 中放 token。用户侧请求则阻塞在阻塞队列1中排队从令牌桶（阻塞队列2）中获取token。这么做也是该图的直观反应，但是这样做不禁效率低下而且过滤复杂，不仅要维护一个 定时器 和 两个 阻塞队列，而且还耗费了一些不必要的内存。
如果我们是按照每秒往令牌桶中放token就会出现上面说到的缺点，假设我们的服务最大处理能力是150QPS，现在设置的限流器是每秒放置100个token进入令牌桶，这样看来服务会很安全，但是实际上了？来看下面的这种情况 :

如果突然有200个请求从第一秒的后0.5秒打过来，此时这200个请求毫无疑问都是会打到后端服务的，这时已经远远超过了后端服务的负载上限了，结果可想而知。

那怎样才能规避这样的问题了？不应担心，Golang 的 timer/rate使用的lazyload的方式就合理规避了上述风险，没有计时器没有阻塞队列，直到每次消费之前才根据时间差更新 token 数目，仅仅是通过计数的方式。

token 的生成和消费

1
2
3
4
5
6
7
8

// NewLimiter returns a new Limiter that allows events up to rate r and permits
// bursts of at most b tokens.
func NewLimiter(r Limit, b int) *Limiter {
​    return &Limiter{
​        limit: r,
​        burst: b,
​    }
}

在 time/rate 中，NewLimiter 的第一个参数就是速率 Limit，表示一秒钟可以产生多少 token，这里强调一下，1秒产生的token只是一个概念，并不是真正的按照1秒进行生产，不然就会有上面描述的临界时间节点的问题。
这里换算一下，就可以知道生产一个 token 的实际时间间隔是多少了，根据这个生成单个token的时间间隔，就可以得到以下的两个核心数据：
1. 生成 N 个新的 token 一共需要多久。durationFromtokens。
2. 给定一段时长，这段时间一共可以生成多少个 token。tokensFromDuration。

好了，有了以上的两个值后，接下来的流程就比较明朗了：

• 计算从上次取 token 的时间到当前时刻，期间一共新产生了多少 token：
  只在取 token 之前生成新的 token，也就意味着每次取 token 的间隔，实际上也是生成 token 的间隔。我们可以利用 tokensFromDuration, 轻易的算出这段时间一共产生 token 的数目。
  那么，当前 token 数目 = 新产生的 token 数目 + 之前剩余的 token 数目 - 要消费的 token 数目。
• 如果消费后剩余 token 数目大于零，说明此时 token 桶内仍不为空，此时 token 充足，无需调用侧等待。
  如果 token 数目小于零，则需等待一段时间。
  那么这个时候，我们可以利用 durationFromtokens 将当前负值的 token 数转化为需要等待的时间。
• 将需要等待的时间等相关结果返回给调用方。
• 根据实际的需要等待时间决定是否放弃该请求。

可以看出，其实整个过程就是利用了 token 数可以和时间相互转化 的原理。而如果 token 数为负，则需要等待相应时间即可。

注意 如果当获取token时，令牌桶中的 token 数目已经为负值了，依然可以按照上述流程进行消费。随着负值越来越小，等待的时间将会越来越长。当然，读写共享内存的行为为了保证线程安全，更新令牌桶相关数据时都用了 mutex 加锁。

举一个实际例子来看下token桶是如何运转的吧：

1. 初始化后，桶内 token 数为 5, 此时 A 线程请求 4 个 token。那么此时桶内 token 是充足的，因此 A 线程不需要等待，直接获取token。且此时桶内 token 数变为 1。
2. 同时，B 线程请求 5 个 token，此时桶内 token 数为 1，所以此时 B 线程需要等待 -1 + 5 = 4 个 token 的时间，且此时桶内 token 数变为 -4。
3. 最后是C线程也在此时请求1个token，此时桶内的token数为-4，所以此时C线程需要等待 4 + 1 = 5个token的事件，且此时桶内的token数变为 -5。后面的情况就以此类推了，是不是很简单。

通常在实际的常用手法中有两个比较重要的方法就是Allow和Wait两个：

• Allow ：只需判断需要等待的时间是否为 0 即可，如果大于 0 说明需要等待，则会立即返回 False，反之立即返回 True。
• Wait ：需要根据t := time.NewTimer(delay)，等待对应的时间，如果等待里时间没ctx没有超时则ok，饭制则fail。

float 精度问题

这其实是一个语言的数值问题，并非是限流器的设计问题，就是在 token 和时间的相互转化函数 durationFromtokens 和 tokensFromDuration 中，会涉及到 float64 的乘除运算。但凡遇到 float 的乘除，丢失精度问题就如吃饭喝水般常见。

Golang 在这里也踩了坑，以下是 tokensFromDuration 最初的实现版本

1
2
3

func (limit Limit) tokensFromDuration(d time.Duration) float64 {
	return d.Seconds() * float64(limit)
}

这个操作看起来一点问题都没：每秒生成的 token 数乘于秒数。
然而，这里的问题在于，d.Seconds() 已经是小数了。两个小数相乘，会带来精度的损失。

修改后新的版本如下：

1
2
3
4
5

func (limit Limit) tokensFromDuration(d time.Duration) float64 {
	sec := float64(d/time.Second) * float64(limit)
	nsec := float64(d%time.Second) * float64(limit)
	return sec + nsec/1e9
}

time.Duration 是 int64 的别名，代表纳秒。分别求出秒的整数部分和小数部分，进行相乘后再相加，这样可以得到最精确的精度。

数值溢出问题

回到上面的实现，计算从当前时间到上次取 token 的时刻，期间一共新产生了多少 token，同时也可得出当前的 token 是多少。

老实说，对于这个需求我的第一实现思路是：

1
2
3
4
5
6
7
8
9

// gapTime 表示过去的时间差
gapTime := now.Sub(limit.last)
// gapTokens 表示这段时间一共新产生了多少 token
gapTokens = tokensFromDuration(now.Sub(limit.last))

nowTokens := limit.tokens + gapTokens
if(nowTokens > limit.burst){
	nowTokens = limit.burst
}

limit.tokens 表示当前剩余的 token数，limit.last 是上次取 token 的时间。limit.burst 是 token 桶的固定容量。
使用 tokensFromDuration 计算出距上次获取token后的时间间隔内新生成了多少 token，累加起来后，不超过桶的固定容量即可。

这个方法乍一看确实挑不出什么毛病，但它是不是真的没什么问题了？先来看一下在 time/rate 里面是怎么做的吧，如下：

1
2
3
4
5
6
7
8
9
10
11
12

maxElapsed := lim.limit.durationFromtokens(float64(lim.burst) - lim.tokens)
elapsed := now.Sub(last)
if elapsed > maxElapsed {
	elapsed = maxElapsed
}

delta := lim.limit.tokensFromDuration(elapsed)

tokens := lim.tokens + delta
if burst := float64(lim.burst); tokens > burst {
	tokens = burst
}

发现没与上面我的实现思路是有不一样的，它并没有直接使用 now.Sub(lim.last) 来计算这个时间差内应该生产的token数，而是
用了 lim.limit.durationFromtokens(float64(lim.burst) - lim.tokens)做兜底，计算把桶填满的时间 maxElapsed。
取 elapsed 和 maxElapsed 二者的最小值。这里的实现着实有些惊讶，但好像又在情理之中；

现在回过头来再思考一下，对于我的设计思路，如果直接使用now.Sub(lim.last)这个值，当now的值非常大或者last的值非常小的时候，这个间隔时间就会变得异常的大 ，如果 lim.limit 也被设置的非常大后，直接将二者相乘，计算结果有可能就会溢出。当然如果你一定要说lim.limit.durationFromtokens(float64(lim.burst) - lim.tokens)这个值也有可能非常大也没毛病，不反驳-_-!!!。

总结，time/rate 就是使用了lim.limit.durationFromtokens(float64(lim.burst) - lim.tokens)来规避了数值溢出的问题。

使用

使用相对就比较简单了，通常在项目初始化时会构建一个limiter放入内存中

1
2
3
4
5
6

func registerGconfig(gConfig *Config) {
	GConfig = gConfig
	tempObtainURLLimiter := rate.NewLimiter(rate.Limit(gConfig.ObtainURLTokenPerSecond),
		gConfig.ObtainURLTokenPerSecond)
	ObtainURLLimiter = tempObtainURLLimiter
}

然后再业务函数中直接使用即可

1
2
3
4
5
6

if err := g.ObtainURLLimiter.Wait(ctx); err != nil {
		metrics.Counter("ObtainDownloadURLWaitErrCount").Incr()
		log.InfoContextf(ctx, "Vdc_Operation | DoObtainDownloadURL | LimiterWaitErr:[%+v]", err)
		servererr.ErrRspObtainDownloadURL(rsp, servererr.NewWithMsg(servererr.ErrDoObtainDownloadURL,             err.Error()))
		return nil
}

最后为了更动态的控制速率，我们会引入服务配置平台，通过长链接和watch机制实现，在配置平台调整速率后下发配置，服务就会自动更新一个新的limiter放入内存中，这样我们就可以根据实时服务负载监控动态的对限流速率进行调整了。

来看一张使用了限流器的服务吞吐量监控图，会发现服务在流量高峰的吞吐量都是非常平稳的～ :

总结

令牌桶是非常适合互联网突发式请求场景的，请求 token 时并不是像漏桶一样严格的限制为固定的速率，而是使用一个完整的桶容量作为缓冲。只要桶中还有 token，请求就可以一直进行，也就可以处理相当于整个桶容量的流量高峰。但是当流量激增到一定程度后，就会按照设置的token放置速率进行处理，通常情况下我们会将桶容量和速率调整为同一个数值。

我的博客即将同步至腾讯云开发者社区，邀请大家一同入驻：https://cloud.tencent.com/developer/support-plan?invite_code=1x761d0b26y42

前言

如何设计一个高性能三地系统

####

前言

前文对kafka进行了一些粗略的介绍，相信大家对MQ届的老牌明星是有了一定了解的，本文将对MQ届的闪耀新星Apache Pulsar进行介绍，分为上（原理篇）下（实践篇）两个篇章。准备好了就开冲～

Pulsar的整体架构

基础架构

Apache Pulsar 当然也是一款MQ，它是Pub/Sub型的消息系统，但是从架构设计上来说与前文介绍的kafka是完全不同的，Pulsar在结构上将计算与存储完全分离。先看下图来初步认识下Pulsar的整体架构：

计算与存储的分离

对上图中的名词来一一解释一下

Apache Pulsar 主要包括 Broker, Apache BookKeeper, Producer, Consumer等核心组件。

• Broker：无状态（stateless）服务层，只负责接收和传递消息、集群负载均衡等工作，Broker 不存储任何元数据信息转由zookeeper存储元数据信息，因此可以快速的上、下线；
• Apache BookKeeper：有状态（stateful）持久层，由一组名为 Bookie 的存储节点组成，Broker层接收到的消息持久化于此处；
• Producer ： 消息生产者，负责生产消息到Broker 的 Topic；
• Consumer：消息消费者，负责从 Topic 订阅数据，这里值得注意的一点是：Pulsar是push模式；

除了上述的组件之外，Apache Pulsar 还依赖 Zookeeper 作为元数据存储。与传统的消息队列（Kafka）相比，Apache Pulsar 在架构设计上采用了计算与存储分离的模式，Pub/Sub 相关的计算逻辑在 无状态的Broker 上完成，数据存储在 Apache BookKeeper 的 Bookie 节点上。

具体如何存储

讲了这么多，那Apache Pulsar在存储上面具体是怎样设计的了？来看下图， 在存储设计上Pulsar也不同于传统 MQ 的分区数据本地存储的模式，采用的是逻辑分区物理分片存储的模式，存储粒度比分区更细化、存储负载更均衡。 Apache Pulsar 中的每个 Topic 分区本质上都是存储在 Apache BookKeeper 中的分布式日志。Topic 可以有多个分区，分区数据持久化时，分区（Partition）是逻辑上的概念，实际存储的单位是分片（Segment）的，例如下图中的一个分区 Topic0-Part1 的数据由多个 Segment 组成， 每个 Segment 作为 Apache BookKeeper 中的一个 Ledger，根据每一个Bookie的负载情况均匀分布并存储在 Apache BookKeeper 群集中的多个 Bookie 节点中， 每个 Segment 具有 3 个副本，当然这是一个可配参数–write quorum。

写到这里，让我们先来总结一下：

Kafka采用的是物理分区模型如下图左 :

Pulsar采用的是逻辑分区物理分片的模型如下图右 :

从云原生的角度来讲肯定是右下的分区模型更为合适的，因为这种模型更适合动态上下线以及动态扩容。

Pulsar架构的优势

Pulsar的这种架构设计优点有如下几点：

• Broker(计算) 和 Bookie(存储) 相互独立，方便实现独立的扩展以及独立的容错；
• Broker 无状态，便于快速上、下线，更加适合于云原生场景；
• 底层采用分区存储不受限于单个节点存储容量；
• 分区数据由于有负载的存在数据分布均匀；

高扩展性

由上面的介绍可知，Pulsar的计算与存储是完全独立的，所以在系统的扩展性上面也是可以分开来看待的.

Broker

Broker是无状态（Stateless）的，可以通过增加节点和删除节点的方式实现快速扩缩容。当流量陡增你需要更多的生产者和消费者来提升系统的吞吐量时，我们可以简单方便地添加 Broker 节点来满足业务需求。Pulsar 支持自动的分区负载均衡，在 Broker 节点的资源使用率达到阈值时，会将负载迁移到负载较低的 Broker 节点，这个过程中分区也将在多个 Broker 节点中做平衡迁移，一些分区的所有权会转移到新的Broker节点。

Bookie

存储层的扩容则由Bookie节点来掌控，通过资源感知和数据存放策略，流量将自动切换到新的 Bookie 节点中，整个过程不会涉及到不必要的数据搬迁，即不需要将旧数据从现有存储节点重新复制到新存储节点， 要知道kafka在这件事情上面是需要复制整个partition数据的。

见上图，先大致了解一下存储扩容后数据存放的逻辑，例如起始状态有5个存储节点，Bookie1, Bookie2, Bookie3, Bookie4, Bookie5，以 Topic0-Part1来说，当这个分区的最新的存储分片是 SegmentX 时，此时由于某种原因对存储层bookie进行扩容，添加了两个新的 Bookie 节点，Bookie6和Bookie7，那么在存储分片滚动之后，新生成的存储分片， SegmentY和SegmentZ就会会优先选择新的 Bookie 节点（Bookie6,Bookie7）来保存数据，其实整个过程中每个Segment也是需要有一个副本数保证的，上文提到过的write quorum参数控制，在后续的Bookkeeper介绍中我们会详细介绍这几个参数，这里就暂时先提一下，其实在BK层是有三个核心参数的，含义分别如下：

• ensemble size : 在初始化 Ledger 时， 首先要选取一个 Bookie 集合作为写入节点，ensemble 表示这个集合中的节点数目
• write quorum : 数据备份数目
• ack quorum : 响应节点数目

高容错性

扩展说明的是扩展机器，而容错则是说明当集群中的机器由于人为或者一些不可抗因素导致宕掉之后，服务依然能够像没事一样对外稳定提供服务，由于计算存储分离的原因，所以容错性依旧从以下两个方面来看，各自独立。

Broker

先来看看当Broker集群中的某一台机器突然挂掉了，会发生什么了？如下图 :

前面我们说了Broker是一个无状态节点，它只负责计算不实际存储数据甚至连元数据信息都没有，所以当Broker-1突然挂掉之后，原本它负责的Topic0-Part1的所有权会根据负载情况自动转移给当前负载较轻的Broker-2节点，整个过程非常的轻便只是发生了一次所有权（ownership）的转移，底层存储并没有任何的影响，也不存在任何的数据拷贝。

Bookie

接下来再看看当某一个存储层的Bookie节点发生无故宕机时会发生什么，如下图所示 :

因为Bookie是实际存储数据的地方，所以当Bookie2挂了之后，存储在它上面的数据会根据当前Bookie集群的负载情况对数据按照Segment的级别进行分发，Apache BookKeeper 中的副本修复是 Segment 级别的多对多快速修复，所有的副本修复都在后台进行，对Broker和应用透明，Broker 会产生新的Segment 来处理写入请求，不会影响分区的可用性，至于为什么要按照这样的方式分发，我猜想是为了维持最少副本数的体制内规则。

分区存储的优势在哪

总结一下，其实上从面的分析来看结论其实已经很明显了，分片存储在很大程度上解决了分区容量受单节点存储空间限制的问题，当容量不够时，可以通过快速扩容 Bookie 节点的方式支撑更多的分区数据， 因为新数据的写入会优先考虑新增的Bookie节点并且数据会均匀的分配在 Bookie 节点上，不会造成单点压力过大。 另外Broker 和 Bookie 高容错架构以及无缝的扩容能力让 Apache Pulsar 具备非常高的可用性。

Pulsar特性

读写模型

之前讲Kafka的时候也说过了，一个MQ的核心无非就是读写存，那接下来就来看看Pulsar的读写模型，Pulsar在其实现上最大的优势就是读写分离，那具体是怎么带来好处的，还是先来看张图 :

不要慌，初次看到这个图我也是懵逼的但其实本图表达的内容非常简单，容我向诸位一一解释，这张图就是很统筹的介绍了Pulsar的整个读写过程，对图中的名词来进行一下解释：

• Journals：中文可以叫它期刊，Journal 文件包含了 BookKeeper事务日志，在 Ledger 更新之前，Journal 保证描述更新的事务写入到 Non-volatile 的存储介质上
• Entry logs：Entry Logger管理写入的 Entry，来自不同 Ledger 的 Entry 会被聚合然后顺序写入
• Index files：每个 Ledger(账本)都有一个对应的索引文件，记录数据在 Entry 日志文件中的 Offset 信息，这种方式在介绍kafka的时候讲过了

写过程解释 :

1. 数据首先会写入 Journal，写入 Journal 的数据会实时落到磁盘，如果你了解Mysql，那应该会想起RedoLog的作用。
2. 数据落Journal Disk成功后会写入到 Memtable ，Memtable 就是一个是读写缓存。
3. 在写入 Memtable 成功之后，就会对写入者进行Ack响应。
4. 最后在 Memtable 写满或者是达到某一个阈值之后，就会 Flush 到 Entry Logger 和 Index cache中，Entry Logger 中保存了实际的消息数据，Index cache 保存了数据的索引信息，最后由后台线程将 Entry Logger 和 Index cache 数据异步落到磁盘。

读过程解释 :

1. Tailing read (跟踪读），会直接从 Memtable 缓存中读取 Entry ， 速度非常快。

2. Catch-up read（滞后读），会先读取 Index信息，然后根据index信息再从 Entry Logger 文件读取 Entry，会有磁盘io的发生，速度较慢。

   这里提出一个问题，那是什么样的场景下会分别用到上面两种读方式了？

Ok，现在回头再来看看我的中文翻译版，相信这张图此时对你就没什么难度了 :

很明显在上述的读写分离的模型中，能看出只有Journal的写入是需要同步的将每一条日志进行落盘处理的，这一点决定的pursal是否会在写入时丢消息，所以一般情况下推荐Journal要使用性能较高的SSD磁盘，至于读区间的磁盘也是需要与其隔离开来的避免相互的性能影响的。所以，数据写入主要是受 Journal 磁盘的负载影响，不会受Ledger 磁盘的影响。当然这并不是说Ledger 磁盘也不需要性能考虑，只是说 Journal 磁盘的性能更为重要～

消息的消费模型

其实之前的一篇简文中也介绍过了，这里就再总结介绍一下：

不同的消费模型决定了每一条消息的处理方式，Pulsar提供了四种消费模型 分别是独占（Exclusive），故障切换（Failover），共享（Share）以及 key共享（key-Share）,上图给出了前三种的处理方式，至于第四种其实与第三种是一样的，只是多了一项可以根据key分发到指定的消费者。解释一下 ：

• Exclusive ：永远都有且只能有一个消费者组（订阅）中有且只有一个消费者来消费 Topic 中的消息。

• Failover ：多个消费者（Consumer）可以同时订阅同一个Topic。 但是，一个订阅中的所有Consumer，有且只有一个Consumer被选为该订阅的主Consumer。 其他Consumer将被指定为故障转移Consumer也就是备胎。 当主Consumer断开连接或者发生故障时，分区将被重新分配给其中一个备胎，而新分配的Consumer将成为新的主消费者获得转正。 如果此时故障Consumer中存在没有ack的消息，那这些消息将会被重传至最新的主Consumer。

• Share ：又叫共享订阅，是实际工作中使用最多的一种模式，同一个订阅用户按照应用的需求挂载任意多的消费者。 订阅中的所有消息以循环分发形式发送给订阅背后的多个消费者，并且一个消息仅传递给一个消费者。当消费者断开连接时，所有传递给它但是未被确认（ack）的消息将被重新分配和组织，以便发送给该订阅上剩余的剩余消费者。

  补充一点，在kafka中是用Consumer Group为单位来进行消息消费的，而在Pulsar中则是用Subscription为单位。现在来回答上面提出的一个问题 ：那是什么样的场景下会分别用到上面两种读方式了？Exclusive和Failover这种stream处理模式会大量用到Tailing read (跟踪读），而Share这种Queun模式则会大量的使用Catch-up read（滞后读）。

消息的确认机制

确认机制是所有消息队列中非常核心的一步，它可以避免消息的重复投递以及消息是否投递成功，Pulsar的消息确认总体分为Client确认和Broker确认，下面来一一详细介绍一下！

客户端确认

通过上面所述我们知道在Pulsar中有多种消费模式，如：Share、Key_share、Failover和Exclusice，但无论使用哪种消费模式其实都会创建一个Subscription(订阅)。Subscription分为持久化订阅（Persistent-subscription）和非持久化订阅（Nonpersistent-Subscription），对Persistent-Subscription而言，Broker上会有一个对应持久化的游标（Cursor），记住这个名词，它是Pulsar中非常重要的一个对象，上文提到过元数据被记录在ZooKeeper，因此Cursor也不例外。Cursor以Subscription为单位，保存了当前Subscription已经消费到哪个位置了。因为不同Consumer使用的Subscription模式不同，可以进行的Ack行为也不一样。总体来说可以分为以下几种Ack场景：

• 单条确认（Individual Ack），单独确认一条消息。 被确认后的消息将不会被重新传递。和kafka不同，Pulsar的一个Partition是允许被多个消费者消费的。这里假定消息1、2发送给了Consumer-A，消息3、4发送给了Consumer-B，消息5、6发送给了Consumer-C，其中Consumer-C消费的比较快，先Ack了消息5，此时Cursor中会单独记录消息5为已Ack状态。如果其他消息都被消费，但没有被Ack，并且三个消费者都下线或Ack超时，则Broker会只推送消息1、2、3、5、6，已经被Ack的消息4不会被再次推送。
• 累积确认（Cumulative Ack），通过累积确认，消费者只需要确认它收到的最后一条消息。与上面的单条确认有差异，假设Consumer-A接受到了消息1、2、3、4、5、6，为了提升Ack的性能，Consumer 并不需要去对这每一条消息都进行ack，而是只需要调用一次AcknowledgeCumulative，然后把位置最后的消息6传入，那Broker会把消息6以及之前的消息全部标记为已Ack，只需要一次Ack即可。
• 累积确认（Cumulative Ack）中的单条确认。这种消息确认模式，调用的接口和单条消息的确认一样。与上面的累积确认一致，只是它可以指定确认某一批消息中的某几条，例如Consumer-A拿到了一个批消息，里面有消息1、2、3、4、5、6，如果不开启Broker中的AcknowledgmentAtBatchIndexLevelEnabled，那就只能消费整个Batch后再统一Ack，否则Broker会以批为单位重新全部投递一次。前面介绍的选项开启之后，我们可以通过Acknowledge方法来确认批消息中的单条消息。
• 否定确认（NegativeAcknowledge）客户端发送一个RedeliverUnacknowledgedMessages命令给Broker，明确告知Broker这条消息我这个客户端暂时处理不了等会再来试试，那么消息将会被重新投递，讲到这里其实又延伸出了另外一个MQ中的核心话题—消息重试，下一张实践篇会详细解析Pulsar中的消息重试方案实现。

看下图中上下分别代表累计确认和单条确认（浅灰色框中的消息代表被ACK）。在累计确认中，M10 之前的消息被标记为 Acked。在单条确认中，仅仅只是确认消息 M5 和 M9， 当消费者失败或者重启等情况下，除了 M5 和 M9 之外，其他所有消息将被重新传送至消费端。

提出一个问题～之前不是说过Pulsar中的一个Partition中的数据是可以被多个Consumer消费的吗，假定这样的场景消息1、2发送给了Consumer-A，消息3、4发送给了Consumer-B，消息5、6发送给了Consumer-C，此时还能用批量确认吗？答案显示是不能的，如果Consumer-C处理速度较快而A和B对应的消息是会处理失败的，这时C处理完了发了一个6的Ack，那不是把A和B的消息也给Ack了吗？这显然是不对的，所以在Pulsar中订阅模式与消息确认之间的关系有如下所示关系：

Acknowledge和AcknowledgeCumulative实现原理

从性能的角度出发，Pulsar在做任何Ack操作时都不会一条条给Broler发送Ack信号，而是把请求转交给AcknowledgmentsGroupingTracker处理。记住它，这是本文介绍的第一个Tracker，它只是一个接口，接口下有两个实现，一个是持久化订阅的实现，另一个是非持久化订阅的实现。由于非持久化订阅的Tracker是空实现，就没有介绍的必要了，本文只对persistent-Subsription的实现——PersistentAcknowledgmentsGroupingTracker来做介绍。

上面提到了从性能的角度考虑，Tracker是默认批量确认的，即使是单条消息的确认，也会先进入Consumer本地缓冲队列，然后再一批次的发往Broker。在创建Consumer时可以设置参数AcknowledgementGroupTimeMicros，如果设置为0，则Consumer每次都会立即发送确认请求。所有的单条确认(IndividualAck)请求会先放入一个名为PendingIndividual Acks的Set，默认是每100ms或者堆积的确认请求超过1000，则发送一批确认请求

对于Batch消息中的单条确认（IndividualBatchAck），则用一个名为PendingIndividualBatchIndexAcks的Map进行保存，而不是普通单条消息的Set。这个Map的Key是Batch消息的MessageId，Value是一个BitMap，记录这批消息里哪些需要Ack。使用BitSet能大幅降低保存消息Id的能存占用，1KB==1024*8bit 就能记录8192个消息是否被确认。由于BitMap保存的内容都是非0即1，因此可以很方便地保存在堆外，BitMap对象也做了池化，可以循环使用，不需要每次都创建新的，对内存非常友好，顺带提一句BloomFilter也是用BitMap来实现的。

对于累计确认（CumulativeAck）实现方式就相对比较简单了，Tracker中只保存最新的确认位置点即可。

最后就是Tracker的Flush，所有的确认最终都需要通过触发Flush方法发送到Broker，但无论是哪种确认，Flush时创建的都是同一个命令并发送给Broker，不过是传参中带的AckType会不一样。来看一张总结图加强一下认识：

NegativeAcknowledge的实现

NegativeAcknowledge一般又叫NAck，同上面一样，NAck和其他消息确认方式是一样的，并不会单条立即请求Broker，而是把请求转交给NegativeAcksTracker进行处理。记住它，这是本文介绍的第二个Tracker，该Tracker中记录着每条消息以及需要延迟的时间。Tracker复用了PulsarClient的时间轮，默认是33ms左右一个时间刻度进行检查，默认延迟时间是1分钟，抽取出已经到期的消息并触发重新投递。Tracker主要存在的意义是为了合并请求。另外如果延迟时间还没到，消息会暂存在内存，如果业务侧有大量的消息需要延迟消费，还是建议使用ReconsumeLater接口。NegativeAck唯一的好处是，不需要每条消息都指定时间，可以全局设置延迟时间。

未确认消息的处理

在讨论以下问题之前，需要先来了解一下一个新的概念叫预拉取，其实Consumer从Broker拿消息时并不是一条一条拿的，而是Broker出于性能考虑一次性推送一批数据至Consumer然后放在一个本地预拉取队列(ReceiveQueue)里面，该队列的大小由参数ReceiveQueueSize控制，由这里的逻辑我们要知道在创建消费者设置ReceiveQueueSize参数时需要十分慎重，避免大量的消息堆积在某一个Consumer的本地预拉取队列，而其他Consumer又没有消息可消费，造成其他Consumer饥饿。见下图 :

了解了预拉取之后现在我们就可以来探讨下面的问题了，如果消费者获取到消息后一直不Ack也不NAck会怎么样？这个问题需要分为以下两种情况来讨论 :

消息已存ReceiveQueue并且Logic已经调用了Receive方法，或者已经回调了正在异步等待的消费者

此时消息的引用会被保存进UnAckedMessageTracker，记住它这是本文介绍Consumer里的第三个Tracker。UnAckedMessageTracker中维护了一个时间轮，时间轮的刻度根据AckTimeout、TickDurationInMs这两个参数生成，每个刻度时间=AckTimeout / TickDurationInMs。新追踪的消息会放入最后一个刻度，每次调度都会移除队列头第一个刻度，并新增一个刻度放入队列尾，保证刻度总数不变。每次调度，队列头刻度里的消息将会被清理，UnAckedMessageTracker会自动把这些消息做重投递。重投递就是客户端发送一个RedeliverUnacknowledgedMessages命令给Broker。每一条推送给消费者但是未Ack的消息，在Broker侧都会有一个集合来记录（PengdingAck），这是用来避免重复投递的。触发重投递后，Broker会把对应的消息从这个集合里移除，然后这些消息就可以再次被消费了。注意，当重投递时，如果消费者不是Share模式是无法重投递单条消息的，只能把这个消费者所有已经接收但是未ack的消息全部重新投递。下图是一个时间轮的简单展示，简单了解一下就行：

消息已存ReceiveQueue但是Logic还未调用Receive方法

此时消息会一直堆积在本地队列ReceiveQueue中。预拉取队列可以在创建消费者时通过ReceiveQueueSize参数来控制预拉取消息的数量。Broker侧会把这些已经推送到Consumer本地的消息记录到PendingAck中，并且这些消息也不会再投递给别的消费者，且不会Ack超时，除非当前Consumer宕机或被关闭，消息才会被重新投递。Broker侧有一个Redelivery Tracker接口，暂时的实现是内存追踪（InMemoryRedeliveryTracker）。这个Tracker会记录消息到底被重新投递了多少次，每条消息推送给消费者时，会先从Tracker的哈希表中查询一下重投递的次数，和消息一并推送给消费者。

最后提一点，PulsarClient上可以设置启用ConsumerStatsRecorder，启用后，消费者会在固定间隔会打印出当前消费者的metrics信息，例如：本地消息堆积量、接受的消息数等，方便业务排查性能问题。

Broker确认

客户端Ack或者NAck的消息最终都会发回确认信息到Broker，那Broker侧是如何进行处理的了？这里就是上文提到的游标（Cursor）的舞台了，其实客户端通过消息确认机制通知Broker哪一些消息已经被消费，告知后面就不要再重复发送这些消息了。Broker侧就是使用Cursor游标来存储当前订阅的消费位置信息，包含了消费位置中的所有元数据，避免Broker重启后，消费者要从头消费的问题。上文其实也说了Pulsar中的订阅分为持久订阅（Persistent-subscription）和非持久订阅（Nonpersistent-subscription），区别就是：持久订阅的游标（Cursor）是持久化的，元数据会保存在ZooKeeper，而非持久化游标只保存在Broker的内存中在broker重启后就有可能会重复发数据。

什么是游标(Cursor)

之前说过在Pulsar中任何统计都是以Subscription（订阅）为单位的，所以Cursor（游标）也不例外，比如多个消费者持有同一个订阅名（在kafka中叫ConsumerGroup（消费组）），那这些消费者们就会共享一个游标。游标的共享又和消费者的消费模式有关，如果是Exclusive或者FailOver模式的订阅，那同一时间只有一个消费者可以使用这个游标。如果是Shared或者Key_Shared模式的订阅，那多个消费者会同时共享这个游标。

接下来就来介绍一下当消费者Ack一条消息时Broker中的游标会有一些什么样的变化～每当消费者Ack一条消息，游标中指针的位置有可能会变化也有可能不会发生变化！！！这并不是废话文学，要理解这句话就需要先来回顾一下上一个章节介绍的客户端确认方式，在客户端确认中我们介绍了Ack的几种方式，分别是单条消息确认（Acknowledge）、批消息中的单个消息确认（Acknowledge）、累积消息确认（AcknowledgeCumulative）和否定应答（NAck），但是因为Nack不会涉及游标的变化，所以这里就不做讨论了。

我们先看单条消息的确认，如果是独占式的消费，每确认一条消息，游标位置都会往后移动一个Entry（这里假设客户端Ack一条消息就会发回到Broker，实际上并不是）如下图所示：

累积消息确认，只需要确认一条消息，里面包含了该次确认的最大值，游标可以往后移动多个Entry，比如：Consumer累积确认了Entry-5，则从0开始的Entry都会被确认，如下图所示：

对于share的消费模型，因为有多个消费者可以同时消费消息，因此消息的确认可能会出现空洞，空洞的形成和去除如下图所示：

这里也解释了为什么MarkeDeletePosition指针的位置是可能发生变化，我们可以从share的消费模式中看出，消息确认是完全有可能出现空洞的，只有当前面所有的Entry都被消费并确认，MarkeDeletePosition指针才会移动。如果存在空洞，MarkeDeletePosition指针是不会往后移动的。那这个MarkeDeletePosition指针和游标是什么关系呢？首先游标（Cursor）是一个对象，里面包含了多个属性，MarkDeletePosition指针只是游标的其中一个属性。正如上面所说的Ack空洞，在游标中有另外专门的属性进行存储。假设如果不单独存储空洞，那Broker重启后，消费者只能从MarkDeletePosition开始消费，此时还是会存在重复消费的问题，例如上图中的上半部分Entry-3和Entry-4就会被重复推送给消费者。

最后来看看游标Cursor到底记录了一些什么重要的元数据，下面只列出本人认为的核心属性：

诶？这里既然用到了CursorLedger，那说明数据都已经被保存到了Bookkeeper中了啊。看到这里聪明的你应该会有一些困惑，既然数据都保存到Bookkeeper中了，那ZooKeeper中保存的Cursor信息又有什么用呢？其实在ZooKeeper中保存的游标信息只是一些索引等轻量级信息并不包含实际的数据，索引Index信息大致是包含以下几个属性：

• 当前的CursorLedger名以及ID，用于快速定位到Bookkeeper中的Ledger；
• LastMarkDeleteEntry，最后被标记为删除的Entry信息，里面包含了LedgerId和EntryId；
• 游标最后的活动时间戳。

游标保存到ZooKeeper的timing有几个：

• 当Cursor被关闭时；
• 当发生Ledger切换导致cursorLedger变化时；
• 当持久化空洞数据到Bookkeeper失败并尝试持久化空洞数据到ZooKeeper时。

其实可以把ZooKeeper中的游标信息看作检查点（Check Point）或者索引快照（IndexSnapshot），当恢复数据时，会先从ZooKeeper中恢复元数据，获取到Bookkeeper Ledger信息，然后再通过Ledger恢复最新的LastMarkDeleteEntry位置和空洞信息。既然游标并不是实时往ZooKeeper中写入数据的，那Pulsar是如何保证消费位置不丢失的呢？其实Bookkeeper中的一个Ledge是能写很多的Entry的，所以高频的保存操作都由Bookkeeper来承担了，换句话说就是专业的事情交给专业的人，ZooKeeper只负责存储低频的轻量级索引更新，至于更新的timing上面已经讲过了。

消息空洞

什么是消息空洞？这是我们上面说到的一个高频词汇了，在游标Object中，使用了一个叫IndividualDeletedMessages的容器来存储所有的空洞信息。Broker中直接使用了Guava Range这个库来实现空洞的存储。看如下一个例子，假设在Ledger-5中的空洞如下：

那么就会用如下的区间表达方式来存储空洞信息 : [ (5:-1, 5:5] , (5:6, 5:7] ]

使用区间表达的好处就是可以用很少的区间数来表示整个Ledger的空洞情况，而不需要每个Entry都记录（但是这并不是绝对的，有一些特殊的情况即便使用区间也是会效率低下的）。当某个范围都已经被消费且确认了，就会出现两个区间merge为一个区间，这都是Guava Range自动支持的能力。如果从当前MarkDeletePosition指针的位置到后面某个Entry为止，都连成了一个连续的区间，那么MarkDeletePosition指针就可以往后移动了，直到后面的某个Entry为止。

虽然记录了这些消息空洞的信息，但是具体是如使用这些信息来避免消息重复消费的呢？

当Broker从Ledger中读取到消息后，会进入一个清洗阶段，如：过滤掉延迟消息等等。在这个阶段，Broker会遍历所有消息，看消息是否存在于Range里，如果存在，则说明已经被确认过了，这条消息会被过滤掉，不再推送给客户端。Guava Range提供了Contains接口，可以快速查看某个位置是否落在区间里。这种Entry需要被过滤的场景，基本上只会出现在Broker重启后，此时游标信息刚恢复。当ReadPosition超过了这段空洞的位置时，就不会出现读到重复消息要被过滤的情况了。

那么现在来看看IndividualDeletedMessages这个容器的实现原理：

IndividualDeletedMessages 的类型是LongPairRangeSet，默认实现是DefaultRangeSet，是一个基于Google Guava Range包装的实现类。另外还有一个Pulsar自己实现的优化版：ConcurrentOpenLongPairRangeSet，这一优化版的实现就是为了解决上面我们说过的那种特殊情况。优化版的RangeSet和Guava Range的存储方式有些不一样，Guava Range使用区间来记录数据，优化版RangeSet对外提供的接口也是Range，但是内部使用了BitSet来记录每个Entry是否被确认。

优化版RangeSet在空洞较多的情况下对内存更加友好。来看下上面说的那种特殊，比如有1000W的消息已被拉取，但是只有500W的消息被Ack，并且是隔条进行的Ack，这样的话就会出现50W个间隔型的空洞。此时如果还是使用Range就会非常麻烦毫无优势可言，会有500W个Range对象，看下图。而优化版的RangeSet还是使用了BitMap，每个Ack只占一位一bit而已，1kb即可表现8192个空洞。

我们可在broker.conf中，通过配置项managedLedgerUnackedRangesOpenCacheSetEnabled=true来开启使用优化版的RangeSet实现。所以，如果整个集群的subscription数比较多，游标对象的数据量是不容小视的。来回顾一下Pulsar的实现方式，MetaDataStore中只保存了游标的索引信息，即记录了游标的具体数据是存储在bk上的哪个Ledger中。真正的游标数据会通过上面介绍的CursorLedger写入到Bookkeeper中持久化。整个游标对象会被写入到一个Entry中，其Pb的定义如下：

1
2
3
4
5
6
7

message PositionInfo {   
  required int64 ledgerId = 1;   
  required int64 entryId = 2;   
  repeated MessageRange individualDeletedMessages = 3;   
  repeated LongProperty properties = 4;   
  repeated BatchedEntryDeletionIndexInfo batchedEntryDeletionIndexInfo = 5;
}

看到这里，其实Batch消息中单条消息确认的实现也清晰了，BatchDeletedIndexes是一个ConcurrentSkipListMap，Key为一个Position对象，对象里面包含了LedgerId和EntryId。Value是一个BitSet，记录了这个Batch里面哪些消息已经被确认。batchedEntryDeletionIndexInfo会和单条消息的空洞一起放在同一个对象（PositionInfo）中，最后持久化到Bookkeeper。

空洞数据如果写入Bookkeeper失败了，Pulsar还会尝试往ZooKeeper中保存，和索引信息一起保存。但是ZooKeeper毕竟不是专业的存储媒介所以并不会保存所有的数据，而是努力的保存一小部分，尽可能的让客户端不出现重复消费。我们可以通过broker.conf中的配置项来决定最多持久化多少数据到ZooKeeper，配置项名为：managedLedgerMaxUnackedRangesToPersistInZooKeeper，默认值是1000，最后提一点就是如果在实际的使用场景中有对幂等性强要求的还是建议在消费端做好校验。

消息空洞管理的优化方案

上文介绍的空洞存储方案看似完美，但是在海量未确认消息的场景下还是会出现一些问题的。首先是大量的订阅会让游标数量暴增，导致Broker内存的占用过大。其次，有很多空洞其实是根本是不会发生变化的，现在每次都要保存全量的空洞数据。最后，虽然优化版RangeSet在内存中使用了BitSet来存储，但是实际存储在Bookkeeper中的数据MessageRange，还是一个个由LedgerId和EntryId组成的对象，每个MessageRange占用16字节。当空洞数量比较多时，总体体积会超过5MB，而现在Bookkeeper能写入的单个Entry大小上限是5MB，如果超过这个阈值就会出现空洞信息持久化失败的情况。

目前新的解决方案中主要使用LRU+分段存储的方式来解决上述问题。由于游标中空洞信息数据量可能会很大，因此内存中只保存少量热点区间，通过LRU算法来切换冷热数据，从而进一步压缩内存的使用率。分段存储主要是把空洞信息存储到不同的Entry中去，这样能避免超过一个Entry最大消息5MB的限制。

如果我们把空洞信息拆分为多个Entry来存储，首先面临的问题是索引。因为使用单个Entry记录时，只需要读取Ledger中最后一个Entry即可，而拆分为多个Entry后，我们不知道要读取多少个Entry。因此，新方案中引入了Marker，如下图所示：

当所有的Entry保存完成后，插入一个Marker，Marker是一个特殊的Entry，记录了当前所有拆分存储的Entry，这里我们可以联想一下之前我介绍过的tlv解析。当数据恢复时，从后往前读，先读出索引，然后再根据索引读取所有的Entry。

由于存储涉及到多个Entry，因此需要保证原子性，只要最后一个Entry读出来不是Marker，则说明上次的保存没有完成就中断了，会继续往前读，直到找到一个完整的Marker。

空洞信息的存储，也不需要每次全量了。以Ledger为单位，记录每个Ledger下的数据是否有修改过，如果空洞数据被修改过会被标识为脏数据，存储时只会保存有脏数据的部分，然后修改Marker中的索引。

如下Entry-3中存储的空洞信息有修改，则Entry-3会被标记为脏数据，下次存储时，只需要存储一个Entry-3–new，再存储一个Marker即可。只有当整个Ledger写满的情况下，才会触发Marker中所有Entry复制到新Ledger的情况。如下图所示：

ManagedLedger在内存中通过LinkedHashMap实现了一个LRU链表，会有线程定时检查空洞信息的内存占用是否已经达到阈值，如果达到了阈值则需要进行LRU换出，切换以Ledger为单位，把最少使用的数据从Map中移除。LRU数据的换入是同步的，当添加或者调用Contains时，发现Marker中存在这个Ledger的索引，但是内存中没有对应的数据，则会触发同步数据的加载。异步换出和同步换入，主要是为了让数据尽量在内存中多待一会，避免出现频繁的换入换出。不过话说回来，这是Pulsar给我们提供的基建功能，非常优秀，但是在实际使用上其实是可以避免这样大规模的消息空洞出现的，下面在实践篇中也会聊到。

跨地域复制

最后就是跨地域复制功能了，它是Pulsar提供的基础功能，不同地域不同Topic之间通过可配的方式实现了数据互通，是一种全连接的异步复制，可以满足多个数据中心数据同步的使用场景。如下图所示，当在成都集群生产了一条消息，该消息会立即复制到深圳和广州集群，这样深圳广州集群的消费者不仅可以消费到自己集群生产者生产的消息，还可以消费到成都集群生产者生产的消息。

总结

写到这里，Pulsar的原理基本上就介绍完了，相信诸位对Pulsar已经有了一个全面且清晰的认识了，理解其原理对于我们用好Pulsar是十分重要的，希望这篇文章可以帮助到你，后面还有一章实践篇，主要就是本人最近在实际使用中的一些踩坑与总结～非常硬核，敬请期待……

Slogan : 香说当你遇事不顺时要相信一句话，一些都是最好的安排～

前言

其实写这篇文章主要有两个目的，第一个还是学而时习之巩固总结一下之前学过的东西，第二个为后面的pulsar介绍做个铺垫，因为本人最近就是用pulsar对系统进行了重构，作为时下最流行的两款MQ，kafka具有广大的群众基础属于是老牌明星，而pulsar就属于是后起之秀，各有千秋，话不多说进入今天的正题~

消息队列能做什么

什么是消息队列

举个简单的例子吧，最近也看了新闻了听说某地的疫情十分严重，想着就提前在家屯点货已备不时之需吧，毕竟现在这个世界一切都很难说了，晚上kuakuakua的在网上买了很多应急食材，隔天快递小哥就打电话来了 : “喂，你好，是xxx吗，下午几点在家吗，我给你送东西过来”，我这边的回复是:”你好，我是xxx，我今天不在家哦，晚上下班我在家，你晚上再过来？”，快递员：”啊？不行哦，我晚上下班了，我只有下午是送货时间，要不周末再给您送？”，我：”不行不行，今晚必须送来！”，好家伙然后双方就像下面一样僵住了，来看下面的图 ：

其实我们完全可以把上面的例子中我和山姆配送员当作是两个需要交互的系统来看，我有空的时候配送员没空，配送员有空的时候我没空，也就导致了上面僵住的局面，双方内心无数mmp，那怎么解决这个问题了？来看看下面的方案：

咋们每个小区不都有保安亭嘛l…山姆配送员在他有空的时候把东西放置在保安亭，然后他就可以去忙别的事情了，我咧等晚上下班了就去保安亭拿自己的东西，完事儿，两者皆大欢喜，如果当作系统来看，这个保安亭就是今天要介绍的消息队列MQ，那么有了这个消息队列有啥好处作用了？

消息队列的作用

1. 异步

   山姆小哥打电话给我后需要一直在你楼下等着，直到我下楼面对面拿走快递再去配送其他人的。山姆小哥将快递放在保安亭后，立马可以去配送其他人的快递，不需要一直在楼下矗着等待我下楼。提高的效率不止一点半点。

2. 解耦

   山姆小哥手上有很多人的快递都需要配送，他每次都需要先电话一一确认收货人是否有空、哪个时间段有空，然后再确定好送货的方案。这样完全依赖收货人了！

   如果快递一多，山姆小哥估计的忙疯了……如果有了保安亭，山姆小哥只需要将同一个片区的货物统一放置在对应保安亭然后一一通知收货人，然后通知收货人收到通知来取货就可以了，这时候山姆小哥和收货人就实现了完全解耦，谁都不需要依赖谁的时间是否有空。

3. 削峰

   这点的体现在每年的电商节都是能看到的，每年双11都会有大量的订单产生，然后发货却如细水长流般，这便是削峰，先通过mq接收大流量的订单，然后后续再慢慢处理每一个订单。

所以我们能看到在系统需要交互的场景中，使用消息队列中间件真的是有很多好处的，但是任何一个东西有利便有弊，比如滥用mq也会增加系统复杂度，你的系统是否可以承受这样复杂度的维护，所以我们的原则应该是在合适的场景选择合适的组建来解决我们的问题才是正解。

通信模式

正所谓无规矩不成方圆，消息队列也有自己的行规，总体就有以下两种模式：

点对点模式

如下图所示，点对点模式通常是基于拉取或者轮询的消息传送模型，这个模型的特点是发送到队列的消息被一个且只有一个消费者进行处理。生产者将消息放入消息队列后，由消费者主动的去拉取消息进行消费。它的优点是消费者拉取消息的频率可以由自己控制。但是消息队列是否有消息需要消费，在消费者端无法感知，所以在消费者端需要额外的线程去监控。

发布订阅模式

如下图所示，发布订阅模式是一个基于消息送的消息传送模型，该模型可以有多种不同的订阅者。生产者将消息放入消息队列后，队列会将消息推送给订阅过该类消息的消费者(比如关注了公众号，关注了主播你会第一时间接收到他们的发布内容)由于是消费者被动接收推送，所以无需感知消息队列是否有待消费的消息！但是consumer1、consumer2、consumer3由于机器性能不一样，所以处理消息的能力也会不一样，但消息队列却无法感知消费者消费的速度！所以推送的速度成了发布订阅模模式的一个问题！假设三个消费者处理速度分别是10M/s、5M/s、2M/s，如果队列推送的速度为6M/s，则consumer2和consumer3无法正常处理该队列的消息内容的，如果队列推送的速度为1M/s，这时三者都能正常处理但是咧consumer1、consumer2又会出现资源的极大浪费！

Kafka

上面我们对消息队列有了一个大体的认识，了解了它的一些作用和工作模式，那么接下来就轮到今天的主角登场了~

Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据，具有高性能、持久化、多副本备份、横向扩展能力（虽然它的横向扩展能力不太得劲儿）……

基础架构及术语

先来看本人整理的一张结构图，来大致了解一下kafka的整体架构，看不懂没关系，我们先来听个响 :

有点懵逼是正常的，容我先来对里面的这些概念进行一下解释，应该就会有一个清晰的理解了：

Producer：生产者，消息的产生者也可以叫投递者，这个角色就是开局例子里的山姆小哥，是消息的起源。

Broker：Broker是kafka实例，每个服务器上有一个或多个kafka的实例，我们暂且认为每个broker对应一台服务器。每个kafka集群内的broker都有一个不重复的编号，如图中的broker-0、broker-1、broker-2等

kafka cluster：kafka集群，由上面的broker组成，主要就是为了系统的高可用，消息不丢失。

Topic：消息的主题，可以理解为消息的分类，kafka的数据就保存在topic。在每个broker上都可以创建多个topic。开局的例子中我们可以理解多个临近的小区构成一个topic。

Partition：Topic的分区，每个topic可以有多个分区，分区的作用是做负载，提高kafka的吞吐量。同一个topic在不同的分区的数据是不重复的，partition的表现形式就是一个一个的文件夹！开局的例子中我们可以理解我所在的小区就是一个partition

Replication:每一个分区都有多个副本，副本的作用是做备胎。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。这个在开局的例子中并没有很好的对应的对象，这属于是kafka为了实现高可用的一种兜底机制。

Message：每一条发送的消息主体。可以理解就是我买的应急物资。

Consumer：消费者，即消息的消费方，可以理解就是开局例子里面的我，是消息的终点。

Consumer Group：我们可以将多个消费组组成一个消费者组，在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！这个也很好理解我所在的那个小区的其他Consumer，我们共同组建了一个Consumer Group！

Zookeeper：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性。它内部通过zab协议来实现，有兴趣的朋友可以自行去了解下。

通过本人这样一类比介绍，相信各位一定对上图有了一定的理解了吧，其实说的简单一点，消息队列无非就是三个核心：写消息、存消息、消费消息。那么下面就来一一介绍一下:

怎么写消息

请看下图为本人整理的写入流程，producer其实只会与Leader进行直接交互，他并不关心Follower的状态，换句话说producer只关注Leader的状态和地址，至于kafka内部是如何选处leader的它并不关心，反正他发送之前会先去集群拉取对应leader的信息

整个的发送的流程就在上图中展示的很明显了，我就不再重述一遍了，值得注意的是第四点消息写入leader后，follower是主动的去leader进行同步的！producer使用的是push模式将数据发送给broker，每条消息追加到分区中，顺序写入磁盘，注意了这一点是kafka之所以那么快的一项保证，kafka在写入的过程中采用了mmap+顺序写（减少了磁盘旋转与寻道的时间，这并不是本文的终点所以不会展开说），所以保证同一分区内的数据是有序的！写入示意图如下：

细心的小伙伴其实应该也发现问题了，不是有了topic了嘛，为啥还要搞一个partition出来了？其实分区的主要目的是：

1、 方便扩展。因为一个topic可以有多个partition，所以我们可以通过扩展机器去轻松的应对日益增长的数据量。

2、 提高并发。以partition为读写单位，可以多个消费者同时消费数据，提高了消息的处理效率。一台机器的性能是有限的，当一个机器处理不了当前消息量的时候，我们就可以水平扩展出多台机器同时进行处理。

诶，细心的朋友这时候又会提问了，既然可以扩展出多台机器，那我咋知道哪台机器应该去处理哪个partition里面的消息了？这样其实就是一个负载均衡的问题了，在kafka里面其实这里是有一些约定原则的，如下 :

kafka中有几个原则：

1、 partition在写入的时候可以指定需要写入的partition，如果有指定，则写入对应的partition。

2、 如果没有指定partition，但是设置了数据的key，则会根据key的值hash出一个partition。

3、 如果既没指定partition，又没有设置key，则会轮询选出一个partition。

kafka如何保证消息不丢失：

安全性是每个mq首当其冲应该保证的问题，如果连投递的消息安全送达到消费者手中都无法保证那这个消息队列存在也就毫无意义，就好像你买了应急物资却一直收不到货但是商场那边显示是已送达～，那kafka是怎么做到消息发送后不丢失的了？

上图中的的写入流程其实有写出来哈，其实就是通过ACK应答机制～！在生产者向队列写入数据的时候可以设置参数来确定是否确认kafka接收到数据，这个参数可设置的值为0、1、all。

0代表producer往集群发送数据不需要等到集群的返回，不确保消息发送成功。安全性最低但是效率最高。

1代表producer往集群发送数据只要leader应答就可以发送下一条，只确保leader发送成功。

all代表producer往集群发送数据需要所有的follower都完成从leader的同步才会发送下一条，确保leader发送成功和所有的副本都完成备份。安全性最高，但是效率最低。总体而言，安全和效率并不能两全其美，就像鱼与熊掌不可兼得一般，应该有所取舍。

最后要注意的是，如果往不存在的topic写数据，能不能写入成功呢？kafka会自动创建topic，分区和副本的数量根据默认配置都是1。

怎么存消息

好，上面我们讲述了消息写入的全过程，那写入后的消息是如何存放的了，山姆小哥将我购买的应急物资难道就是随手一扔？答案肯定是否哈，生产者将数据写入broker后，此时cluster就需要对数据进行存储了，毫无疑问数据是被存在硬盘里的，只有这样才能保证数据在broker发生异常时不丢失数据，可能在我们的一般的认知里，写入磁盘是比较耗时的操作，不适合这种高并发的组件。但是上面写消息的时候已经讲过了 MMAP + 顺序写 是kafka高性能的保证之一，在消费消息的时候其实还有一个sendfile的零拷贝方案也是其高性能的保证之一，这个我们后面有机会再说。Partition 结构前面说过了每个topic都可以分为一个或多个partition，如果觉得topic比较抽象，那partition就是实实在在的东西！Partition在服务器上的表现形式就是一个一个的文件夹，每个partition的文件夹下面会有多组segment文件，每组segment文件又包含.index文件、.log文件、.timeindex文件（早期版本中没有）三个文件， log文件就实际是存储message的地方，而index和timeindex文件为索引文件，用于检索消息。

来看上图，这个partition 0有三组segment文件，每个log文件的大小是一样的，但是存储的message数量是不一定相等的（每条的message大小不一致）。文件的命名是以该segment最小offset来命名的，如000.index存储offset为0~555555的消息，kafka就是利用分段+索引的方式来解决查找效率的问题。

Message结构上面说到log文件就实际是存储message的地方，我们在producer往kafka写入的也是一条一条的message，那存储在log中的message是什么样子的呢？消息主要包含消息体、消息大小、offset、压缩类型……等等！

名词解释

1、 offset：又叫偏移量是一个占8byte的有序id号，它可以唯一确定每条消息在parition内的位置！通过起始位置+偏移量可以迅速获取到对应消息的起始位置。bingo～

2、 消息大小：消息大小占用xxbyte，用于描述消息内容实际占用的空间大小。

3、 消息体：消息体存放的是实际的消息数据（被压缩过，后面有时间我也会出一篇关于压缩算法的文章十分有趣哈），占用的空间根据具体的消息而不一样。

存储策略

写到这里，醒目的小伙伴应该会有一个疑问了，生产者发了那么多消息到broker集群，先不管有没有consumer对其进行消费处理，那这些消息会一直存在吗？如果一直存在broker总有撑爆的一天吧？那kafka也是有对应的策略去处理过期消息的，kafka会在以下两种方案中择其一来对数据进行删除：

1、 基于时间，默认配置是168小时（7天）。也就是说kafka的内部线程会对存在超过168小时的数据进行删除。

2、 基于大小，默认配置是1073741824。需要注意的是，kafka读取特定消息的时间复杂度是O(1)，所以这里删除过期的文件并不会提高kafka的性能！

如何消费消息

上面我们讲解了消息是如何发送和存储的，那么接下来就是最后的消费重头戏了，消息存储在log文件后，consumer们就可以对其进行消费了。kafka的consumer会主动的去kafka集群拉取消息，消费者在拉取消息的时候也是只会关注Leader的，这点与producer完全一致，consumer通常也不是孤军奋战的，它们通常都是结伴而行，多个consumer可以组成一个消费者组（consumer group），每个消费者组都有一个组id(唯一id)！同一个消费组者的消费者可以消费同一topic下不同分区的数据，但是永远不会出现一个组内多个消费者对同一分区的数据进行消费～，来看我整理的下图进行加深理解 ：

图中一个的topic有4个partition，但是对应的消费者组中只有3个消费者，这时就会出现上图中出现的情况，最后的consumer2承担了两个partition的消费工作，那如果是消费者组的消费者多于partition的数量，那会不会出现多个消费者消费同一个partition的数据呢？答案是不会的，多出来的消费者不消费任何partition的数据，它就像一个废人一样存在的毫无意义，所以在实际的应用中，应该将消费者组的consumer的数量与partition的数量保持一致！在保存数据的小节里面，我们聊到了partition划分为多组segment，每个segment又包含.log、.index、.timeindex文件，存放的每条message包含offset、消息大小、消息体。

其实我们已经不止一次的提到segment和offset，那么我们最后就来看看kafka是如何通过这两个参数快速定位到对应消息的。

消息查找策略

1、 先找到offset为555560的message所在的segment文件（利用二分法查找），这里找到的就是segment 2文件夹。

2、 打开找到的segment中的.index文件（也就是555555.index文件，该文件起始偏移量为55555+1，我们要查找的offset为555560的message在该index内的偏移量为55555+5=555560，所以这里要查找的相对offset为5）。由于该文件采用的是稀疏索引的方式存储着相对offset及对应message物理偏移量的关系，所以直接找相对offset为5的索引找不到，这里同样利用二分法查找相对offset小于或者等于指定的相对offset的索引条目中最大的那个相对offset，所以找到的是相对offset为4的这个索引。

3、 根据找到的相对offset为4的索引确定message存储的物理偏移位置为256。打开数据文件，从位置为256的那个地方开始顺序扫描直到找到offset为555560的那条Message。

这整套机制是建立在offset为有序的基础上，如果offset无序那这套机制也就毫无卵用了，利用segment+有序offset+稀疏索引+二分查找+顺序查找等多种手段来高效的查找数据！

Ok，通过以上三步consumer就能准确的拿到需要处理的message进行处理了。接下来就只需要记录下这个位置然后下次继续从此处往后进行消费就可以了，那最后的问题又来了到底该如何存放这个offset位置的值了？直接公布答案，其实在早期的版本中，消费者将消费到的offset维护zookeeper中，consumer每间隔一段时间上报一次，这里容易导致重复消费，且性能不好！在新的版本中消费者消费到的offset已经直接维护在kafka集群的_consumeroffsets这个topic中！

总结

噗，终于讲完了～，希望本文能让你对kafka有一个全面且深刻的了解，其实kafka里面还有很多知识这里篇幅有限不能一一讲解，譬如rebalance又譬如零拷贝等等，有兴趣的可以自行去学习了解一下哈，毕竟本人是为了分享pulsar才拿kafka做铺垫的哈，后面pulsar见，完结撒花～。

前言

上一篇我们了解了golang的string是如何玩转的，趁热打铁今天就来学习常用的slice结构，这个slice跟我们以前做java的时候用到的动态数组(List)很相似，但是又有区别，那么今天就来探究下它的原理。

slice

三要素

slice包含以下图示的三个核心元素，数据、已存放数据长度以及容量 :

举个例子如下 :

我们声明了一个变量intSlice，实际上这只是一个声明，底层并没有分配对应的数组进行支持，所以指向底层数组的指针data就是nil，而长度和容量都是0值。

在golang中有两种初始化slice的方式分别是make 和 new

make

从下图中我们可以看到通过make的方式声明变量intSlice就会在底层分配一个对应的数据类型的数组，并将data指向底层数组的起始地址，操作1就是向slice中添加一个元素，操作2就是改变slice已经存在元素的值，操作3就是访问超过len的数据这种情况会直接panic。

new

接下来就是通过new关键字声明变量，可以知道并未在底层初始化对应的数组，所以操作1直接给slice下标为0的元素赋值会直接panic，这时只有通过操作2 append操作才会在底层初始化对应的数组并将data指向底层数组的起始位置，其实这里有一点绕，slice的data指向底层的一个字符串数组，上一篇中我们讲过string的结构不记得的可以回去再看一遍，所以字符串的data还会指向另外一个底层实际存储字符编码的数组。

通过以上对比我们发现其实两者还是有很大区别的，两者虽然都可以用来声明slice变量，但是前者会在声明变量的同时在底层分配好对应类型的数组结构而后者则不会，实际上我本人在实际的开发过程中也只会用make。make不仅可以初始化slice，还可以用在map和chan的初始化上面。

底层数组

上面一直在说底层数组，那么底层数组究竟是什么？数组其实就是一段连续的内存空间，在这一段连续的内存空间内一个挨着一个的存储着同种类型的数据，int型的slice底层其实就是一个int数组，string型的slice底层就是一个string数组，从上面说明的例子我们可以看到slice都是指向了底层数组的起始地址，但这是必须的吗？我们来看接下来的一个列子 :

我们先来声明一个int类型的数组arr，容量为10，数组的容量一旦声明就不能再变了，我们可以通过下面s1和s2的声明方式（左闭右开原则）将slice变量关联到同一个arr数组，可以看到s1的指向地址其实是底层数组第二个元素的地址位置，而s2的指向地址其实是底层数组第八个元素的地址位置，这其实不难理解，那么本文的核心来了，请仔细看操作 s2 = append(s2,11)（这个动作并不是线程安全的后面我们会细聊），该操作会触发扩容动作？那么什么是扩容了？slice又是怎么扩容的了？

扩容

上面提出了一个扩容的概念，其实这个也很好理解s2指向的底层数组已经满了无法将11存放，这时s2就需要另谋高去开辟另外的一片内存空间将自己原来的数据复制过去后再将11添加的末尾，最后s2就指向了该新内存空间的起始地址。在这个过程中我们其实会有几个疑问的，第一个我怎么知道那片新的内存空间要多大了，随便还是无限的？我能想到golang当然也能想到，扩容整体来说就是下面的三个步骤。

步骤1 : 预估扩容后的容量

这个规则其实没啥好说的，源码总结规则如下：

步骤2 : 扩容后需要多大内存

这个其实跟我们的元素类型是息息相关的，但是真的就是按照下图的方式直接分配至么多内存吗？答案是否定的。

步骤3 : 匹配合适的内存规格

上面这个问题简单来说就是编程语言(c除外)去申请内存并不是直接向操作系统申请的，中间还夹着一层代理人(内存管理模块)，系统启动时代理人先向os提前申请好一批内存，分成常用的规格管理起来，当golang向其申请内存空间时，代理人就会挑选对应刚大于申请空间大小的内存块并分配给申请人，这样就完成了整个的扩容动作。

总结

今天我们聊了聊slice的相关原理，是不是很有趣，现在只是开篇稍微简单一点，后面会逐步深入，难度也会逐步增大，下一篇我们聊聊内存对齐哈，slice完结撒花～。

前言

从java转golang已经一年多了，过去的一年确实也挺坎坷的，国家打压部门解散，刚在支付组有一点感觉然后就没然后了，也没事后面活水到新部门继续肝，这不刚用go重构完一个十年前的c++系统，过程之痛苦一言难尽，但是痛苦归痛苦从中还是学到了不少，例如redis，pulsar这个之前已经写过好几篇基本的文章了，后续会更新一些自己在用法上的硬核知识，啊哦跑题了，本系列的核心是golang，用了一年多的golang是时候该总结一下了，俗话说的好学而时习之才能逆水行舟， 今天的主题是string～

string

比特与字节

一个bit或者是0或者是1，8个bit组成一个字节，全部为0代表0，全部为1代表数字255，相信这个不必多说大家都知道了，一个字节可以表示256个数字，两个字节就是65536个数字了，更多的字节就可以表示更大的数字，如下图所示 :

字符集

以上我们说的都是整数，那字符是怎样存储的了？一堆二进制是怎么转换成字符的了？不能直接展示那就通过数字中转一下，如下图 : 存储保持不变，多添加一层映射关系就能解决这个问题啦，尽可能多的将世界上出现的字符收录进来然后一一进行编号构建一张映射表，这个映射表就叫做字符集，下面展示了字符集的进化旅程，其实就是一个字符映射表不断完善的过程，直到最后被unicode统一规范。

字符串存储（UTF-8）

上面说明字符集，但是真的只是有了字符集就万事大吉了吗？来思考下该怎么存储 “世界aABb” ？

最直接的想法是不是根据字符集找到每一个字符的编号存成二进制完事儿？来看看是不是就像下面这样，乍一看没有问题，但是仔细一看不对啊有的占一个字节有的占了三个字节，那我咋知道那一长串玩意儿要怎么划分了？这个方案显然不行…

于是我们应该能想到第一种方案 : 定长编码

这个方案，乍一看没有问题，然后再仔细一看确实也没有啥问题，就是浪费的字节稍微有点多……那可以怎么解决了？这里就引出了另外一种方案 : 变长编码，如下图所示：

这个图看着是不是有点懵逼，没关系，容我来稍微解释一下，每个字节分为标示位和实际数据两部分，例如第一行的0??? ????,如果数据在0～127之间，对应的编码模版就是以0标示开头后7位表示实际数据，如果数据在128～2047之间数据就占两字节，每个字节分别以固定的110和10两个固定标示开头，后面的也一样，其实就是通过固定的标示位来重新组装原来的二进制数据从而降低对内存的浪费。看到这里现在应该知道最开始的那个字符串该怎么存储了吧？

这其实就是我们熟知的UTF-8编码，也就是golang默认的编码方式，字符集和编码方式是需要相互配合才能达到最优方案的。接下来我们就可以来了解下golang的string是如何实现的了。

go的string

碰巧了前段时间重构c++，也了解了一下c++的string是如何实现的，来我们先来看看C语言是如何实现的，如下图 :

C语言如是说，会在结尾处通过 \0 的特殊字符来标示结尾，那这也就意味着存放的字符串中是无法出现 \0 这样的字符的，-_-!!!本人很坦诚的说一句这方案忒差了，所以golang并没有采用这种方案，来看看golang是如何处理的:

golang通过添加一个len变量来存储字符串中字节的长度，还是上面的例子”世界aABb”就应该是这样 :

好了关于golang的string今天就介绍到这里，对了，最后非常值得注意的一点是，无论是java或是golang，string都会被认为是不可变变量，是不允许修改的，golang编译器会把 s1:= “世界aABb”这样的变量分配到只读内存中，多个str是可以共享底层同一个数组变量的，可以重新复制这时会重新开辟一个数组空间，但是绝对不允许修改，完结撒花。

总结

今天属于是golang系列的开篇，后续的该系列也会不断完善，下一篇我们讲slice。

前言

之前做java开发，服务间相互调用用的最多的就是java原生序列化和json序列化协议，然而后面去了大厂做golang开发基本都是采用pb了，刚开始以为只是规范使然，然而通过后面的学习发现pb确实是全方位领先于其他各种序列化协议的，简单来说就是pb不仅更快而且更小，下面就来细细的探讨一番。

bp为什么这么屌

官方测试

那么pb究竟实战有多屌了，先看两幅官方测试报告图：

​ 解包耗时

​ 数据包压缩后大小

可以很明显的看到，一条消息数据，用protobuf序列化后的大小是json的10分之一，是xml格式的20分之一，但是性能却是它们的5~100倍。

为什么用pb对数据包压缩后更小

下面以json数据为基础出发，通过一步一步的对它进行优化，来理解protobuf的实现原理。

例如有一条信息，用json的表示方式如下：

1

{ "age": 32, "name": "xxx",  "height": 170, "weight": 106 }

很明显，整个json串种包含着很多可有可无的字符，例如”,” “:”等，为了把数据变的更小一点，我可能就会采取如下处理方式：

这里直接舍去了全部不必要的冗余字符，这里其实已经对数据进行了大幅缩减了，但是这时会出现一些新的问题，接收端接收到数据后咋知道32对应的是那个字段，xxx又是对应的哪个字段，也就是字段的对应问题这时是无法解决的。

那我们可以对字段都编个号，接收端接收到数据后就按照这个编号进行解析即可，如下：

这样看来就完美达成目的了

新的问题以及解决方案

虽然上述方案可以达到解决问题的目的，但是我们来假设一下下面的情况height这个字段为null，也就是没有值，那么传递的数据就会变成如下：

但是在接收端，解析数据并按照顺序进行字段匹配的时候就会出问题：

很明显数据已经乱套了，原本weight的值解析到了height字段，那为了解决这个问题，pb引入了一个名为tag的技术：

也就是说，每个字段我们都用tag|value的方式来存储的，在tag当中记录两种信息，一个是value对应的字段的编号，另一个是value的数据类型（比如是整形还是字符串等），因为tag中有字段编号信息，所以即使没有传递height字段的value值，根据编号也能正确的配对。细心的朋友可能已经发现了，乍一看这个方案跟json的key/value方案无异啊，绕了一圈又回到了原点？哈哈，莫急，让我慢慢道来。

Tag的开销

接着上面的问题我们继续……

这个问题其实问的相当好，json中的key其实是字符串，我们知道每个字符会占据一个字节，所以像name这个key就会占据4个字节，但在protobuf中，tag使用二进制进行存储，一般只会占据一个字节，它的核心代码为：

1
2
3

static int makeTag(final int fieldNumber, final int wireType) {
  return (fieldNumber << 3) | wireType;
}

fieldNumber表示后面的value所对应的字段的编号是多少，比如fieldNumber为1，就表示age，如果为2，就表示name等；wireType表示value的数据类型，以此来计算value占用字节的大小。在protobuf当中，wireType可以支持的字段类型如下：

因为tag一般占用一个字节，开销还算是比较小的，所以protobuf整体的存储空间占用还是相对小了很多的。

看完上面的说辞是否还是不太理解，下面我来举个例子就很清晰了，例如上面的0-5种数据类型分别对应二进制位 000～101，排序第一的字段age生成的tag就是00001000，一个字节前5位表示序号后三位表示类型。

此时出现一个新问题，那么Tag分隔符为一个字节，如果传输的内容中出现相同的字节，会导致解析错误吗？这里就需要了解一下什么是Varint编码。

Varint编码

这里直接通过实例来进行说明更为直观，如图：

图中对数字123456进行varint编码，123456用二进制表示为 11110001001000000，每次从低向高取7位再加上最高有效位变成 11000000 11000100 00000111 所以经过varint编码后123456占用三个字节分别为 1921967,同样解码的时候就逆向操作即可，通过这样的方式我们就省掉一个字节的开销，其实通常在实际项目我们传输的int一般来说都是比较小的，所以这样的设计也是非常ok的。

Zigzag编码

上面的设计看似完美，但是其实我们仔细思考一下，如果是-1这种负数改如何是好？

-1 –> 11111111 11111111 11111111 11111111

如果继续采用上面的方式并不是一个好的选择，这里就不得不说zigzag编码方式了

直接看表格吧

能看到编码方式就是 ：

负数 2 *|x| - 1 正数 2 * |x| 是不是很简单又很神奇。这里提一句即使使用该编码方式后续也还是会使用varint再进行编码的。

Tag-Length-Value(TLV)

我们之前讲的varint又或者是zigtag都是以传输数字为基础的，那如果我们传输的是字符串了？那么引出了另一主角TLV，Tag为分隔符，Length为长度但是我们同样采用varint编码的方式。

接下来我们回到最开始的问题，会出现解析错误的问题吗？

我们来尝试推导一下解析过程 : 如果一开始是要传输一个数字，我们拿到了第一个Tag，解析出它的fieldNumber和wireType，因为采用的是varint的编码方式（zigzag后也是采用varint再次进行编码的），高位为1表示下一个字节还是数字，如果为0则表示下一个字节就是Tag了。如果一开始传输的是一个字符串，那么拿到Tag后就知道接下来的是一个字符串，那么下一个字节就开始解析Length，Length同样还是使用varint编码，遇到高位为0后表示该Length解析完毕，我们就能拿到value的长度了，接下来按照长度取完字符串后，下一个字节就是Tag了。以此类推，pb永远都清楚的知道哪一个字节是Tag，所以现在还疑惑吗？

问题回顾

纠正上面的一个问题，Tag其实并不是固定只占一个字节，而是采用varint的方式呈动态的！试想如果真的固定只占一个字节，那最高也就5位来表示字段序号，这显然是不行的！

以下是来自Google Protobuf Documents里的一句话：

As you can see, each field in the message definition has a unique numbered tag. These tags are used to identify your fields in the message binary format, and should not be changed once your message type is in use. Note that tags with values in the range 1 through 15 take one byte to encode. Tags in the range 16 through 2047 take two bytes. So you should reserve the tags 1 through 15 for very frequently occurring message elements. Remember to leave some room for frequently occurring elements that might be added in the future.

至于这里为什么是1到15 AND 16到2047呢？

1. 1到15，仅使用1byte。每个byte包含两个部分，前5位是field_number后三位是tag，其中field-number就是protobuf中每个值后等号后的数字也叫字段序号。那么一个byte用来表达这个值就是00000000，其中首位的0表示是否有后续字节，如果为0表示没有也就是这是一个字节，第2到第5位表示field-number，后三位部分则是wire_type部分，表示数据类型。也就是(field_number << 3) | wire_type。其中wire_type只有3位，表示数据类型。那么能够表示field_number的就是第2到第5位的数字，4位数字能够表达的最大范围就是1-15（其中0是无效的）。
2. 至于16到2047，则与上面的规则其实类似，需要用varint规则进行考虑，下面以2bytes为例子，那么就有10000000 00000000，其中首位的1依旧是符号位，因为varint规则中每个byte的第一位都用来表示下一byte是否和自己有关，那么对于>1byte的数据，第一位一定是1，因为这里假设是2byte，那么第二个byte的第一位是0，刨除这两位，再扣掉3个wire_type位，剩下11位，能够表达的最大数字就是2047，一般我们日常中接口传参数量超过这个数的微乎其微。

总结

Protobuf确实是目前最好的数据传输协议没有之一，当然我们不仅是要会用也要知道为啥要用～

前言

前面对pulsar整体的执行原理进行了一番介绍，下面来对使用实践相关，首先要介绍的就是订阅模式，订阅模式决定了消费者如何对一个topic/partition中的数据进行消费，了解其相关的知识对使用来说是至关重要的。

订阅模式

为了适用不同场景的需求，Pulsar 支持四种订阅模式：分别是Exclusive(独占)、Shared(共享)、Failover(灾备)、Key_Shared(key共享)。下面分别对这四种模式进行一个介绍

独占模式（Exclusive）

Exclusive 独占模式（默认模式）：一个 Subscription 只能与一个 Consumer 关联，没错，是一个！！！只有这个 Consumer 可以接收到 Topic 的全部消息，如果该 Consumer 出现故障了就会停止消费。

Exclusive 订阅模式下，同一个 Subscription 里只有一个 Consumer 能消费 Topic，如果多个 Consumer 订阅则会报错，适用于全局有序消费的场景。

当启动多个消费者时，就会报错。

共享模式（Shared）

消息默认通过轮询机制（也可以自定义）分发给不同的消费者，并且每个消息仅会被分发给一个消费者。当消费者断开连接，所有被发送给他，但没有被确认的消息将被重新安排，分发给其它存活的消费者。

这也是我们使用最频繁的一种消费模式，目前我重构的vdc基本都是使用这种模式，可以在管理端看到如下多个消费者消费同一个topic

灾备模式（Failover）

当存在多个 consumer 时，将会按字典顺序排序，第一个 consumer 被初始化为唯一接受消息的消费者。当第一个 consumer 断开时，所有的消息（未被确认和后续进入的）将会被分发给队列中的下一个 consumer。

这种模式跟第一种独占模式很相似，优势就是有备胎，主消费者挂了，备胎就会马上顶替上来。

KEY 共享模式（Key_Shared）

当存在多个 consumer 时，将根据消息的 key 进行分发，key 相同的消息只会被分发到同一个消费者。这种消费模式也是使用比较频繁，当我们需要对消息进行分类消费时我们就可以使用这种模式，举个例子我们的vdc系统需要消费多款不同的引擎的扫描结果，我们需要把引擎分类做不同的逻辑处理，这时就可以使用到这种消费模式了。

总结

以上我们总结了pulsar的四种消费模式，了解其相关的原理对于我们使用上是至关重要哈。

前言

上一篇文章对Pulsar的消息存储原理和ID规则进行了介绍，本篇文章来对消息副本以及存储机制来介绍一下。lets go

消息元数据组成

Pulsar 中每个分区 Topic 的消息数据以 ledger 的形式存储在 BookKeeper 集群的 bookie 存储节点上，每个 ledger 包含一组 entry，而 bookie 只会按照 entry 维度进行写入、查找、获取。

说明：

批量生产消息的情况下，一个 entry 中可能包含多条消息，所以 entry 和消息并不一定是一一对应的。

Ledger 和 entry 分别对应不同的元数据。

• ledger 的元数据存储在 zk 上。
• entry 除了消息数据部分之外，还包含元数据，entry 的数据存储在 bookie 存储节点上。

每个 ledger 在创建的时候，会在现有的 BookKeeper 集群中的可写状态的 bookie 候选节点列表中，选用 ensemble size 对应个数的 bookie 节点，如果没有足够的候选节点则会抛出 BKNotEnoughBookiesException 异常。选出候选节点后，将这些信息组成 <entry id, ensembles> 元组，存储到 ledger 的元数据里的 ensembles 中。

消息副本机制

消息写入流程

客户端在写入消息时，每个 entry 会向 ledger 当前使用的 ensemble 列表中的 Qw 个 bookie 节点发送写入请求，当收到 Qa 个写确认后，即认为当前消息写入存储成功。同时会通过 LAP（lastAddPushed）和 LAC（LastAddConfirmed）分别标识当前推送的位置和已经收到存储确认的位置。

每个正在推送的 entry 中的 LAC 元数据值，为当前时刻创建发送 entry 请求时，已经收到最新的确认位置值。LAC 所在位置及之前的消息对读客户端是可见的。

同时，pulsar 通过 fencing 机制，来避免同时有多个客户端对同一个 ledger 进行写操作。这里主要适用于一个 topic/partition 的归属关系从一个 broker 变迁到另一个 broker 的场景。

消息副本分布

每个 entry 写入时，会根据当前消息的 entry id 和当前使用的 ensembles 列表的开始 entry id（即key值），计算出在当前 entry 需要使用 ensemble 列表中由哪组 Qw 个 bookie 节点进行写入。之后，broker 会向这些 bookie 节点发送写请求，当收到 Qa 个写确认后，即认为当前消息写入存储成功。这时至少能够保证 Qa 个消息的副本个数。

如上图所示，ledger 选用了4个 bookie 节点（bookie1-4 这4个节点），每次写入3个节点，当收到2个写入确认即代表消息存储成功。当前 ledger 选中的 ensemble 从 entry 1开始，使用 bookie1、bookie2、bookie3 进行写入，写入 entry 2的时候选用 bookie2、bookie3、bookie4写入，而 entry 3 则会根据计算结果，写入 bookie3、bookie4、bookie1。

消息恢复机制

Pulsar 的 BookKeeper 集群中的每个 bookie 在启动的时候，默认自动开启 recovery 的服务，这个服务会进行如下几个事情：

1. auditorElector 审计选举。
2. replicationWorker 复制任务。
3. deathWatcher 宕机监控。

BookKeeper 集群中的每个 bookie 节点，会通过 zookeeper 的临时节点机制进行选主，主 bookie 主要处理如下几个事情：

1. 负责监控 bookie 节点的变化。
2. 到 zk 上面标记出宕机的 bookie 上面的 ledger 为 Underreplicated 状态。
3. 检查所有的 ledger 的副本数（默认一周一个周期）。
4. Entry 副本数检查（默认未开启）。

其中 ledger 中的数据是按照 Fragment 维度进行恢复的（每个 Fragment 对应 ledger 下的一组 ensemble 列表，如果一个 ledger 下有多个 ensemble 列表，则需要处理多个 Fragment）。

在进行恢复时，首先要判断出当前的 ledger 中的哪几个 Fragment 中的哪些存储节点需要用新的候选节点进行替换和恢复数据。当 Fragment 中关联的部分 bookie 节点上面没有对应的 entry 数据（默认是按照首、尾 entry 是否存在判断），则这个 bookie 节点需要被替换，当前的这个 Fragment 需要进行数据恢复。

Fragment 的数据用新的 bookie 节点进行数据恢复完毕后，更新 ledger 的元数据中当前 Fragment 对应的 ensemble 列表的原数据。

经过此过程，因 bookie 节点宕机引起的数据副本数减少的场景，数据的副本数会逐步的恢复成 Qw（后台指定的副本数，TDMQ 默认3副本）个。

总结