论文笔记：Scaling Memcache at Facebook

论文介绍了Facebook如何使用memcache，以及相关魔改和维护memcache集群。

背景

facebook使用一致性哈希来构建memcahce集群。

We provision hundreds of memcached servers in a cluster to reduce load on databases and other services.
Items are distributed across the memcached servers through consistent hashing [22]. 
Thus web servers have to routinely communicate with many memcached servers to satisfy a user request

使用

cache aside模式

facebook根据他们的场景，也就是读多写少使用了cache aside模式。原文叫demand-filled look-aside cache。具体使用方式如下图。

facebook 使用delete而不是update缓存的原因是因为delete是幂等的，而update不是(we choose to delete cached data instead of updating it because deletes are idempotent).如果直接更新可能出现先更新的后放入缓存。

在facebook也是用memcache作为大数据平台计算的中间结果存储平台，用来存放一些中间结果

魔改

魔改背景

一个web请求会请求平均521个独立的缓存项.(A single user web request can often result in hundreds of individual memcache get requests. For example, loading one of our popular pages results in an average of 521 distinct items fetched from memcache)

魔改目的

魔改的目的是为了降低延迟，降低负载。

魔改方法

延迟与负载：1.网络延迟 2.由于cache miss带来的延迟和后端的负载。解决方案如下：

1. 代码批量化访问缓存，减少来回的网络开销
2. 对于网络延迟的降低，facebook对于GET使用UDP传输，SET和DELETE使用TCP传输。因为UDP是无连接的，可以降低TCP维护链接的开销。使用UDP客户端自己记录请求序号，丢包和乱序被视为客户端错误，在实际生产环境中，有0.25%的get请求被扔掉。其中的80%是因为回包延迟，其他的是因为乱序。对于上面的错误，客户端直接认为是cache miss。直接取后端，但是不会放入缓存是为了避免额外的网络开销
   效果如下图:
   
3. 使用类似TCP拥塞控制的方法来防止缓存集群负载过高。客户端维护一个滑动窗口，当服务端正常响应，就加大窗口，当服务端没响应或者超时就减小窗口。

后端负载降低

1. 引入“租约Leases”，租约用来解决两个方面的问题 1. 过期的设置 2.惊群效应（stale sets and thundering herds）。

• 过期的设置主要来源于放入缓存的顺序错乱（后来先到）。
• 惊群效应发生在某个key被频繁的读写。由于写是invalid cache，所以大量读请求就会无法命中缓存，导致进入了后端的数据库。
• 租约就是一个对应于请求的key的64bit的token。（The lease is a 64-bit token bound to the specific key the client originally requested）。当set cache的时候，客户端需要把这个key带上一起给到缓存服务器，当服务端发现token不匹配的时候，就说明这是一个过期的写请求。
• 租约同时可以降低惊群效应。缓存服务器控制发token的频率，默认是10秒一次，缓存服务器发过一次token后，后续再来的请求就直接等一会（等前面那个拿到token的返回之后，缓存就有数据了）
  Without leases, all of the cache misses resulted in a peak database query rate of 17K/s. With leases, the peak database query rate was 1.3K/s

2. 使用过期值

• 通过返回slightly out-of-date data，当一个key被删除的时候，在真正flush之前，会被放入一个特殊的数据结构中。get请求可以返回一个租约或者这个被标记为过期的值。

3. 处理失败，这里的失败有两种

• 有一小部分机器由于网络原因无法访问（facebook依赖一个自动化的组件去发现这些机器，但需要时间 few minutes，在这一段时间内可能会带来一些连锁失败）
  解决办法：有一个叫gutter的小池子，大概是cache集群的1%大小，用来替代那些无法访问的机器。当client无法收到缓存服务器的响应，他们就去访问gutter池子里面的机器。注意这根一般一致性哈希的玩法不一样，原因是有可能有些key非常的热，如果把它hash到下一台机器，就会增加那台机器的负载，导致又爆炸。把它导到空闲的gutter机器上就可以避免这个问题。一旦出现一台缓存机器无法访问，gutter通常能在4分钟内，缓存命中率能超过35%，有时能达到50%
• 集群中的大部分机器都无法访问
  解决办法：切换流量到其他可用region

维护memcache集群

待续

基数估计算法简介

注：本文是之前工作时在团队内分享的一个PPT的文字版本.

下文中的sqrt表示开根号(sqrt(4)=2)，m^n表示m的n次方

什么是基数（Cardinality）

基数指的是一个可重复集合中不重复元素的个数。

什么是基数计算

给定一个含有重复元素的有限集合，计算其不重复元素的个数。

应用场景举例:

• 某家店铺今天有多少不同用户访问
• 某家店铺今天接待了多少不同买家

简单来说就是各种UV的计算

常见的实现方式

Hash集合 + 计数 或者 BitMap + 计数

缺陷：占用空间过大

什么是基数估计算法

基数估计是一类概率算法，可以在有一定误差的前提下以远低于精确计算的时间和空间消耗对基数进行估计

特点

• 有误差
• 时间复杂度和空间复杂度仅与误差和基数上限有关
• 易于合并

有哪些基数估计算法

• Linear Counting
• LogLog Counting
• Adaptive Counting
• HyperLogLog Counting
• HyperLogLog++ Counting

它们之间的关系如下图所示：

Linear Counting

Linear Counting是在1990年的一篇论文
A linear-time probabilistic counting algorithm for database applications 中被提出。

作为一个早期的基数估计算法，Linear Counting的空间复杂度方面与简单bitmap方法是一样的（但是有个常数项级别的降低，约1/10），都是O(N)。 因此很少单独使用。目前只在Adaptive Counting中被混合使用

特点：在基数较小时比较准确

具体算法如下:

设有一哈希函数H，其哈希结果空间有m个值，并且哈希结果服从均匀分布。

使用一个长度为m的bitmap，每个bit为一个桶，均初始化为0

设一个集合的基数为n，此集合所有元素通过H哈希到bitmap中，如果某一个元素被哈希到第k个比特并且第k个比特为0，则将其置为1。

当集合所有元素哈希完成后，设bitmap中还有u个bit为0

估计值 n = -mLn(u/m)

误差 = (e^t-t-1)/2n 其中 t = n/m 也就是基数与哈希空间的比值

标准差 = (sqrt(m)*(e^t-t-1)^0.5)/n

误差控制，以标准差作为误差，记容许误差为E，则 m > (e^t -t -1)/(E*t)^2

例子:

n=11(实际数目) m=8(哈希空间) u=2(0的个数)

估计值=-8*ln(0.25)=11.04

满桶控制:

如果m比n小太多，则很有可能所有桶都被哈希到了，此时u的值为0，估计公式就不起作用了（变成无穷大）。

因此m的选择除了要满足上面误差控制的需求外，还要保证满桶的概率非常小。

1
2

m的选取应该遵从:
m > beta(e^t-t-1) 其中 beta=max(5,1/((E*t)^2))

LogLog Counting

LogLog Counting 出自论文Loglog Counting of Large Cardinalities

Loglog Counting的空间复杂度仅有O(log2(log2(Nmax)))，所以叫LogLog Counting

算法前提

均匀随机化:与LC一样，在使用LLC之前需要选取一个哈希函数H应用于所有元素，然后对哈希值进行基数估计。H必须满足如下条件

• H的结果具有很好的均匀性，无论原始集合元素的值分布如何，其哈希结果的值几乎服从均匀分布

• H的碰撞几乎可以忽略不计。认为对于不同的原始值，其哈希结果相同的概率非常小以至于可以忽略不计。

• H的哈希结果是固定长度的

算法基本思路

设a为待估集合（哈希后）中的一个元素，由上面对H的定义可知，a可以看做一个长度固定的比特串（也就是a的二进制表示）.

设H哈希后的结果长度为L比特，将这L个比特位从左到右分别编号为1、2、…、L：

又因为a是从服从均与分布的样本空间中随机抽取的一个样本，因此a每个比特位服从如下分布且相互独立。

P(x=k)=0.5

就是a的每个比特位为0和1的概率各为0.5，且相互之间是独立的

设ρ(a)为a的比特串中第一个“1”出现的位置，显然1≤ρ(a)≤L，这里我们忽略比特串全为0的情况（概率为1/2^L）。如果我们遍历集合中所有元素的比特串，取ρmax 为所有ρ(a) 的最大值。

此时我们可以将2^ρmax 作为基数的一个粗糙估计，即

1

n 约为 2^ρmax

减小误差

• 分桶平均
• 偏差修正

分桶平均

将哈希空间平均分成m份，每份称之为一个桶

对于每一个元素，其哈希值的前k比特作为桶编号， 2^k = m

后L-k个比特作为真正用于基数估计的比特串

桶编号相同的元素被分配到同一个桶，在进行基数估计时，首先计算每个桶内元素最大的第一个“1”的位置，设为M[j]，然后对这m个值取平均后再进行估计

误差修正:

误差分析
当m不太小（不小于64）时:

StdError = 1.30/sqrt(m)

误差控制:
在应用LLC时，主要需要考虑的是分桶数m，而这个m主要取决于误差。根据上面的误差分析，如果要将误差控制在E之内，则：

1

m > (1.30/E)^2

内存使用分析:

假设H的值为32bit，由于ρmax≤32，因此每个桶需要5bit空间存储这个桶的ρmax，m个桶就是5m/8字节。

例如基数上限为一亿（约2^27），当分桶数m为1024时，每个桶的基数上限约为2^27/2^10=2^17

因为每个桶需要5bit，需要字节数就是5×1024/8=640，误差为1.30/sqrt(1024)=0.040625，也就是约为4%。

LLC标准差是渐近组合计数，也就是说，随着n趋向于无穷大，标准差趋向于1.30/sqrt(m)，而不是说n多大时其值都一致为1.30/sqrt(m).

其无偏性也是渐近的，只有当n远远大于m时，其估计值才近似无偏。因此当n不太大时，LLC的效果并不好。

通过统计分析方法，我们可以得到n具体小到什么程度我们就不可忍受了，另外就是当n太小时可不可以用别的估计方法替代LLC来弥补LLC这个缺陷。Adaptive Counting 及HyperLogLog Counting都是基于这个思想实现的。

Adaptive Counting

Adaptive Counting在Fast and accurate traffic matrix measurement using adaptive cardinality counting中被提出。

Adaptive Counting只是简单将LC和LLC组合使用，根据基数量级决定是使用LC还是LLC。具体是通过分析两者的标准差，给出一个阈值，根据阈值选择使用哪种估计。

特点：在基数较大时比较准确

分析一下LC和LLC的存储结构，可以发现两者是兼容的，区别仅仅在于LLC关心每个桶的ρmax，而LC仅关心此桶是否为空。因此只要简单认为ρmax值不为0的桶为非空，0为空就可以使用LLC的数据结构做LC估计了。

联立linear counting和loglog Counting的误差

HyperLogLog Counting

HyperLogLog Counting的基本思想也是在LLC的基础上做改进， 来自论文 HyperLogLog: the analysis of a near-optimal cardinality estimation algorithm

改进:用调和平均数替代几何平均数

调和平均数

LLC是对各个桶取算数平均数，最终被应用到2的指数上，所以LLC取得是几何平均数。由于几何平均数对于离群值（例如这里的0）特别敏感，因此当存在离群值时，LLC的偏差就会很大.

当n较小,可能存在较多空桶，而这些特殊的离群值强烈干扰了几何平均数的稳定性。

估值公式以及误差:

根据论文中的分析结论，与LLC一样HLLC是渐近无偏估计，且其渐近标准差为：

SE=1.04/sqrt(m)

分段偏差修正:

在HLLC的论文中，作者在实现建议部分还给出了在n相对于m较小或较大时的偏差修正方案

HyperLogLog Plus

HyperLogLog plus 是Google的改进版。来自论文 HyperLogLog in Practice: Algorithmic Engineering of a
State of The Art Cardinality Estimation Algorithm

是一个工程改进版，包括使用FNV64代替原始的32位Hash函数，分段偏差修正等工程上的改进，无法严格证明

Java中的相关实现

• twitter/algebird a scala library from twitter which contains lots of different algorithms including HLL
• prasanthj/hyperloglog a detached java library for HLL
• addthis/stream-lib - another java lib(本人之前在项目中使用的这个库，还不错)

• aggregateknowledge/java-hll a low-level java implementation of HLL

• Comparison of HLL implementations 各种实现的相关比较

JDK 11 ZGC简介

注：本文翻译自这篇文章

前言

ZGC是最近由Oracle为OpenJDK开源的新垃圾收集器。它主要由Per Liden编写。ZGC类似于Shenandoah或Azul的C4，专注于减少暂停时间的同时仍然压缩堆 。

虽然我不会在这里给出完整的介绍，但“压缩堆”只是意味着将仍然存活的对象移动到堆的其他区域.这样做有助于减少碎片，但通常这也意味着整个应用程序（包括其所有线程）需要暂停,这通常被称为Stop the world 。只有GC完成后，才能恢复应用程序。

在GC相关的文献中，应用程序通常称为mutator ，因为从GC的角度来看，应用程序会改变堆(mutates the heap)。根据堆的大小，这样的暂停可能需要几秒钟，这对于交互式应用程序来说可能是难以接受的。

有几种方法可以减少暂停时间：

• GC可以在压缩时使用多个线程（并行压缩 parallel compaction）
• 压缩工作也可以分为多个暂停（增量压缩 incremental compaction）
• 压缩堆的同时不暂停应用程序，或者只是很短时间暂停（并发压缩 concurrent compaction）
• Go的GC就是完全不压缩堆

如前所述，ZGC会进行并发压缩，这当然不是一个简单的实现功能，因此我想描述一下这是如何工作的。为什么这很复杂？

你需要将对象复制到另一个内存地址，同时另一个线程仍然可以读写旧对象。

如果对象已经复制成功，那么堆中仍有许多指向旧地址的引用需要更新到新地址。

虽然并发压缩（concurrent compaction）似乎是上述方案中降低暂停时间的最佳解决方案，但肯定会涉及一些权衡。因此，如果您不关心暂停时间，那么最好使用专注于吞吐量的GC。

GC屏障 (GC Barriers)

理解ZGC如何进行并发压缩的关键是Load barrier (通常在GC文献中称为Read barrier).这里简单介绍一下，详细的描述请看下面的Load Barrier一节。

如果GC有读取屏障（Load barrier），则在从堆读取引用时，GC需要执行一些额外操作。在Java中,也就是像执行这样的代码Object xxx=obj.field时需要额外操作。

对于像obj.field = value这样的操作，GC也可能需要写入屏障(叫做Write Barrier或者Store Barrier)[译注：在分代GC还有引用计数中会用到写入屏障].

这两个操作都比较特殊因为它们在每次读取或写入堆时发生的。Load Barrier和Store Barrier的名称有点令人困惑，但注意这个屏障与CPU的内存障碍是完全不同的两个概念

堆中的读取和写入都非常常见，因此两种GC屏障都需要非常高效，在常见情况下就是一些汇编代码。Read barrier通常比Write Barrier大一个数量级（可能会因应用程序而异），因此Read Barrier对性能要求更高。

例如，分代GC通常只需要一个写屏障，不需要读屏障。ZGC则需要一个读屏障但没有写屏障。对于并发压缩，我没有看到没有读取障碍的解决方案。

这里需要注意：即使GC需要某种类型的屏障，只有在读取或写入堆中的引用时需要它们。读取或写入像int或double这样的基本类型是不需要屏障的.

指针标记（Pointer tagging Or Colored Pointers ）

ZGC在堆引用中存储额外的元数据 ，在x64上是64 bit（ZGC目前不支持compressed oops和 class pointers）。64位中的48位用做x64上的虚拟内存地址 。虽然确切地说只有47位，因为第47位确定了位48-63的值（目前这些位都是0）。ZGC保留对象实际地址的前42位（在源代码中称为偏移量 ）。42位地址理论上就会有4TB的堆大小限制。其余的位用于这些标志： finalizable ， remapped ， marked1和marked0 （保留一位用于将来使用）。如下图所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

 6                 4 4 4  4 4                                             0
 3                 7 6 5  2 1                                             0
+-------------------+-+----+-----------------------------------------------+
|00000000 00000000 0|0|1111|11 11111111 11111111 11111111 11111111 11111111|
+-------------------+-+----+-----------------------------------------------+
|                   | |    |
|                   | |    * 41-0 Object Offset (42-bits, 4TB address space)
|                   | |
|                   | * 45-42 Metadata Bits (4-bits)  0001 = Marked0
|                   |                                 0010 = Marked1
|                   |                                 0100 = Remapped
|                   |                                 1000 = Finalizable
|                   |
|                   * 46-46 Unused (1-bit, always zero)
|
* 63-47 Fixed (17-bits, always zero)

在堆引用中具有元数据信息使得解引用更加昂贵，因为需要mask地址以获得没有元信息的真实地址。ZGC采用了一个很好的技巧来避免这种情况：

当从内存中读取时，会设置marked0 , marked1或remapped中的一个。

在偏移x处分配页面（allocating a page）时，ZGC将同一页面映射到3个不同的地址 ：

• for marked0 ：(0b0001 << 42) | x
• for marked1 ： (0b0010 << 42) | x
• for remapped ： (0b0100 << 42) | x

因此，ZGC从地址4TB开始保留16TB的地址空间（但实际上并未使用所有这些内存）。如下图：

1
2
3
4
5
6
7
8
9

+--------------------------------+ 0x0000140000000000 (20TB)
|         Remapped View          |
+--------------------------------+ 0x0000100000000000 (16TB)
|     (Reserved, but unused)     |
+--------------------------------+ 0x00000c0000000000 (12TB)
|         Marked1 View           |
+--------------------------------+ 0x0000080000000000 (8TB)
|         Marked0 View           |
+--------------------------------+ 0x0000040000000000 (4TB)

在任何时间点，只使用这三个视图中的一个。调试时可以取消映射（unmapped）未使用的视图来验证正确性。

Pages & Physical & Virtual Memory

Shenandoah将堆分成大量同样大小的区域 。除了不适合单个区域的大对象外，对象通常不会跨越多个区域。大对象被分配在多个连续区域中。我非常喜欢这种方法，因为它非常简单。

在这方面，ZGC与Shenandoah非常相似。在ZGC的说法中，区域称为页面Pages 。

与Shenandoah的主要区别：ZGC中的页面可以有不同的大小（但在x64上总是2MB的倍数）。

ZGC有3种不同的页面类型： 小型 （2MB大小）， 中型 （32MB大小）和大型 （2MB的倍数）。

在小页面中分配小对象（最大256KB大小），在中型页面中分配中型对象（最多4MB）。大页面中分配大于4MB的对象。大页面只能存储一个对象.小页面或中间页面可以分配多个。

有些令人困惑的是大页面实际上可能小于中等页面（例如，对于大小为6MB的大对象）。

ZGC的另一个不错的特性是，它还可以区分物理内存和虚拟内存。这背后的想法是通常有足够的虚拟内存（ZGC总是4TB），而物理内存更稀缺。物理内存可以扩展到最大堆大小（使用-Xmx设置），因此这比4 TB的虚拟内存要小得多。在ZGC中分配特定大小的页面意味着分配物理和虚拟内存。在ZGC中，物理内存不需要是连续的，虚拟内存空间是连续的。

为什么说这是一个不错的属性？

分配连续范围的虚拟内存是很容易的，因为我们通常有足够的虚拟内存。但在物理内存中有3个大小为2MB的空闲页面的情况很普通，但是对于大型对象分配我们需要6MB的连续内存。有足够的空闲物理内存，但不幸的是这个内存是不连续的。ZGC能够将这些非连续的物理页面映射到单个连续的虚拟内存空间。如果无法映射，我们就会耗尽内存（发生OOM）

标记和重新安置对象（Marking & Relocating objects）

垃圾回收主要分为两个阶段：标记和重新安置（实际上不止这两个阶段，你可以查阅源码）。

[译注：重新安置（Relocating）指的是把对象从一个内存区域移到另外一个区域，重映射(Remapping)只的是把指向老的地址的引用更新到新的地址]

一次GC从标记阶段开始，标记所有可到达的对象。在这个阶段结束时，我们知道哪些对象仍然存活，哪些对象是垃圾。ZGC将此信息存储在每个页面的Live Map中。Live Map是一个位图(bitmap) ，用于存储给定索引处的对象是否可达和/或最终可达（对于具有finalize method的对象而言）。

在标记阶段，应用程序线程中的load-barrier将未标记的引用推送到线程局部标记缓冲区。只要此缓冲区已满，GC线程就可以获得此缓冲区的所有权，并以递归方式遍历此缓冲区中的所有可到达对象。在应用程序线程中标记只是将引用推送到缓冲区，GC线程负责遍历对象图并更新Live map.

标记阶段结束后，ZGC要重新安置 Relocation set中的所有活动对象。

Relocation Set表示一组需要被回收的页面（Pages)，例如那些垃圾最多的页面。存活的对象由GC线程或应用程序线程通过读取屏障（Load Barrier）重新安置（relocated）（也就是放到新的地址去）.ZGC为Relocation set中的每个页面分配Forwarding table.

Forwarding table基本上是一个hash map，它存储一个对象已被重新安置到的地址（如果该对象已经被重新安置）。

ZGC方法的优点是我们只需要为relocation set中的页面分配forwarding table的空间.
相比之下，Shenandoah将转发指针存储在每个对象本身，这样就谁有一些额外的内存开销。

GC线程遍历 Relocation set中的存活对象，并重新安置（relocate）尚未重新安置的对象。这时可能发生应用程序线程和GC线程同时重新安置（relocate）同一个对象，在这种情况下，谁先relocate谁获胜，ZGC使用原子CAS操作来确定胜者。

当不处于marking阶段时，load-barrier会重新安置(relocates )/重新映射(remaps )从堆加载的所有引用。这确保了mutator看到的每个新引用都已指向对象的最新副本。重新映射（remaps）对象就是在forwarding table中查找新的对象地址。

一旦GC线程完成了relocation set的处理，重新安置阶段就完成了。虽然这意味着所有对象都已重新安置，但通常仍会有引用指向relocation set，需要将其重新映射（remapped ）到新地址。这些引用会被Load-Barrier自我修复。如果对于这些引用的读取发生的不够快，（也就是这段时间内，应用程序没有读到这些指向relocation set的引用），这些引用会在下一次mark阶段给修复。这意味着标记阶段还需要检查 forward table以重新映射(remap) （但不重新安置 ，所有对象之前阶段都保证被重新安置）对象到它们的新地址。

这也解释了为什么对象引用中有两个标记位（marked0 和marked1 ）。标记阶段在标记的marked0和marked1位之间交替。在重新安置阶段之后，仍可能存在未重定向（remapped）的引用，所以我们需要知道上一个gc周期的情况。如果新的标记阶段使用相同的标记位，则Load-Barrier就知道该引用为已标记。

(译注：这里看起来像是GC周期remap和mark可以重叠，实际上确实是重叠的。如图所示：

更详细的信息可以看这个Slide)

Load-Barrier

从堆中读取引用时，ZGC需要一个所谓的load-barrier（也称为read-barrier）。每次Java程序访问对象类型的字段时，我们都需要插入此load-barrier，例如obj.field 。访问某些其他原始类型的字段不需要屏障，例如obj.anInt或obj.anDouble 。ZGC不需要obj.field = someValue存储/写入障碍。

根据GC当前所处的阶段（存储在全局变量ZGlobalPhase中 ），如果尚未标记或重新安置对象，则屏障会标记对象或重新安置它

全局变量ZAddressGoodMask和ZAddressBadMask
存储对应的掩码，该掩码确定引用是否已被认为是好的（这意味着已经标记或重新映射/重新安置remapped/relocated）或者是否仍然需要一些操作。这些变量仅在标记开始阶段和重新安置阶段同时改变.ZGC源代码中的这个表格可以很好地概述这些掩码的状态：

1
2
3
4
5

               GoodMask         BadMask          WeakGoodMask     WeakBadMask
               --------------------------------------------------------------
Marked0        001              110              101              010
Marked1        010              101              110              001
Remapped       100              011              100              011

屏障的汇编代码可以在MacroAssembler for x64中看到，我只会为这个屏障显示一些伪汇编代码：

1
2
3
4
5

mov rax, [r10 + some_field_offset]
test rax, [address of ZAddressBadMask]
jnz load_barrier_mark_or_relocate

# otherwise reference in rax is considered good

第一个汇编指令从堆读取引用： r10存储对象引用， some_field_offset是一些字段偏移常量。加载的引用存储在rax寄存器中。

然后针对当前的坏掩码测试该引用（这只是一个位与）。此处不需要同步，因为ZAddressBadMask仅在STW时才更新。如果结果不为零，我们需要执行屏障。

屏障需要根据我们当前所处的GC阶段标记或重新安置对象。在此操作之后， 他需要更新存储在r10 + some_field_offset中的引用来指向新引用。这步操作是必要的，以便来该字段的后续加载返回正确的引用。

由于我们可能需要更新引用地址，因此我们需要使用两个寄存器r10和rax作为加载的引用和对象地址。正确的引用也需要存储到寄存器rax中 ，这样在后面的执行过程中我们就已经加载了正确的引用。

由于每个引用都需要标记或重新安置，因此在开始标记或重新安置阶段后，吞吐量可能会立即降低。当大多数引用被修复时，这应该会变得更快。

Stop-the-World 停顿

ZGC并没有彻底摆脱STW。收集器在开始标记,结束标记和开始重新安置时需要暂停。但这种暂停通常很短,只有几毫秒。

当开始标记时，ZGC遍历所有线程堆栈以标记root set。root set是遍历对象图的开始的地方。root set通常由本地和全局变量组成，但也包括其他内部VM结构（例如JNI句柄）。

结束标记阶段时需要再次暂停。在此暂停中，GC需要清空并遍历所有线程局部标记缓冲区。由于GC可能会发现一个未标记的大型子图，因此可能需要更长时间。ZGC试图通过在1毫秒后停止标记阶段的结束来避免这种情况。它返回到并发标记阶段，直到遍历整个对象图，然后可以再次开始结束标记阶段

启动重新安置阶段会再次暂停应用程序。此阶段与开始标记非常相似，不同之处在于此阶段重新安置Root Set中的对象。

结论

我希望我能简单介绍一下ZGC。我当然无法在一篇博客文章中描述有关此GC的所有细节。如果您需要更多信息，ZGC是开源的因此可以研究整个实现。

Kafka源码学习:网络通信层(2)

上一篇里面讲了Kafka的服务端是如何收发请求的，这一篇主要学习一下网络层收到的数据是如何跟业务逻辑层打交道的

上篇文章讲到了processCompletedReceives方法里面把completedReceives队列里面的NetworkReceive拿出来组成了RequestChannel.Session对象，然后把这个Session对象放到了RequestChannel.Request对象里面。
其中Session由一个principal和一个远端地址构成

Request则是由之前读到的请求的payload（NetworkReceive.payload）和当前的processorID，connectionId等信息组成，还有header:RequestHeader跟body:AbstractRequest两个成员。使用payload构造header和body。
请求头里面包含了api_key,api_version等信息。

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

public class RequestHeader extends AbstractRequestResponse {

    private static final Field API_KEY_FIELD = REQUEST_HEADER.get("api_key");
    private static final Field API_VERSION_FIELD = REQUEST_HEADER.get("api_version");
    private static final Field CLIENT_ID_FIELD = REQUEST_HEADER.get("client_id");
    private static final Field CORRELATION_ID_FIELD = REQUEST_HEADER.get("correlation_id");

    private final short apiKey;
    private final short apiVersion;
    private final String clientId;
    private final int correlationId;
    //省略
}

请求体根据请求头中不同的apiKey来构造不同的请求体。

1
2
3
4
5
6
7
8
9
10

// AbstractRequest.getRequest(header.apiKey, header.apiVersion, buffer)
 public static AbstractRequest getRequest(int requestId, short versionId, ByteBuffer buffer) {
        ApiKeys apiKey = ApiKeys.forId(requestId);
        switch (apiKey) {
            case PRODUCE:
                return ProduceRequest.parse(buffer, versionId);
            case FETCH:
                return FetchRequest.parse(buffer, versionId);
            case LIST_OFFSETS:
            //以下省略

这样一个完整的request就构造出来了，构造出来之后，放入了这个processor所有的请求队列里面（RequestChannel.requestQueue.put(request)）,processCompletedReceives方法结束。

放入队列之后谁又来拿出来呢？

放入队列的请求被一个叫KafkaRequestHandler的类拿出来并且处理，代码如下：（我通过IDE的OPEN CALL hierarchy找到的）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

/**
 * A thread that answers kafka requests.
 */
class KafkaRequestHandler(id: Int,
                          brokerId: Int,
                          val aggregateIdleMeter: Meter,
                          val totalHandlerThreads: Int,
                          val requestChannel: RequestChannel,
                          apis: KafkaApis,
                          time: Time) extends Runnable with Logging {
  this.logIdent = "[Kafka Request Handler " + id + " on Broker " + brokerId + "], "

  def run() {
    while(true) {
      try {
        var req : RequestChannel.Request = null
        while (req == null) {
          // We use a single meter for aggregate idle percentage for the thread pool.
          // Since meter is calculated as total_recorded_value / time_window and
          // time_window is independent of the number of threads, each recorded idle
          // time should be discounted by # threads.
          val startSelectTime = time.nanoseconds
          req = requestChannel.receiveRequest(300)
          val idleTime = time.nanoseconds - startSelectTime
          aggregateIdleMeter.mark(idleTime / totalHandlerThreads)
        }

        if(req eq RequestChannel.AllDone) {
          debug("Kafka request handler %d on broker %d received shut down command".format(
            id, brokerId))
          return
        }
        req.requestDequeueTimeMs = time.milliseconds
        trace("Kafka request handler %d on broker %d handling request %s".format(id, brokerId, req))
        apis.handle(req)
      } catch {
        case e: Throwable => error("Exception when handling request", e)
      }
    }
  }

  def shutdown(): Unit = requestChannel.sendRequest(RequestChannel.AllDone)
}


class KafkaRequestHandlerPool(val brokerId: Int,
                              val requestChannel: RequestChannel,
                              val apis: KafkaApis,
                              time: Time,
                              numThreads: Int) extends Logging with KafkaMetricsGroup {

  /* a meter to track the average free capacity of the request handlers */
  private val aggregateIdleMeter = newMeter("RequestHandlerAvgIdlePercent", "percent", TimeUnit.NANOSECONDS)

  this.logIdent = "[Kafka Request Handler on Broker " + brokerId + "], "
  val threads = new Array[Thread](numThreads)
  val runnables = new Array[KafkaRequestHandler](numThreads)
  for(i <- 0 until numThreads) {
    runnables(i) = new KafkaRequestHandler(i, brokerId, aggregateIdleMeter, numThreads, requestChannel, apis, time)
    threads(i) = Utils.daemonThread("kafka-request-handler-" + i, runnables(i))
    threads(i).start()
  }

  def shutdown() {
    info("shutting down")
    for(handler <- runnables)
      handler.shutdown
    for(thread <- threads)
      thread.join
    info("shut down completely")
  }
}

可以看到这是一个runnable，并且通过while(true)不停的拿请求出来，最后交给 apis.handle(req)方法来处理这个请求。

这个KafkaRequestHandler被下面的KafkaRequestHandlerPool给拉起来的。可以从代码看到是直接拉起了numThreads个线程，共用一个apis和requestChannel实例。

这个api跟kafkaRequestHandlerPool都是在kafkaServer.startup里面拉起来的。

1
2
3
4
5
6
7
8

//kafkaServer.startup方法
/* start processing requests */
apis = new KafkaApis(socketServer.requestChannel, replicaManager, adminManager, groupCoordinator,
  kafkaController, zkUtils, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
  clusterId, time)

requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, time,
  config.numIoThreads)

从上面的代码可以看到KafkaRequestHandlerPool这个处理请求的线程池的requestChannel是从socketServer里面直接拿到的，这个池的大小就是config.numIoThreads，与processor数量一致。

现在回到KafkaRequestHandler里面，看他的run方法最后是apis.handle(req)，这里简单看一下是如何处理请求的。

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

def handle(request: RequestChannel.Request) {
   try {
     trace("Handling request:%s from connection %s;securityProtocol:%s,principal:%s".
       format(request.requestDesc(true), request.connectionId, request.securityProtocol, request.session.principal))
     ApiKeys.forId(request.requestId) match {
       case ApiKeys.PRODUCE => handleProducerRequest(request)
       case ApiKeys.FETCH => handleFetchRequest(request)
       case ApiKeys.LIST_OFFSETS => handleOffsetRequest(request)
       case ApiKeys.METADATA => handleTopicMetadataRequest(request)
       //后面省略
     }
   }
}

可以这里是根据requestId来的一个模式匹配（可以理解成java里面的switch，scala没有switch）。然后进入对应的handleXXX函数，在具体的业务逻辑完成之后，调用

1

requestChannel.sendResponse(new RequestChannel.Response(request, respBody))

把响应发送到对应processor的响应队列中。然后响应被上一篇文章说的流程给发送出去.
handleXXX里面由于是多线程共同调用的，不少代码都是有锁来保护的。

总结一下：
Processor线程负责解析请求，并把请求放入到请求队列中，另一个HandlerPool不停的从请求队列里面拿出已经解析好的请求，并执行对应的业务逻辑，完成业务逻辑之后把响应组成response放到对应processorID的响应队列中，由对应的processor发出去。
这个handlerPool的大小与IO线程数目一致。

Kafka源码学习:网络通信层(1)

因为工作过程中基本没有涉及到直接使用网络API编程的，所以借这次学习Kafka源码的机会，一起把Java NIO Api以及网络的编程模型套路一起学习一下。

本文主要目的是学习Kafka如何使用NIO与客户端通信的:包括连接的建立，数据的读取，给客户端的应答。

本文涉及的Kafka的源码版本为0.10.2。

文章比较长，不想看细节可以直接看总结

服务器的整体启动流程

服务器是在KafkaServer类的startUp方法中的ServerSocket.startUp方法拉起来。下面看下这个ServerSocket.startUp具体是怎么做的。

startUp根据配置信息中的每个Endpoint（Ip+Port）拉起一个Acceptor，然后这个Acceptor再拉起对应num.network.threads数目的Processor（这里的Acceptor和Processor都是runnable，都是拉起单独的线程跑的）

Acceptor的作用是接受请求，它在ServerChannel上注册了一个Selector，这个Selector注册了SelectionKey.OP_ACCEPT事件用来接受请求。如果有Accept进来，就使用RoundRobin的方式决定一个Processor，然后拿到客户端的Channel，配置好非阻塞模式，放入选出的Processor的newConnections队列里面。

Acceptor启动的时候会拉起num.network.threads数目的Processor。

这一块的代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

     //初始化Acceptor与Processor
     var processorBeginIndex = 0
     config.listeners.foreach { endpoint =>
     val listenerName = endpoint.listenerName
     val securityProtocol = endpoint.securityProtocol
     val processorEndIndex = processorBeginIndex + numProcessorThreads

     for (i <- processorBeginIndex until processorEndIndex)
       processors(i) = newProcessor(i, connectionQuotas, listenerName, securityProtocol)

     val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId,
       processors.slice(processorBeginIndex, processorEndIndex), connectionQuotas)
     acceptors.put(endpoint, acceptor)
     Utils.newThread(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor, false).start()
     acceptor.awaitStartup()

     processorBeginIndex = processorEndIndex
    }

     //Acceptor拉起Processor
     private[kafka] class Acceptor(val endPoint: EndPoint,
                           val sendBufferSize: Int,
                           val recvBufferSize: Int,
                           brokerId: Int,
                           processors: Array[Processor],
                           connectionQuotas: ConnectionQuotas) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {

 private val nioSelector = NSelector.open()
 val serverChannel = openServerSocket(endPoint.host, endPoint.port)

 this.synchronized {
 //拉起Processor
 processors.foreach { processor =>
   Utils.newThread(s"kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",
     processor, false).start()
 }
}


 //Acceptor 的主逻辑
   var currentProcessor = 0
   while (isRunning) {
     try {
       val ready = nioSelector.select(500)
       if (ready > 0) {
         val keys = nioSelector.selectedKeys()
         val iter = keys.iterator()
         while (iter.hasNext && isRunning) {
           try {
             val key = iter.next
             iter.remove()
             if (key.isAcceptable)
               accept(key, processors(currentProcessor))
             else
               throw new IllegalStateException("Unrecognized key state for acceptor thread.")

             // round robin to the next processor thread
             currentProcessor = (currentProcessor + 1) % processors.length
           } catch {
             case e: Throwable => error("Error while accepting connection", e)
           }
         }
       }
     }
     catch {
       // We catch all the throwables to prevent the acceptor thread from exiting on exceptions due
       // to a select operation on a specific channel or a bad request. We don't want
       // the broker to stop responding to requests from other clients in these scenarios.
       case e: ControlThrowable => throw e
       case e: Throwable => error("Error occurred", e)
     }
   }


   //Accept方法具体实现
   def accept(key: SelectionKey, processor: Processor) {
     val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
     val socketChannel = serverSocketChannel.accept() //这个socketChannel就是客户端的channel了
     try {
     connectionQuotas.inc(socketChannel.socket().getInetAddress)
     socketChannel.configureBlocking(false)
     socketChannel.socket().setTcpNoDelay(true)
     socketChannel.socket().setKeepAlive(true)
     if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
         socketChannel.socket().setSendBufferSize(sendBufferSize)

     debug("Accepted connection from %s on %s and assigned it to processor %d, sendBufferSize [actual|requested]: [%d|%d] recvBufferSize [actual|requested]: [%d|%d]"
             .format(socketChannel.socket.getRemoteSocketAddress, socketChannel.socket.getLocalSocketAddress, processor.id,
                 socketChannel.socket.getSendBufferSize, sendBufferSize,
                 socketChannel.socket.getReceiveBufferSize, recvBufferSize))

     processor.accept(socketChannel)
     } catch {
     case e: TooManyConnectionsException =>
         info("Rejected connection from %s, address already has the configured maximum of %d connections.".format(e.ip, e.count))
         close(socketChannel)
     }
 }

 //processor.accept具体实现，其实是放入了newConnections队列里面
  def accept(socketChannel: SocketChannel) {
     newConnections.add(socketChannel)
     wakeup()
 }

Processor的作用主要是处理新建的连接，读取客户端的请求，响应客户端，处理客户端关闭。
下面主要来说下Processor的构成还有是如何处理这些动作的。

Processor结构

Processor里面组成部分先介绍3个：一个Selector(这个selector是kafka包装过nioSelector的,里面定义了已收到和已发出的队列等等)，一个上文提到的newConnections队列和一个RequestChannel。
这个RequestChannel是一个类，重要的结构有两个，一个RequestQueue ，一个ResponseQueues看名字就知道是用来存储请求和响应的。具体代码如下：

1
2
3
4
5
6
7
8

class RequestChannel(val numProcessors: Int, val queueSize: Int) extends KafkaMetricsGroup {
  private var responseListeners: List[(Int) => Unit] = Nil
  private val requestQueue = new ArrayBlockingQueue[RequestChannel.Request](queueSize)
  private val responseQueues = new Array[BlockingQueue[RequestChannel.Response]](numProcessors)
  for(i <- 0 until numProcessors)
    responseQueues(i) = new LinkedBlockingQueue[RequestChannel.Response]()

上文说过Processor是个Runnable，那么直接看它的run方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

override def run() {
  startupComplete()
  while (isRunning) {
    try {
      // setup any new connections that have been queued up
      configureNewConnections()
      // register any new responses for writing
      // 这个Response只是把response写到了transportLayer，并没有实际发送，实际发送是在下面的poll做的
      processNewResponses()
      poll()
      processCompletedReceives()
      processCompletedSends()
      processDisconnected()
    } catch {
      // We catch all the throwables here to prevent the processor thread from exiting. We do this because
      // letting a processor exit might cause a bigger impact on the broker. Usually the exceptions thrown would
      // be either associated with a specific socket channel or a bad request. We just ignore the bad socket channel
      // or request. This behavior might need to be reviewed if we see an exception that need the entire broker to stop.
      case e: ControlThrowable => throw e
      case e: Throwable =>
        error("Processor got uncaught exception.", e)
    }
  }

  debug("Closing selector - processor " + id)
  swallowError(closeAll())
  shutdownComplete()
}

try里面的代码说明了processor做的事情，下面逐一说明

配置新进连接 ConfigureNewConnections

配置新连接主要做了一件事情就是把Acceptor放到newConnection队列里面的客户端的channel拿出来，然后把channel注册上Selector和对应的SelectionKey.OP_READ，这样在select的时候就能拿到客户端发出的请求了，然后把channel进行包装成KafkaChannel并attach到原始channel上。根据channel的基本信息构建一个key，放入Selector维护的ChannelMap里面。具体代码如下:

这里说下KafkaChannel这个类，这个类里面有个TransportLayer和一个Send属性。TransportLayerk可以由一个SelectionKey构建出来，他主要负责底层对channel的read,write,close等操作（因为之前把channel attach到SelectionKey上了所以可以拿到channel）。send属性是用来维护要发送的应答的。Kafka实际是一个request-response模型，一个channel同一时间只有一个请求，不会有多个，所以send只有一个。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

//配置新连接
private def configureNewConnections() {
    while (!newConnections.isEmpty) {
      val channel = newConnections.poll()
      try {
        debug(s"Processor $id listening to new connection from ${channel.socket.getRemoteSocketAddress}")
        val localHost = channel.socket().getLocalAddress.getHostAddress
        val localPort = channel.socket().getLocalPort
        val remoteHost = channel.socket().getInetAddress.getHostAddress
        val remotePort = channel.socket().getPort
        val connectionId = ConnectionId(localHost, localPort, remoteHost, remotePort).toString
        selector.register(connectionId, channel)
      } catch {
        // We explicitly catch all non fatal exceptions and close the socket to avoid a socket leak. The other
        // throwables will be caught in processor and logged as uncaught exceptions.
        case NonFatal(e) =>
          val remoteAddress = channel.getRemoteAddress
          // need to close the channel here to avoid a socket leak.
          close(channel)
          error(s"Processor $id closed connection from $remoteAddress", e)
      }
    }
  }

   public void register(String id, SocketChannel socketChannel) throws ClosedChannelException {
        SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_READ);
        KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
        key.attach(channel);
        this.channels.put(id, channel);
    }

处理响应 ProcessNewResponses

根据当前到Processor的序号（上文提到了Processor是多个）从RequestChannel里面拿到属于自己的响应队列，然后拿出响应并根据responseAction做出响应。这里的responseAction有3种：

1. NoOpAction 啥也不做，只把SelectionKey.OP_READ重新注册到selector上
2. SendAction 根据response里面的destination（就是channelId）拿到对应的KafkaChannel对象，把要发送的Send对象放到KafkaChannel对应的Send字段上，然后把写操作打开( this.transportLayer.addInterestOps(SelectionKey.OP_WRITE))，注意并没有实际发送.然后在inflightResponses记录下这个响应
3. CloseConnectionAction 把对应channel给close掉。

SendAction有关代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

//processor.sendResponse
  protected[network] def sendResponse(response: RequestChannel.Response) {
    trace(s"Socket server received response to send, registering for write and sending data: $response")
    val channel = selector.channel(response.responseSend.destination)
    // `channel` can be null if the selector closed the connection because it was idle for too long
    if (channel == null) {
      warn(s"Attempting to send response via channel for which there is no open connection, connection id $id")
      response.request.updateRequestMetrics()
    }
    else {
      selector.send(response.responseSend)
      inflightResponses += (response.request.connectionId -> response)
    }
  }

//selector.send
    public void send(Send send) {
        String connectionId = send.destination();
        if (closingChannels.containsKey(connectionId))
            this.failedSends.add(connectionId);
        else {
            KafkaChannel channel = channelOrFail(connectionId, false);
            try {
                channel.setSend(send);
            } catch (CancelledKeyException e) {
                this.failedSends.add(connectionId);
                close(channel, false);
            }
        }
    }

    public void setSend(Send send) {
        if (this.send != null)
            throw new IllegalStateException("Attempt to begin a send operation with prior send operation still in progress.");
        this.send = send;
        this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);
    }

这里明显能看出对于一个Channel上，send只能有一个

真正的读写 poll

poll方法通过对selector进行select来确定readyKeys的数量，如果大于0，就pollSelectionKeys，具体看下poll和pollSelectionKeys这个方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

//processor.poll
private def poll() {
    try selector.poll(300)
    catch {
      case e @ (_: IllegalStateException | _: IOException) =>
        error(s"Closing processor $id due to illegal state or IO exception")
        swallow(closeAll())
        shutdownComplete()
        throw e
    }
  }
//selector.poll
 public void poll(long timeout) throws IOException {
        if (timeout < 0)
            throw new IllegalArgumentException("timeout should be >= 0");

        clear();

        if (hasStagedReceives() || !immediatelyConnectedKeys.isEmpty())
            timeout = 0;

        /* check ready keys */
        long startSelect = time.nanoseconds();
        int readyKeys = select(timeout);
        long endSelect = time.nanoseconds();
        this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());

        if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
            pollSelectionKeys(this.nioSelector.selectedKeys(), false, endSelect);
            pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
        }

        addToCompletedReceives();

        long endIo = time.nanoseconds();
        this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());

        // we use the time at the end of select to ensure that we don't close any connections that
        // have just been processed in pollSelectionKeys
        maybeCloseOldestConnection(endSelect);
    }

 private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys,
                                   boolean isImmediatelyConnected,
                                   long currentTimeNanos) {
        Iterator<SelectionKey> iterator = selectionKeys.iterator();
        while (iterator.hasNext()) {
            SelectionKey key = iterator.next();
            iterator.remove();
            KafkaChannel channel = channel(key);

            // register all per-connection metrics at once
            sensors.maybeRegisterConnectionMetrics(channel.id());
            if (idleExpiryManager != null)
                idleExpiryManager.update(channel.id(), currentTimeNanos);

            try {

                /* complete any connections that have finished their handshake (either normally or immediately) */
            	//似乎是客户端连接服务器的逻辑
                if (isImmediatelyConnected || key.isConnectable()) {
                    if (channel.finishConnect()) {
                        this.connected.add(channel.id());
                        this.sensors.connectionCreated.record();
                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        log.debug("Created socket with SO_RCVBUF = {}, SO_SNDBUF = {}, SO_TIMEOUT = {} to node {}",
                                socketChannel.socket().getReceiveBufferSize(),
                                socketChannel.socket().getSendBufferSize(),
                                socketChannel.socket().getSoTimeout(),
                                channel.id());
                    } else
                        continue;
                }

                /* if channel is not ready finish prepare */
                if (channel.isConnected() && !channel.ready())
                    channel.prepare();

                /* if channel is ready read from any connections that have readable data */
                //从底层channel读，按照包来读
                if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
                    NetworkReceive networkReceive;
                    while ((networkReceive = channel.read()) != null)
                        addToStagedReceives(channel, networkReceive);
                }

                /* if channel is ready write to any sockets that have space in their buffer and for which we have data */
                if (channel.ready() && key.isWritable()) {
                    Send send = channel.write();
                    if (send != null) {
                        this.completedSends.add(send);
                        this.sensors.recordBytesSent(channel.id(), send.size());
                    }
                }

                /* cancel any defunct sockets */
                if (!key.isValid())
                    close(channel, true);

            } catch (Exception e) {
                String desc = channel.socketDescription();
                if (e instanceof IOException)
                    log.debug("Connection with {} disconnected", desc, e);
                else
                    log.warn("Unexpected error from {}; closing connection", desc, e);
                close(channel, true);
            }
        }
    }

这里主要看后面的两个逻辑读和写，前面的似乎是客户端的逻辑，因为这个类是公用的，目前还不确定.
先看读逻辑

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

channel.ready() && key.isReadable() && !hasStagedReceive(channel)
NetworkReceive networkReceive;
                   while ((networkReceive = channel.read()) != null)
                       addToStagedReceives(channel, networkReceive);
   //addToStagedReceives方法
   private void addToStagedReceives(KafkaChannel channel, NetworkReceive receive) {
       if (!stagedReceives.containsKey(channel))
           stagedReceives.put(channel, new ArrayDeque<NetworkReceive>());

       Deque<NetworkReceive> deque = stagedReceives.get(channel);
       deque.add(receive);
   }

可以看到如果channel已经就绪，同时可读，也没有未处理的请求，就把数据读出来，放到stagedReceives这个队列里面去.看到这个while似乎请求包可以有多个，但是前面我们已经看到了，应答包只能有1个，这个有点奇怪了。这里有一篇2015年1月的文章也是分析kafka网络层的，那个时候对于channel，read还只能read一个包.这里需要对照客户端的代码一起看，看一下是不是客户端也用了send这个结构，只能send一个。

channel.read方法读取数据是标准的按包读，先读4个字节来确定包长度，然后开对应大小的buffer把剩下的读进来。
具体的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

public NetworkReceive read() throws IOException {
       NetworkReceive result = null;

       if (receive == null) {
           receive = new NetworkReceive(maxReceiveSize, id);
       }

       receive(receive);
       if (receive.complete()) {
           receive.payload().rewind();
           result = receive;
           receive = null;
       }
       return result;
   }

   private long receive(NetworkReceive receive) throws IOException {
       return receive.readFrom(transportLayer);
   }

   public long readFrom(ScatteringByteChannel channel) throws IOException {
       return readFromReadableChannel(channel);
   }

   // Need a method to read from ReadableByteChannel because BlockingChannel requires read with timeout
   // See: http://stackoverflow.com/questions/2866557/timeout-for-socketchannel-doesnt-work
   // This can go away after we get rid of BlockingChannel
   @Deprecated
   public long readFromReadableChannel(ReadableByteChannel channel) throws IOException {
       int read = 0;
       if (size.hasRemaining()) {
           int bytesRead = channel.read(size); //这个size是一个长度为4的bytebuffer
           if (bytesRead < 0)
               throw new EOFException();
           read += bytesRead;
           if (!size.hasRemaining()) {
               size.rewind();
               int receiveSize = size.getInt();
               if (receiveSize < 0)
                   throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");
               if (maxSize != UNLIMITED && receiveSize > maxSize)
                   throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");

               this.buffer = ByteBuffer.allocate(receiveSize);
           }
       }
       if (buffer != null) {
           int bytesRead = channel.read(buffer);
           if (bytesRead < 0)
               throw new EOFException();
           read += bytesRead;
       }

       return read;
   }

    @Override
   public boolean complete() {
       return !size.hasRemaining() && !buffer.hasRemaining();
   }

这里可以看到如果网络层数据不完整，就是size或者buffer有remaining的时候，就返回了一个空的result，等下一次poll的时候再来。

下面的写方法就是直接把channel上持有的send发出去，然后放入completedSends队列里面。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

//in selector poll
 if (channel.ready() && key.isWritable()) {
                    Send send = channel.write();
                    if (send != null) {
                        this.completedSends.add(send);
                        this.sensors.recordBytesSent(channel.id(), send.size());
                    }
                }
 //channel.write
 public Send write() throws IOException {
        Send result = null;
        if (send != null && send(send)) {
            result = send;
            send = null;
        }
        return result;
    }

   

    private boolean send(Send send) throws IOException {
        send.writeTo(transportLayer);
        if (send.completed())
            transportLayer.removeInterestOps(SelectionKey.OP_WRITE);

        return send.completed();
    }

返回到poll方法，在pollSelectionKeys之后会通过 addToCompletedReceives方法把stagedReceive放到CompletedReceives里面去。注意这两个结构都是kafka自己包装的Selector类里面的属性（这个类里面有个nio的Selector）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

private void addToCompletedReceives() {
    if (!this.stagedReceives.isEmpty()) {
        Iterator<Map.Entry<KafkaChannel, Deque<NetworkReceive>>> iter = this.stagedReceives.entrySet().iterator();
        while (iter.hasNext()) {
            Map.Entry<KafkaChannel, Deque<NetworkReceive>> entry = iter.next();
            KafkaChannel channel = entry.getKey();
            if (!channel.isMute()) {
                Deque<NetworkReceive> deque = entry.getValue();
                addToCompletedReceives(channel, deque);
                if (deque.isEmpty())
                    iter.remove();
            }
        }
    }
}

private void addToCompletedReceives(KafkaChannel channel, Deque<NetworkReceive> stagedDeque) {
    NetworkReceive networkReceive = stagedDeque.poll();
    this.completedReceives.add(networkReceive);
    this.sensors.recordBytesReceived(channel.id(), networkReceive.payload().limit());
}

处理收到的结果 processCompletedReceives

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

//processor.processCompletedReceives
private def processCompletedReceives() {
    selector.completedReceives.asScala.foreach { receive =>
      try {
        val openChannel = selector.channel(receive.source)
        val session = {
          // Only methods that are safe to call on a disconnected channel should be invoked on 'channel'.
          val channel = if (openChannel != null) openChannel else selector.closingChannel(receive.source)
          RequestChannel.Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, channel.principal.getName), channel.socketAddress)
        }
        val req = RequestChannel.Request(processor = id, connectionId = receive.source, session = session,
          buffer = receive.payload, startTimeMs = time.milliseconds, listenerName = listenerName,
          securityProtocol = securityProtocol)
        requestChannel.sendRequest(req)
        selector.mute(receive.source)
      } catch {
        case e @ (_: InvalidRequestException | _: SchemaException) =>
          // note that even though we got an exception, we can assume that receive.source is valid. Issues with constructing a valid receive object were handled earlier
          error(s"Closing socket for ${receive.source} because of error", e)
          close(selector, receive.source)
      }
    }
  }

这个方法就是把之前selector读到的请求包装城session对象和request对象，放入一开始说的requestChannel队列里面。然后把对应的channel mute掉，就是transportLayer.removeInterestOps(SelectionKey.OP_READ);

处理已经发送的请求 processCompletedSends

1
2
3
4
5
6
7
8
9
10

//processor.processCompletedSends
private def processCompletedSends() {
    selector.completedSends.asScala.foreach { send =>
      val resp = inflightResponses.remove(send.destination).getOrElse {
        throw new IllegalStateException(s"Send for ${send.destination} completed, but not in `inflightResponses`")
      }
      resp.request.updateRequestMetrics()
      selector.unmute(send.destination)
    }
  }

这个方法就是根据processor.poll方法中已经成功发送应答列表，把在ProcessNewResponses这一步放入的inflightResponses的response给remove掉，然后unmute，即使把OP_READ给加回来。

处理已经断开的连接processDisconnected

这个方法就是做一些清理工作，把还没发送的响应给remove掉。connectionQuotas减掉。其中selector的disconnected队列会在客户端发起close，或者客户端超时，或者poll方法在清理过期连接等等情况的时候被加入。
selector的close会先把channel给close掉，再放入这个队列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

//processor.processDisconnected
 private def processDisconnected() {
    selector.disconnected.asScala.foreach { connectionId =>
      val remoteHost = ConnectionId.fromString(connectionId).getOrElse {
        throw new IllegalStateException(s"connectionId has unexpected format: $connectionId")
      }.remoteHost
      inflightResponses.remove(connectionId).foreach(_.request.updateRequestMetrics())
      // the channel has been closed by the selector but the quotas still need to be updated
      connectionQuotas.dec(InetAddress.getByName(remoteHost))
    }
  }

  //selector.doClose
   private void doClose(KafkaChannel channel, boolean notifyDisconnect) {
        try {
            channel.close();
        } catch (IOException e) {
            log.error("Exception closing connection to node {}:", channel.id(), e);
        }
        this.sensors.connectionClosed.record();
        this.stagedReceives.remove(channel);
        if (notifyDisconnect)
            this.disconnected.add(channel.id());
    }

总结一下整个流程：

服务器通过一个Acceptor来接受请求，多个Processor来处理读写。Processor里面维护这kafka封装过的selector。Selector维护着注册在上面的channel，并且处理对于channel的实际读写。

Acepptor把接受到的连接交给Processor的newConnection队列，Processor拿到连接并注册到自己的Selector上。

Processor通过requestChannel（里面包含了请求与响应2个队列）拿到自己要处理的响应，通过selector发出，selector这时候只是找到对应的Channel并设置send属性，并打开OP_WRITE，并记录到inflightResponses队列。

然后processor通过poll方法，实际调用selector的poll方法把客户端的请求读出来放入CompletedReceives队列，把对客户端的响应写到客户端,并移除OP_WRITE，放入completedSends队列。

接着Processor把从CompletedReceives读到的请求构造成request，扔到requestChannel队列里面去，然后移除OP_READ（读完了，请求还没处理，不能在读了）

然后Processor从completedSends队列连取出之前成功发出的响应，把对应inflightResponses移除，同时打开OP_READ。（写好了响应，可以接着接受请求了）

最后处理连接断开，把对应inflightResponses里面的响应移除。

可以看到Selector自己维护着channels,completedSends，completedReceives，processor从selector连获取这些信息，而且发送与接受实际完成者都是Seletor。

从性能上来说，selector有多个，可以增加读写性能，因为建立连接远小于在连接上读写数据。

接下来会写一下接受到的请求如何与上层API交互。