Introduction：收纳技术相关的基础知识 Redis、RocketMQ、Zookeeper、Netty、Tomcat 等总结！

[TOC]

SPI

SPI 全称为 Service Provider Interface，是一种服务发现机制。SPI 的本质是将接口实现类的全限定名，配置在文件中，并由服务加载器读取配置文件，加载实现类。这样可以在运行时，动态为接口替换实现类，正因为该特性，我们可以很容易的通过 SPI 机制为程序提供拓展功能。

Java SPI

JDK 中 提供了一个 SPI 的功能，核心类是 java.util.ServiceLoader。其作用就是，可以通过类名获取在 META-INF/services/ 下的多个配置实现文件。为了解决上面的扩展问题，现在我们在META-INF/services/下创建一个com.github.yu120.test.SuperLoggerConfiguration文件（没有后缀）。文件中只有一行代码，那就是我们默认的com.github.yu120.test.XMLConfiguration（注意，一个文件里也可以写多个实现，回车分隔）。然后通过 ServiceLoader 获取我们的 SPI 机制配置的实现类：

// META-INF/services/com.github.test.test.SuperLoggerConfiguration:
com.github.yu120.test.XMLConfiguration

ServiceLoader<SuperLoggerConfiguration> serviceLoader = ServiceLoader.load(SuperLoggerConfiguration.class);
Iterator<SuperLoggerConfiguration> iterator = serviceLoader.iterator();
SuperLoggerConfiguration configuration;
while(iterator.hasNext()) {
        // 加载并初始化实现类
         configuration = iterator.next();
}
// 对最后一个configuration类调用configure方法
configuration.configure(configFile);

Dubbo SPI

Dubbo SPI 的相关逻辑被封装在了 ExtensionLoader类中，它的getExtensionLoader方法用于从缓存中获取与接口对应的ExtensionLoader，若缓存未命中，则创建一个新的实例。Dubbo SPI的核心思想其实很简单：

• 通过配置文件，解耦拓展接口和拓展实现类
• 通过IOC自动注入依赖的拓展实现类对象
• 通过URL参数，在运行时确认真正的自定义拓展类对象

Dubbo SPI 所需的配置文件需放置在 META-INF/dubbo 路径下，配置内容如下（以下demo来自dubbo官方文档）。

optimusPrime = org.apache.spi.OptimusPrime
bumblebee = org.apache.spi.Bumblebee

与 Java SPI 实现类配置不同，Dubbo SPI 是通过键值对的方式进行配置，这样我们可以按需加载指定的实现类。另外在使用时还需要在接口上标注 @SPI 注解。

Motan SPI

Motan使用SPI机制来实现模块间的访问，基于接口和name来获取实现类，降低了模块间的耦合。

Motan的SPI的实现在 motan-core/com/weibo/api/motan/core/extension 中。组织结构如下：

motan-core/com.weibo.api.motan.core.extension
    |-Activation：SPI的扩展功能，例如过滤、排序
    |-ActivationComparator：排序比较器
    |-ExtensionLoader：核心，主要负责SPI的扫描和加载
    |-Scope：模式枚举，单例、多例
    |-Spi：注解，作用在接口上，表明这个接口的实现可以通过SPI的形式加载
    |-SpiMeta：注解，作用在具体的SPI接口的实现类上，标注该扩展的名称

SpringBoot SPI

Spring 的 SPI 配置文件是一个固定的文件 - META-INF/spring.factories，功能上和 JDK 的类似，每个接口可以有多个扩展实现，使用起来非常简单：

// 获取所有factories文件中配置的LoggingSystemFactory
List<LoggingSystemFactory>> factories = SpringFactoriesLoader.loadFactories(LoggingSystemFactory.class, classLoader);

下面是一段 Spring Boot 中 spring.factories 的配置

# Logging Systems
org.springframework.boot.logging.LoggingSystemFactory=\
org.springframework.boot.logging.logback.LogbackLoggingSystem.Factory,\
org.springframework.boot.logging.log4j2.Log4J2LoggingSystem.Factory

# PropertySource Loaders
org.springframework.boot.env.PropertySourceLoader=\
org.springframework.boot.env.PropertiesPropertySourceLoader,\
org.springframework.boot.env.YamlPropertySourceLoader

Redis

Java类型所占字节数（或bit数）

注意：

• 英文的数字、字母或符号：1个字符 = 1个字节数
• 中文的数字、字母或符号：1个字符 = 2个字节数
• 计算机的基本单位：bit 。一个bit代表一个0或1，1个字节是8个bit
• 1TB=1024GB，1GB=1024MB，1MB=1024KB，1KB=1024B（字节，byte），1B=8b（bit，位）

线程模型

Redis内部使用文件事件处理器File Event Handler，这个文件事件处理器是单线程的所以Redis才叫做单线程的模型。它采用I/O多路复用机制同时监听多个Socket，将产生事件的Socket压入到内存队列中，事件分派器根据Socket上的事件类型来选择对应的事件处理器来进行处理。文件事件处理器包含5个部分：

• 多个Socket
• I/O多路复用程序
• Scocket队列
• 文件事件分派器
• 事件处理器（连接应答处理器、命令请求处理器、命令回复处理器）

通信流程

客户端与redis的一次通信过程：

• 请求类型1：客户端发起建立连接的请求

  • 服务端会产生一个AE_READABLE事件，I/O多路复用程序接收到server socket事件后，将该socket压入队列中
  • 文件事件分派器从队列中获取socket，交给连接应答处理器，创建一个可以和客户端交流的socket01
  • 将socket01的AE_READABLE事件与命令请求处理器关联

• 请求类型2：客户端发起set key value请求

  • socket01产生AE_READABLE事件，socket01压入队列
  • 将获取到的socket01与命令请求处理器关联
  • 命令请求处理器读取socket01中的key value，并在内存中完成对应的设置
  • 将socket01的AE_WRITABLE事件与命令回复处理器关联

• 请求类型3：服务端返回结果

  • Redis中的socket01会产生一个AE_WRITABLE事件，压入到队列中
  • 将获取到的socket01与命令回复处理器关联
  • 回复处理器对socket01输入操作结果，如ok。之后解除socket01的AE_WRITABLE事件与命令回复处理器的关联

文件事件处理器

• 基于 Reactor 模式开发了自己的网络事件处理器（文件事件处理器，file event handler）
• 文件事件处理器 使用 I/O 多路复用（multiplexing）程序来同时监听多个套接字，并根据套接字目前执行的任务来为套接字关联不同的事件处理器
• 当被监听的套接字准备好执行连接应答（accept）、读取（read）、写入（write）、关闭（close）等操作时， 与操作相对应的文件事件就会产生， 这时文件事件处理器就会调用套接字之前关联好的事件处理器来处理这些事件
• 文件事件处理器以单线程方式运行， 但通过使用 I/O 多路复用程序来监听多个套接字， 文件事件处理器既实现了高性能的网络通信模型， 又可以很好地与 redis 服务器中其他同样以单线程方式运行的模块进行对接， 这保持了 Redis 内部单线程设计的简单性

I/O多路复用

I/O多路复用的I/O是指网络I/O，多路指多个TCP连接(即socket或者channel），复用指复用一个或几个线程。意思说一个或一组线程处理多个TCP连接。最大优势是减少系统开销小，不必创建过多的进程/线程，也不必维护这些进程/线程。 I/O多路复用使用两个系统调用(select/poll/epoll和recvfrom)，blocking I/O只调用了recvfrom；select/poll/epoll 核心是可以同时处理多个connection，而不是更快，所以连接数不高的话，性能不一定比多线程+阻塞I/O好,多路复用模型中，每一个socket，设置为non-blocking,阻塞是被select这个函数block，而不是被socket阻塞的。

select机制 基本原理 客户端操作服务器时就会产生这三种文件描述符(简称fd)：writefds(写)、readfds(读)、和exceptfds(异常)。select会阻塞住监视3类文件描述符，等有数据、可读、可写、出异常 或超时、就会返回；返回后通过遍历fdset整个数组来找到就绪的描述符fd，然后进行对应的I/O操作。 优点

• 几乎在所有的平台上支持，跨平台支持性好

缺点

• 由于是采用轮询方式全盘扫描，会随着文件描述符FD数量增多而性能下降
• 每次调用 select()，需要把 fd 集合从用户态拷贝到内核态，并进行遍历(消息传递都是从内核到用户空间)
• 默认单个进程打开的FD有限制是1024个，可修改宏定义，但是效率仍然慢。

poll机制 基本原理与select一致，也是轮询+遍历；唯一的区别就是poll没有最大文件描述符限制（使用链表的方式存储fd）。

epoll机制 基本原理 没有fd个数限制，用户态拷贝到内核态只需要一次，使用时间通知机制来触发。通过epoll_ctl注册fd，一旦fd就绪就会通过callback回调机制来激活对应fd，进行相关的io操作。epoll之所以高性能是得益于它的三个函数：

• epoll_create()：系统启动时，在Linux内核里面申请一个B+树结构文件系统，返回epoll对象，也是一个fd
• epoll_ctl()：每新建一个连接，都通过该函数操作epoll对象，在这个对象里面修改添加删除对应的链接fd, 绑定一个callback函数
• epoll_wait()：轮训所有的callback集合，并完成对应的IO操作

优点

• 没fd这个限制，所支持的FD上限是操作系统的最大文件句柄数，1G内存大概支持10万个句柄
• 效率提高，使用回调通知而不是轮询的方式，不会随着FD数目的增加效率下降
• 内核和用户空间mmap同一块内存实现(mmap是一种内存映射文件方法，即将一个文件或其它对象映射到进程的地址空间)

例子：100万个连接，里面有1万个连接是活跃，我们可以对比 select、poll、epoll 的性能表现：

• select：不修改宏定义默认是1024，则需要100w/1024=977个进程才可以支持 100万连接，会使得CPU性能特别的差
• poll： 没有最大文件描述符限制，100万个链接则需要100w个fd，遍历都响应不过来了，还有空间的拷贝消耗大量资源
• epoll: 请求进来时就创建fd并绑定一个callback，主需要遍历1w个活跃连接的callback即可，即高效又不用内存拷贝

执行效率高

Redis是单线程模型为什么效率还这么高？

• 纯内存操作：数据存放在内存中，内存的响应时间大约是100纳秒，这是Redis每秒万亿级别访问的重要基础
• 非阻塞的I/O多路复用机制：Redis采用epoll做为I/O多路复用技术的实现，再加上Redis自身的事件处理模型将epoll中的连接，读写，关闭都转换为了时间，不在I/O上浪费过多的时间
• C语言实现：距离操作系统更近，执行速度会更快
• 单线程避免切换开销：单线程避免了多线程上下文切换的时间开销，预防了多线程可能产生的竞争问题

数据类型

在 Redis 中，常用的 5 种数据类型和应用场景如下：

• String： 缓存、计数器、分布式锁等
• List： 链表、队列、微博关注人时间轴列表等
• Hash： 用户信息、Hash 表等
• Set： 去重、赞、踩、共同好友等
• Zset： 访问量排行榜、点击量排行榜等

String(字符串)

String 数据结构是简单的 key-value 类型，value 不仅可以是 String，也可以是数字（当数字类型用 Long 可以表示的时候encoding 就是整型，其他都存储在 sdshdr 当做字符串）。

Hash(字典)

在 Memcached 中，我们经常将一些结构化的信息打包成 hashmap，在客户端序列化后存储为一个字符串的值（一般是 JSON 格式），比如用户的昵称、年龄、性别、积分等。这时候在需要修改其中某一项时，通常需要将字符串（JSON）取出来，然后进行反序列化，修改某一项的值，再序列化成字符串（JSON）存储回去。简单修改一个属性就干这么多事情，消耗必定是很大的，也不适用于一些可能并发操作的场合（比如两个并发的操作都需要修改积分）。而 Redis 的 Hash 结构可以使你像在数据库中 Update 一个属性一样只修改某一项属性值。

• 存储、读取、修改用户属性

实战场景

• 缓存： 经典使用场景，把常用信息，字符串，图片或者视频等信息放到redis中，redis作为缓存层，mysql做持久化层，降低mysql的读写压力
• 计数器：redis是单线程模型，一个命令执行完才会执行下一个，同时数据可以一步落地到其他的数据源
• session：常见方案spring session + redis实现session共享

List(列表)

List 说白了就是链表（redis 使用双端链表实现的 List），相信学过数据结构知识的人都应该能理解其结构。使用 List 结构，我们可以轻松地实现最新消息排行等功能（比如新浪微博的 TimeLine ）。List 的另一个应用就是消息队列，可以利用 List 的 *PUSH 操作，将任务存在 List 中，然后工作线程再用 POP 操作将任务取出进行执行。Redis 还提供了操作 List 中某一段元素的 API，你可以直接查询，删除 List 中某一段的元素。

• 微博 TimeLine
• 消息队列

使用列表的技巧

• lpush+lpop=Stack(栈)
• lpush+rpop=Queue（队列）
• lpush+ltrim=Capped Collection（有限集合）
• lpush+brpop=Message Queue（消息队列）

实战场景：

• timeline：例如微博的时间轴，有人发布微博，用lpush加入时间轴，展示新的列表信息。

Set(集合)

Set 就是一个集合，集合的概念就是一堆不重复值的组合。利用 Redis 提供的 Set 数据结构，可以存储一些集合性的数据。比如在微博应用中，可以将一个用户所有的关注人存在一个集合中，将其所有粉丝存在一个集合。因为 Redis 非常人性化的为集合提供了求交集、并集、差集等操作，那么就可以非常方便的实现如共同关注、共同喜好、二度好友等功能，对上面的所有集合操作，你还可以使用不同的命令选择将结果返回给客户端还是存集到一个新的集合中。

• 共同好友、二度好友
• 利用唯一性，可以统计访问网站的所有独立 IP
• 好友推荐的时候，根据 tag 求交集，大于某个 threshold 就可以推荐

实战场景;

• 标签（tag）,给用户添加标签，或者用户给消息添加标签，这样有同一标签或者类似标签的可以给推荐关注的事或者关注的人
• 点赞，或点踩，收藏等，可以放到set中实现

Sorted Set(有序集合)

和Sets相比，Sorted Sets是将 Set 中的元素增加了一个权重参数 score，使得集合中的元素能够按 score 进行有序排列，比如一个存储全班同学成绩的 Sorted Sets，其集合 value 可以是同学的学号，而 score 就可以是其考试得分，这样在数据插入集合的时候，就已经进行了天然的排序。另外还可以用 Sorted Sets 来做带权重的队列，比如普通消息的 score 为1，重要消息的 score 为2，然后工作线程可以选择按 score 的倒序来获取工作任务。让重要的任务优先执行。

• 带有权重的元素，比如一个游戏的用户得分排行榜
• 比较复杂的数据结构，一般用到的场景不算太多

实战场景：

• 排行榜：有序集合经典使用场景。例如小说视频等网站需要对用户上传的小说视频做排行榜，榜单可以按照用户关注数，更新时间，字数等打分，做排行。

特殊数据结构

HyperLogLog(基数统计)

HyperLogLog 主要的应用场景就是进行基数统计。实际上不会存储每个元素的值，它使用的是概率算法，通过存储元素的hash值的第一个1的位置，来计算元素数量。HyperLogLog 可用极小空间完成独立数统计。命令如下：

应用案例

如何统计 Google 主页面每天被多少个不同的账户访问过？

对于 Google 这种访问量巨大的网页而言，其实统计出有十亿的访问量或十亿零十万的访问量其实是没有太多的区别的，因此，在这种业务场景下，为了节省成本，其实可以只计算出一个大概的值，而没有必要计算出精准的值。

对于上面的场景，可以使用HashMap、BitMap和HyperLogLog来解决。对于这三种解决方案，这边做下对比：

• HashMap：算法简单，统计精度高，对于少量数据建议使用，但是对于大量的数据会占用很大内存空间
• BitMap：位图算法，具体内容可以参考我的这篇，统计精度高，虽然内存占用要比HashMap少，但是对于大量数据还是会占用较大内存
• HyperLogLog：存在一定误差，占用内存少，稳定占用 12k 左右内存，可以统计 2^64 个元素，对于上面举例的应用场景，建议使用

Geo(地理空间信息)

Geo主要用于存储地理位置信息，并对存储的信息进行操作（添加、获取、计算两位置之间距离、获取指定范围内位置集合、获取某地点指定范围内集合）。Redis支持将Geo信息存储到有序集合(zset)中，再通过Geohash算法进行填充。命令如下：

GEORADIUS

GEORADIUS key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count]

以给定的经纬度为中心， 返回键包含的位置元素当中， 与中心的距离不超过给定最大距离的所有位置元素。范围可以使用以下其中一个单位：

• m 表示单位为米
• km 表示单位为千米
• mi 表示单位为英里
• ft 表示单位为英尺

在给定以下可选项时， 命令会返回额外的信息：

• WITHDIST: 在返回位置元素的同时， 将位置元素与中心之间的距离也一并返回。 距离单位和范围单位保持一致
• WITHCOORD: 将位置元素的经度和维度也一并返回
• WITHHASH: 以 52 位有符号整数的形式， 返回位置元素经过原始 geohash 编码的有序集合分值。 这个选项主要用于底层应用或者调试， 实际中的作用并不大

命令默认返回未排序的位置元素。 通过以下两个参数， 用户可以指定被返回位置元素的排序方式：

• ASC: 根据中心的位置， 按照从近到远的方式返回位置元素
• DESC: 根据中心的位置， 按照从远到近的方式返回位置元素

在默认情况下， GEORADIUS 命令会返回所有匹配的位置元素。 虽然用户可以使用 COUNT <count> 选项去获取前 N 个匹配元素， 但是因为命令在内部可能会需要对所有被匹配的元素进行处理， 所以在对一个非常大的区域进行搜索时， 即使只使用 COUNT 选项去获取少量元素， 命令的执行速度也可能会非常慢。 但是从另一方面来说， 使用 COUNT 选项去减少需要返回的元素数量， 对于减少带宽来说仍然是非常有用的。

Pub/Sub(发布订阅)

发布订阅类似于广播功能。redis发布订阅包括 发布者、订阅者、Channel。常用命令如下：

Bitmap(位图)

Bitmap就是位图，其实也就是字节数组（byte array），用一串连续的2进制数字（0或1）表示，每一位所在的位置为偏移(offset)，位图就是用每一个二进制位来存放或者标记某个元素对应的值。通常是用来判断某个数据存不存在的，因为是用bit为单位来存储所以Bitmap本身会极大的节省储存空间。常用命令如下：

应用案例

有1亿用户，5千万登陆用户，那么统计每日用户的登录数。每一位标识一个用户ID，当某个用户访问我们的网站就在Bitmap中把标识此用户的位设置为1。使用set集合和Bitmap存储的对比：

应用场景

• 用户在线状态
• 用户签到状态
• 统计独立用户

BloomFilter(布隆过滤)

当一个元素被加入集合时，通过K个散列函数将这个元素映射成一个位数组中的K个点（使用多个哈希函数对元素key (bloom中不存value) 进行哈希，算出一个整数索引值，然后对位数组长度进行取模运算得到一个位置，每个无偏哈希函数都会得到一个不同的位置），把它们置为1。检索时，我们只要看看这些点是不是都是1就（大约）知道集合中有没有它了：

• 如果这些点有任何一个为0，则被检元素一定不在
• 如果都是1，并不能完全说明这个元素就一定存在其中，有可能这些位置为1是因为其他元素的存在，这就是布隆过滤器会出现误判的原因

应用场景

• 解决缓存穿透：事先把存在的key都放到redis的Bloom Filter 中，他的用途就是存在性检测，如果 BloomFilter 中不存在，那么数据一定不存在；如果 BloomFilter 中存在，实际数据也有可能会不存
• 黑名单校验：假设黑名单的数量是数以亿计的，存放起来就是非常耗费存储空间的，布隆过滤器则是一个较好的解决方案。把所有黑名单都放在布隆过滤器中，再收到邮件时，判断邮件地址是否在布隆过滤器中即可
• Web拦截器：用户第一次请求，将请求参数放入布隆过滤器中，当第二次请求时，先判断请求参数是否被布隆过滤器命中，从而提高缓存命中率

基于Bitmap数据结构

import com.google.common.base.Preconditions;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
public class RedisService {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 根据给定的布隆过滤器添加值
     */
    public <T> void addByBloomFilter(BloomFilterHelper<T> bloomFilterHelper, String key, T value) {
        Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");
        int[] offset = bloomFilterHelper.murmurHashOffset(value);
        for (int i : offset) {
            redisTemplate.opsForValue().setBit(key, i, true);
        }
    }

    /**
     * 根据给定的布隆过滤器判断值是否存在
     */
    public <T> boolean includeByBloomFilter(BloomFilterHelper<T> bloomFilterHelper, String key, T value) {
        Preconditions.checkArgument(bloomFilterHelper != null, "bloomFilterHelper不能为空");
        int[] offset = bloomFilterHelper.murmurHashOffset(value);
        for (int i : offset) {
            if (!redisTemplate.opsForValue().getBit(key, i)) {
                return false;
            }
        }

        return true;
    }
}

基于RedisBloom模块

RedisBloom模块提供了四种数据类型：

• Bloom Filter (布隆过滤器)
• Cuckoo Filter（布谷鸟过滤器）
• Count-Mins-Sketch
• Top-K

Bloom Filter 和 Cuckoo 用于确定（以给定的确定性）集合中是否存在某项。使用 Count-Min Sketch 来估算子线性空间中的项目数，使用 Top-K 维护K个最频繁项目的列表。

# 1.git 下载
[root@test ~]# git clone https://github.com/RedisBloom/RedisBloom.git
[root@test ~]# cd redisbloom
[root@test ~]# make

# 2.wget 下载
[root@test ~]# wget https://github.com/RedisBloom/RedisBloom/archive/v2.0.3.tar.gz
[root@test ~]# tar -zxvf RedisBloom-2.0.3.tar.gz
[root@test ~]# cd RedisBloom-2.0.3/
[root@test ~]# make

# 3.修改Redis Conf
[root@test ~]#vim /etc/redis.conf
# 在文件中添加下行
loadmodule /root/RedisBloom-2.0.3/redisbloom.so

# 4.启动Redis server
[root@test ~]# /redis-server /etc/redis.conf
# 或者启动服务时加载os文件
[root@test ~]# /redis-server /etc/redis.conf --loadmodule /root/RedisBloom/redisbloom.so

# 5.测试RedisBloom
[root@test ~]# redis-cli
127.0.0.1:6379> bf.add bloomFilter foo
127.0.0.1:6379> bf.exists bloomFilter foo
127.0.0.1:6379> cf.add cuckooFilter foo
127.0.0.1:6379> cf.exists cuckooFilter foo

底层数据结构

当然是为了追求速度，不同数据类型使用不同的数据结构速度才得以提升。每种数据类型都有一种或者多种数据结构来支撑，底层数据结构有 6 种。

SDS(简单动态字符)

C 语言字符串结构与 SDS 字符串结构对比图如上所示：

• SDS 中 len 保存这字符串的长度，O(1) 时间复杂度查询字符串长度信息
• 空间预分配：SDS 被修改后，程序不仅会为 SDS 分配所需要的必须空间，还会分配额外的未使用空间
• 惰性空间释放：当对 SDS 进行缩短操作时，程序并不会回收多余的内存空间，而是使用 free 字段将这些字节数量记录下来不释放，后面如果需要 append 操作，则直接使用 free 中未使用的空间，减少了内存的分配

hash表(字典)

Redis 整体就是一个 哈希表来保存所有的键值对，无论数据类型是 5 种的任意一种。哈希表，本质就是一个数组，每个元素被叫做哈希桶，不管什么数据类型，每个桶里面的 entry 保存着实际具体值的指针。

整个数据库就是一个全局哈希表，而哈希表的时间复杂度是 O(1)，只需要计算每个键的哈希值，便知道对应的哈希桶位置，定位桶里面的 entry 找到对应数据，这个也是 Redis 快的原因之一。

那 Hash 冲突怎么办？

当写入 Redis 的数据越来越多的时候，哈希冲突不可避免，会出现不同的 key 计算出一样的哈希值。Redis 通过链式哈希解决冲突：也就是同一个 桶里面的元素使用链表保存。但是当链表过长就会导致查找性能变差可能，所以 Redis 为了追求快，使用了两个全局哈希表。用于 rehash 操作，增加现有的哈希桶数量，减少哈希冲突。开始默认使用 hash 表 1 保存键值对数据，哈希表 2 此刻没有分配空间。

linkedList(双端列表)

后续版本对列表数据结构进行了改造，使用 quicklist 代替了 ziplist 和 linkedlist。quicklist 是 ziplist 和 linkedlist 的混合体，它将 linkedlist 按段切分，每一段使用 ziplist 来紧凑存储，多个 ziplist 之间使用双向指针串接起来。这也是为何 Redis 快的原因，不放过任何一个可以提升性能的细节。

zipList(压缩列表)

压缩列表是 List 、hash、 sorted Set 三种数据类型底层实现之一。当一个列表只有少量数据的时候，并且每个列表项要么就是小整数值，要么就是长度比较短的字符串，那么 Redis 就会使用压缩列表来做列表键的底层实现。

ziplist 是由一系列特殊编码的连续内存块组成的顺序型的数据结构，ziplist 中可以包含多个 entry 节点，每个节点可以存放整数或者字符串。ziplist 在表头有三个字段 zlbytes、zltail 和 zllen，分别表示列表占用字节数、列表尾的偏移量和列表中的 entry 个数；压缩列表在表尾还有一个 zlend，表示列表结束。

struct ziplist<T> {
    int32 zlbytes;           // 整个压缩列表占用字节数
    int32 zltail_offset;  // 最后一个元素距离压缩列表起始位置的偏移量，用于快速定位到最后一个节点
    int16 zllength;          // 元素个数
    T[] entries;              // 元素内容列表，挨个挨个紧凑存储
    int8 zlend;               // 标志压缩列表的结束，值恒为 0xFF
}

如果我们要查找定位第一个元素和最后一个元素，可以通过表头三个字段的长度直接定位，复杂度是 O(1)。而查找其他元素时，就没有这么高效了，只能逐个查找，此时的复杂度就是 O(N)。

skipList(跳跃表)

sorted set 类型的排序功能便是通过「跳跃列表」数据结构来实现。跳跃表（skiplist）是一种有序数据结构，它通过在每个节点中维持多个指向其他节点的指针，从而达到快速访问节点的目的。

跳跃表支持平均 O（logN）、最坏 O（N）复杂度的节点查找，还可以通过顺序性操作来批量处理节点。跳表在链表的基础上，增加了多层级索引，通过索引位置的几个跳转，实现数据的快速定位。

intset(整数数组)

当一个集合只包含整数值元素，并且这个集合的元素数量不多时，Redis 就会使用整数集合作为集合键的底层实现，节省内存。结构如下：

typedef struct intset{
     //编码方式
     uint32_t encoding;
     //集合包含的元素数量
     uint32_t length;
     //保存元素的数组
     int8_t contents[];
}intset;

contents 数组是整数集合的底层实现：整数集合的每个元素都是 contents 数组的一个数组项（item），各个项在数组中按值的大小从小到大有序地排列，并且数组中不包含任何重复项。length 属性记录了整数集合包含的元素数量，也即是 contents 数组的长度。

持久化机制

通常来说，应该同时使用两种持久化方案，以保证数据安全：

• 如果数据不敏感，且可以从其他地方重新生成，可以关闭持久化
• 如果数据比较重要，且能够承受几分钟的数据丢失，比如缓存等，只需要使用RDB即可
• 如果是用做内存数据，要使用Redis的持久化，建议是RDB和AOF都开启
• 如果只用AOF，优先使用everysec的配置选择，因为它在可靠性和性能之间取了一个平衡

当RDB与AOF两种方式都开启时，Redis会优先使用AOF恢复数据，因为AOF保存的文件比RDB文件更完整

RDB模式(内存快照)

RDB（Redis Database Backup File，Redis数据备份文件）持久化方式：是指用数据集快照的方式半持久化模式记录 Redis 数据库的所有键值对，在某个时间点将数据写入一个临时文件，持久化结束后，用这个临时文件替换上次持久化的文件，达到数据恢复。

优点

• RDB快照是一个压缩过的非常紧凑的文件。保存着某个时间点的数据集，适合做数据的备份，灾难恢复
• 可最大化Redis的的性能。在保存RDB文件，服务器进程只需要fork一个子进程来完成RDB文件创建，父进程不需要做IO操作
• 与AOF相比，恢复大数据集的时候会更快

缺点

• RDB的数据安全性是不如AOF的，保存整个数据集的过程是比繁重的，根据配置可能要几分钟才快照一次，如果服务器宕机，那么就可能丢失几分钟的数据
• Redis数据集较大时，fork的子进程要完成快照会比较耗CPU、耗时

① 创建

当 Redis 持久化时，程序会将当前内存中的数据库状态保存到磁盘中。创建 RDB 文件主要有两个 Redis 命令：SAVE 和 BGSAVE。

② 载入

服务器在载入 RDB 文件期间，会一直处于阻塞状态，直到载入工作完成为止。

save同步保存

save 命令是同步操作，执行命令时，会 阻塞 Redis 服务器进程，拒绝客户端发送的命令请求。

具体流程如下：

由于 save 命令是同步命令，会占用Redis的主进程。若Redis数据非常多时，save命令执行速度会非常慢，阻塞所有客户端的请求。因此很少在生产环境直接使用SAVE 命令，可以使用BGSAVE 命令代替。如果在BGSAVE命令的保存数据的子进程发生错误的时，用 SAVE命令保存最新的数据是最后的手段。

bgsave异步保存

bgsave 命令是异步操作，执行命令时，子进程执行保存工作，服务器还可以继续让主线程处理客户端发送的命令请求。

具体流程如下：

Redis使用Linux系统的fock()生成一个子进程来将DB数据保存到磁盘，主进程继续提供服务以供客户端调用。如果操作成功，可以通过客户端命令LASTSAVE来检查操作结果。

自动保存

可通过配置文件对 Redis 进行设置， 让它在“ N 秒内数据集至少有 M 个改动”这一条件被满足时， 自动进行数据集保存操作:

# RDB自动持久化规则
# 当 900 秒内有至少有 1 个键被改动时，自动进行数据集保存操作
save 900 1
# 当 300 秒内有至少有 10 个键被改动时，自动进行数据集保存操作
save 300 10
# 当 60 秒内有至少有 10000 个键被改动时，自动进行数据集保存操作
save 60 10000

# RDB持久化文件名
dbfilename dump-<port>.rdb

# 数据持久化文件存储目录
dir /var/lib/redis

# bgsave发生错误时是否停止写入，通常为yes
stop-writes-on-bgsave-error yes

# rdb文件是否使用压缩格式
rdbcompression yes

# 是否对rdb文件进行校验和检验，通常为yes
rdbchecksum yes

默认配置

RDB 文件默认的配置如下：

################################ SNAPSHOTTING  ################################
#
# Save the DB on disk:
# 在给定的秒数和给定的对数据库的写操作数下，自动持久化操作。
#  save <seconds> <changes>
# 
save 900 1
save 300 10
save 60 10000
#bgsave发生错误时是否停止写入，一般为yes
stop-writes-on-bgsave-error yes
#持久化时是否使用LZF压缩字符串对象?
rdbcompression yes
#是否对rdb文件进行校验和检验，通常为yes
rdbchecksum yes
# RDB持久化文件名
dbfilename dump.rdb
#持久化文件存储目录
dir ./

AOF模式(日志追加)

AOF（Append Only File，追加日志文件）持久化方式：是指所有的命令行记录以 Redis 命令请求协议的格式完全持久化存储保存为 aof 文件。Redis 是先执行命令，把数据写入内存，然后才记录日志。因为该模式是只追加的方式，所以没有任何磁盘寻址的开销，所以很快，有点像 Mysql 中的binlog，AOF更适合做热备。

优点

• 数据更完整，安全性更高，秒级数据丢失（取决fsync策略，如果是everysec，最多丢失1秒的数据）
• AOF文件是一个只进行追加的日志文件，且写入操作是以Redis协议的格式保存的，内容是可读的，适合误删紧急恢复

缺点

• 对于相同的数据集，AOF文件的体积要大于RDB文件，数据恢复也会比较慢
• 根据所使用的fsync策略，AOF的速度可能会慢于RDB。 不过在一般情况下，每秒fsync的性能依然非常高

持久化流程

① 命令追加

若 AOF 持久化功能处于打开状态，服务器在执行完一个命令后，会以协议格式将被执行的写命令追加到服务器状态的 aof_buf 缓冲区的末尾。

② 文件同步

服务器每次结束一个事件循环之前，都会调用 flushAppendOnlyFile 函数，这个函数会考虑是否需要将 aof_buf 缓冲区中的内容写入和保存到 AOF 文件里。flushAppendOnlyFile 函数执行以下流程：

• WRITE：根据条件，将 aof_buf 中的缓存写入到 AOF 文件；
• SAVE：根据条件，调用 fsync 或 fdatasync 函数，将 AOF 文件保存到磁盘中。

这个函数是由服务器配置的 appendfsync 的三个值：always、everysec、no 来影响的，也被称为三种策略：

• always：每条命令都会 fsync 到硬盘中，这样 redis 的写入数据就不会丢失。

  

• everysec：每秒都会刷新缓冲区到硬盘中(默认值)。

  

• no：根据当前操作系统的规则决定什么时候刷新到硬盘中，不需要我们来考虑。

  

数据加载

• 创建一个不带网络连接的伪客户端
• 从 AOF 文件中分析并读取出一条写命令
• 使用伪客户端执行被读出的写命令
• 一直执行步骤 2 和 3，直到 AOF 文件中的所有写命令都被处理完毕为止

文件重写

为何需要文件重写

• 为了解决 AOF 文件体积膨胀的问题
• 通过重写创建一个新的 AOF 文件来替代现有的 AOF 文件，新的 AOF 文件不会包含任何浪费空间的冗余命令

文件重写的实现原理

• 不需要对现有的 AOF 文件进行任何操作
• 从数据库中直接读取键现在的值
• 用一条命令记录键值对，从而代替之前记录这个键值对的多条命令

后台重写

为不阻塞父进程，Redis将AOF重写程序放到子进程里执行。在子进程执行AOF重写期间，服务器进程需要执行三个流程：

• 执行客户端发来的命令
• 将执行后的写命令追加到 AOF 缓冲区
• 将执行后的写命令追加到 AOF 重写缓冲区

默认配置

AOF 文件默认的配置如下：

############################## APPEND ONLY MODE ###############################
#开启AOF持久化方式
appendonly no
#AOF持久化文件名
appendfilename "appendonly.aof"
#每秒把缓冲区的数据fsync到磁盘
appendfsync everysec
# appendfsync no
#是否在执行重写时不同步数据到AOF文件
no-appendfsync-on-rewrite no
# 触发AOF文件执行重写的增长率
auto-aof-rewrite-percentage 100
#触发AOF文件执行重写的最小size
auto-aof-rewrite-min-size 64mb
#redis在恢复时，会忽略最后一条可能存在问题的指令
aof-load-truncated yes
#是否打开混合开关
aof-use-rdb-preamble yes

过期策略

过期策略用于处理过期缓存数据。Redis采用过期策略：惰性删除 + 定期删除。memcached采用过期策略：惰性删除。

定时过期

每个设置过期时间的key都需要创建一个定时器，到过期时间就会立即对key进行清除。该策略可以立即清除过期的数据，对内存很友好；但是会占用大量的CPU资源去处理过期的数据，从而影响缓存的响应时间和吞吐量。

惰性过期

只有当访问一个key时，才会判断该key是否已过期，过期则清除。该策略可以最大化地节省CPU资源，却对内存非常不友好。极端情况可能出现大量的过期key没有再次被访问，从而不会被清除，占用大量内存。

定期过期

每隔一定的时间，会扫描一定数量的数据库的expires字典中一定数量的key，并清除其中已过期的key。该策略是前两者的一个折中方案。通过调整定时扫描的时间间隔和每次扫描的限定耗时，可以在不同情况下使得CPU和内存资源达到最优的平衡效果。expires字典会保存所有设置了过期时间的key的过期时间数据，其中 key 是指向键空间中的某个键的指针，value是该键的毫秒精度的UNIX时间戳表示的过期时间。键空间是指该Redis集群中保存的所有键。

淘汰策略

Redis淘汰机制的存在是为了更好的使用内存，用一定的缓存丢失来换取内存的使用效率。当Redis内存快耗尽时，Redis会启动内存淘汰机制，将部分key清掉以腾出内存。当达到内存使用上限超过maxmemory时，可在配置文件redis.conf中指定 maxmemory-policy 的清理缓存方式。

# 配置最大内存限制
maxmemory 1000mb
# 配置淘汰策略
maxmemory-policy volatile-lru

LRU(最近最少使用)

• volatile-lru：从已设置过期时间的key中，挑选**最近最少使用(最长时间没有使用)**的key进行淘汰
• allkeys-lru：从所有key中，挑选最近最少使用的数据淘汰

LFU(最近最不经常使用)

• volatile-lfu：从已设置过期时间的key中，挑选**最近最不经常使用(使用次数最少)**的key进行淘汰
• allkeys-lfu：从所有key中，选择某段时间内内最近最不经常使用的数据淘汰

Random(随机淘汰)

• volatile-random：从已设置过期时间的key中，任意选择数据淘汰
• allkeys-random：从所有key中，任意选择数据淘汰

TTL(过期时间)

• volatile-ttl：从已设置过期时间的key中，挑选将要过期的数据淘汰
• allkeys-random：从所有key中，任意选择数据淘汰

No-Enviction(驱逐)

• noenviction（驱逐）：当达到最大内存时直接返回错误，不覆盖或逐出任何数据

部署架构

单节点(Single)

优点

• 架构简单，部署方便
• 高性价比：缓存使用时无需备用节点(单实例可用性可以用 supervisor 或 crontab 保证)，当然为了满足业务的高可用性，也可以牺牲一个备用节点，但同时刻只有一个实例对外提供服务
• 高性能

缺点

• 不保证数据的可靠性
• 在缓存使用，进程重启后，数据丢失，即使有备用的节点解决高可用性，但是仍然不能解决缓存预热问题，因此不适用于数据可靠性要求高的业务
• 高性能受限于单核CPU的处理能力(Redis是单线程机制)，CPU为主要瓶颈，所以适合操作命令简单，排序/计算较少场景

主从复制(Replication)

基本原理

主从复制模式中包含一个主数据库实例（Master）与一个或多个从数据库实例（Slave），如下图：

哨兵(Sentinel)

Sentinel主要作用如下：

• 监控：Sentinel 会不断的检查主服务器和从服务器是否正常运行
• 通知：当被监控的某个Redis服务器出现问题，Sentinel通过API脚本向管理员或者其他的应用程序发送通知
• 自动故障转移：当主节点不能正常工作时，Sentinel会开始一次自动的故障转移操作，它会将与失效主节点是主从关系的其中一个从节点升级为新的主节点，并且将其他的从节点指向新的主节点

集群(Cluster)

环境搭建

Redis安装及配置

Redis的安装十分简单，打开redis的官网 http://redis.io 。

• 下载一个最新版本的安装包，如 redis-version.tar.gz
• 解压 tar zxvf redis-version.tar.gz
• 执行 make (执行此命令可能会报错，例如确实gcc，一个个解决即可)

如果是 mac 电脑，安装redis将十分简单执行brew install redis即可。安装好redis之后，我们先不慌使用，先进行一些配置。打开redis.conf文件，我们主要关注以下配置：

port 6379             # 指定端口为 6379，也可自行修改 
daemonize yes         # 指定后台运行

单节点(Single)

安装好redis之后，我们来运行一下。启动redis的命令为 :

$ <redishome>/bin/redis-server path/to/redis.config

假设我们没有配置后台运行（即：daemonize no）,那么我们会看到如下启动日志：

93825:C 20 Jan 2019 11:43:22.640 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo
93825:C 20 Jan 2019 11:43:22.640 # Redis version=5.0.3, bits=64, commit=00000000, modified=0, pid=93825, just started
93825:C 20 Jan 2019 11:43:22.640 # Configuration loaded
93825:S 20 Jan 2019 11:43:22.641 * Increased maximum number of open files to 10032 (it was originally set to 256).
                _._                                                  
           _.-``__ ''-._                                             
      _.-``    `.  `_.  ''-._           Redis 5.0.3 (00000000/0) 64 bit
  .-`` .-```.  ```\/    _.,_ ''-._                                   
 (    '      ,       .-`  | `,    )     Running in standalone mode
 |`-._`-...-` __...-.``-._|'` _.-'|     Port: 6380
 |    `-._   `._    /     _.-'    |     PID: 93825
  `-._    `-._  `-./  _.-'    _.-'                                   
 |`-._`-._    `-.__.-'    _.-'_.-'|                                  
 |    `-._`-._        _.-'_.-'    |           http://redis.io        
  `-._    `-._`-.__.-'_.-'    _.-'                                   
 |`-._`-._    `-.__.-'    _.-'_.-'|                                  
 |    `-._`-._        _.-'_.-'    |                                  
  `-._    `-._`-.__.-'_.-'    _.-'                                   
      `-._    `-.__.-'    _.-'                                       
          `-._        _.-'                                           
              `-.__.-'   

主从复制(Replication)

Redis主从配置非常简单，过程如下(演示情况下主从配置在一台电脑上)：

第一步：复制两个redis配置文件（启动两个redis，只需要一份redis程序，两个不同的redis配置文件即可）

mkdir redis-master-slave
cp path/to/redis/conf/redis.conf path/to/redis-master-slave master.conf
cp path/to/redis/conf/redis.conf path/to/redis-master-slave slave.conf

第二步：修改配置

## master.conf
port 6379
 
## master.conf
port 6380
slaveof 127.0.0.1 6379

第三步：分别启动两个redis

redis-server path/to/redis-master-slave/master.conf
redis-server path/to/redis-master-slave/slave.conf

启动之后，打开两个命令行窗口，分别执行 telnet localhost 6379 和 telnet localhost 6380，然后分别在两个窗口中执行 info 命令,可以看到：

# Replication
role:master
 
# Replication
role:slave
master_host:127.0.0.1
master_port:6379

主从配置没问题。然后在master 窗口执行 set 之后，到slave窗口执行get，可以get到，说明主从同步成功。这时，我们如果在slave窗口执行 set ，会报错:

-READONLY You can't write against a read only replica.

因为从节点是只读的。

哨兵(Sentinel)

Sentinel是用来监控主从节点的健康情况。客户端连接Redis主从的时候，先连接Sentinel，Sentinel会告诉客户端主Redis的地址是多少，然后客户端连接上Redis并进行后续的操作。当主节点挂掉的时候，客户端就得不到连接了因而报错了，客户端重新向Sentinel询问主master的地址，然后客户端得到了[新选举出来的主Redis]，然后又可以愉快的操作了。

哨兵sentinel配置

为了说明sentinel的用处，我们做个试验。配置3个redis（1主2从），1个哨兵。步骤如下：

mkdir redis-sentinel
cd redis-sentinel
cp redis/path/conf/redis.conf path/to/redis-sentinel/redis01.conf
cp redis/path/conf/redis.conf path/to/redis-sentinel/redis02.conf
cp redis/path/conf/redis.conf path/to/redis-sentinel/redis03.conf
touch sentinel.conf

上我们创建了 3个redis配置文件，1个哨兵配置文件。我们将 redis01设置为master，将redis02，redis03设置为slave。

vim redis01.conf
port 63791
 
vim redis02.conf
port 63792
slaveof 127.0.0.1 63791
 
vim redis03.conf
port 63793
slaveof 127.0.0.1 63791
 
vim sentinel.conf
daemonize yes
port 26379
sentinel monitor mymaster 127.0.0.1 63793 1   # 下面解释含义

上面的主从配置都熟悉，只有哨兵配置 sentinel.conf，需要解释一下：

mymaster        # 为主节点名字，可以随便取，后面程序里边连接的时候要用到
127.0.0.1 63793 # 为主节点的 ip,port
1               # 后面的数字 1 表示选举主节点的时候，投票数。1表示有一个sentinel同意即可升级为master

启动哨兵

上面我们配置好了redis主从，1主2从，以及1个哨兵。下面我们分别启动redis，并启动哨兵：

redis-server path/to/redis-sentinel/redis01.conf
redis-server path/to/redis-sentinel/redis02.conf
redis-server path/to/redis-sentinel/redis03.conf
 
redis-server path/to/redis-sentinel/sentinel.conf --sentinel

启动之后，可以分别连接到 3个redis上，执行info查看主从信息。

模拟主节点宕机情况

运行上面的程序（注意，在实验这个效果的时候，可以将sleep时间加长或者for循环增多，以防程序提前停止，不便看整体效果），然后将主redis关掉，模拟redis挂掉的情况。现在主redis为redis01,端口为63791

redis-cli -p 63791 shutdown

集群(Cluster)

上述所做的这些工作只是保证了数据备份以及高可用，目前为止我们的程序一直都是向1台redis写数据，其他的redis只是备份而已。实际场景中，单个redis节点可能不满足要求，因为：

• 单个redis并发有限
• 单个redis接收所有数据，最终回导致内存太大，内存太大回导致rdb文件过大，从很大的rdb文件中同步恢复数据会很慢

所以需要redis cluster 即redis集群。Redis 集群是一个提供在多个Redis间节点间共享数据的程序集。Redis集群并不支持处理多个keys的命令，因为这需要在不同的节点间移动数据，从而达不到像Redis那样的性能,在高负载的情况下可能会导致不可预料的错误。Redis 集群通过分区来提供一定程度的可用性，在实际环境中当某个节点宕机或者不可达的情况下继续处理命令.。Redis 集群的优势：

• 自动分割数据到不同的节点上
• 整个集群的部分节点失败或者不可达的情况下能够继续处理命令

为了配置一个redis cluster,我们需要准备至少6台redis，为啥至少6台呢？我们可以在redis的官方文档中找到如下一句话：

Note that the minimal cluster that works as expected requires to contain at least three master nodes.

因为最小的redis集群，需要至少3个主节点，既然有3个主节点，而一个主节点搭配至少一个从节点，因此至少得6台redis。然而对我来说，就是复制6个redis配置文件。本实验的redis集群搭建依然在一台电脑上模拟。

配置 redis cluster 集群

上面提到，配置redis集群需要至少6个redis节点。因此我们需要准备及配置的节点如下：

# 主：redis01  从 redis02    slaveof redis01
# 主：redis03  从 redis04    slaveof redis03
# 主：redis05  从 redis06    slaveof redis05
mkdir redis-cluster
cd redis-cluster
mkdir redis01 到 redis06 6个文件夹
cp redis.conf 到 redis01 ... redis06
# 修改端口, 分别配置3组主从关系

启动redis集群

上面的配置完成之后，分别启动6个redis实例。配置正确的情况下，都可以启动成功。然后运行如下命令创建集群：

redis-5.0.3/src/redis-cli --cluster create 127.0.0.1:6371 127.0.0.1:6372 127.0.0.1:6373 127.0.0.1:6374 127.0.0.1:6375 127.0.0.1:6376 --cluster-replicas 1

注意，这里使用的是ip:port，而不是 domain:port ，因为我在使用 localhost:6371 之类的写法执行的时候碰到错误：

ERR Invalid node address specified: localhost:6371

执行成功之后，连接一台redis，执行 cluster info 会看到类似如下信息：

cluster_state:ok
cluster_slots_assigned:16384
cluster_slots_ok:16384
cluster_slots_pfail:0
cluster_slots_fail:0
cluster_known_nodes:6
cluster_size:3
cluster_current_epoch:6
cluster_my_epoch:1
cluster_stats_messages_ping_sent:1515
cluster_stats_messages_pong_sent:1506
cluster_stats_messages_sent:3021
cluster_stats_messages_ping_received:1501
cluster_stats_messages_pong_received:1515
cluster_stats_messages_meet_received:5
cluster_stats_messages_received:3021

我们可以看到cluster_state:ok，cluster_slots_ok:16384，cluster_size:3。

拓展方案

分区(Partitioning)

指在面临单机的存储空间瓶颈时，即将全部数据分散在多个Redis实例中，每个实例不需要关联，可以是完全独立的。

使用方式

• 客户端处理 和传统的数据库分库分表一样，可以从key入手，先进行计算，找到对应数据存储的实例在进行操作。
  • 范围角度，比如orderId:1orderId:1000放入实例1，orderId:1001orderId:2000放入实例2
  • 哈希计算，就像我们的hashmap一样，用hash函数加上位运算或者取模，高级玩法还有一致性Hash等操作，找到对应的实例进行操作
• 使用代理中间件 我们可以开发独立的代理中间件，屏蔽掉处理数据分片的逻辑，独立运行。当然Redis也有优秀的代理中间件，譬如Twemproxy，或者codis，可以结合场景选择是否使用

缺点

• 无缘多key操作，key都不一定在一个实例上，那么多key操作或者多key事务自然是不支持
• 维护成本，由于每个实例在物理和逻辑上，都属于单独的一个节点，缺乏统一管理
• 灵活性有限，范围分片还好，比如hash+MOD这种方式，如果想动态调整Redis实例的数量，就要考虑大量数据迁移

主从(Master-Slave)

分区暂时能解决单点无法容纳的数据量问题，但是一个Key还是只在一个实例上。主从则将数据从主节点同步到从节点，然后可做读写分离，将读流量均摊在各个从节点，可靠性也能提高。主从(Master-Slave)也就是复制(Replication)方式。

使用方式

• 作为主节点的Redis实例，并不要求配置任何参数，只需要正常启动
• 作为从节点的实例，使用配置文件或命令方式REPLICAOF 主节点Host 主节点port即可完成主从配置

缺点

• slave节点都是只读的，如果写流量大的场景，就有些力不从心
• 故障转移不友好，主节点挂掉后，写处理就无处安放，需要手工的设定新的主节点，如使用REPLICAOF no one 晋升为主节点，再梳理其他slave节点的新主配置，相对来说比较麻烦

哨兵(Sentinel)

主从的手工故障转移，肯定让人很难接受，自然就出现了高可用方案-哨兵（Sentinel）。我们可以在主从架构不变的场景，直接加入Redis Sentinel，对节点进行监控，来完成自动的故障发现与转移。并且还能够充当配置提供者，提供主节点的信息，就算发生了故障转移，也能提供正确的地址。

使用方式

Sentinel的最小配置，一行即可：

sentinel monitor <主节点别名> <主节点host> <主节点端口> <票数>

只需要配置master即可，然后用redis-sentinel <配置文件> 命令即可启用。哨兵数量建议在三个以上且为奇数。

使用场景问题

• 故障转移期间短暂的不可用，但其实官网的例子也给出了parallel-syncs参数来指定并行的同步实例数量，以免全部实例都在同步出现整体不可用的情况，相对来说要比手工的故障转移更加方便
• 分区逻辑需要自定义处理，虽然解决了主从下的高可用问题，但是Sentinel并没有提供分区解决方案，还需开发者考虑如何建设
• 既然是还是主从，如果异常的写流量搞垮了主节点，那么自动的“故障转移”会不会变成自动“灾难传递”，即slave提升为Master之后挂掉，又进行提升又被挂掉

集群(Cluster)

Cluster在分区管理上，使用了“哈希槽”(hash slot)这么一个概念，一共有16384个槽位，每个实例负责一部分槽，通过CRC16（key）&16383这样的公式，计算出来key所对应的槽位。

使用方式

配置文件

cluster-enabled yes
cluster-config-file "redis-node.conf"

启动命令

redis-cli --cluster create 127.0.0.1:7000 127.0.0.1:7001 \
127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 \
--cluster-replicas 1

存在问题

• 虽然是对分区良好支持，但也有一些分区的老问题。如如果不在同一个“槽”的数据，是没法使用类似mset的多键操作
• 在select命令页有提到, 集群模式下只能使用一个库，虽然平时一般也是这么用的，但是要了解一下
• 运维上也要谨慎，俗话说得好，“使用越简单底层越复杂”，启动搭建是很方便，使用时面对带宽消耗，数据倾斜等等具体问题时，还需人工介入，或者研究合适的配置参数

常见问题

题目：保证Redis 中的 20w 数据都是热点数据 说明是 被频繁访问的数据，并且要保证Redis的内存能够存放20w数据，要计算出Redis内存的大小。

• **保留热点数据：对于保留 Redis 热点数据来说，我们可以使用 Redis 的内存淘汰策略来实现，可以使用allkeys-lru淘汰策略，**该淘汰策略是从 Redis 的数据中挑选最近最少使用的数据删除，这样频繁被访问的数据就可以保留下来了

• **保证 Redis 只存20w的数据：**1个中文占2个字节，假如1条数据有100个中文，则1条数据占200字节，20w数据 乘以 200字节 等于 4000 字节（大概等于38M）;所以要保证能存20w数据，Redis 需要38M的内存

题目：MySQL里有2000w数据，redis中只存20w的数据，如何保证redis中的数据都是热点数据?

限定 Redis 占用的内存，Redis 会根据自身数据淘汰策略，加载热数据到内存。所以，计算一下 20W 数据大约占用的内存，然后设置一下 Redis 内存限制即可。

题目：假如Redis里面有1亿个key，其中有10w个key是以某个固定的已知的前缀开头的，如果将它们全部找出来？

使用 keys 指令可以扫出指定模式的 key 列表。对方接着追问：如果这个 Redis 正在给线上的业务提供服务，那使用 keys 指令会有什么问题？这个时候你要回答 Redis 关键的一个特性：Redis 的单线程的。keys 指令会导致线程阻塞一段时间，线上服务会停顿，直到指令执行完毕，服务才能恢复。这个时候可以使用 scan 指令，scan 指令可以无阻塞地提取出指定模式的 key 列表，但是会有一定的重复概率，在客户端做一次去重就可以了，但是整体所花费的时间会比直接用 keys 指令长。

Kafka

RocketMQ

RocketMQ 是阿里巴巴开源的分布式消息中间件。支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。它里面有几个区别于标准消息中件间的概念，如Group、Topic、Queue等。系统组成则由Producer、Consumer、Broker、NameServer等。

功能优势

• 削峰填谷：主要解决瞬时写压力大于应用服务能力导致消息丢失、系统奔溃等问题
• 应用解耦：解决不同重要程度、不同能力级别系统之间依赖导致一死全死
• 提升性能：当存在一对多调用时，可以发一条消息给消息系统，让消息系统通知相关系统
• 蓄流压测：线上有些链路不好压测，可以通过堆积一定量消息再放开来压测
• 异步处理：不需要同步执行的远程调用可以有效提高响应时间

架构设计

部署模型

角色

Broker

• 理解成RocketMQ本身
• Broker主要用于Producer和Consumer接收和发送消息
• Broker会定时向NameSrver提交自己的信息
• 是消息中间件的消息存储、转发服务器
• 每个Broker节点在启动时都会遍历NameServer列表，与每个NameServer建立长连接，注册自己的信息，之后定时上报

NameServer

• 理解成Zookeeper的效果，只是他没用zk，而是自己写了个NameServer来替代zk
• 底层由Netty实现，提供了路由管理、服务注册、服务发现的功能，是一个无状态节点
• NameServer是服务发现者，集群中各个角色（Producer、Broker、Consumer等）都需要定时向NameServer上报自己的状态，以便互相发现彼此，超时不上报的话，NameServer会把它从列表中剔除
• NameServer可以部署多个，当多个NameServer存在的时候，其他角色同时向他们上报信息，以保证高可用，
• NameServer集群间互不通信，没有主备的概念
• NameServer内存式存储，NameServer中的Broker、Topic等信息默认不会持久化，所以他是无状态节点

Producer

• 消息的生产者
• 随机选择其中一个NameServer节点建立长连接，获得Topic路由信息（包括Topic下的Queue，这些Queue分布在哪些Broker上等等）
• 接下来向提供Topic服务的Master建立长连接（因为RocketMQ只有Master才能写消息），且定时向Master发送心跳

Consumer

• 消息的消费者
• 通过NameServer集群获得Topic的路由信息，连接到对应的Broker上消费消息
• 由于Master和Slave都可以读取消息，因此Consumer会与Master和Slave都建立连接进行消费消息

核心流程

• Broker都注册到Nameserver上
• Producer发消息的时候会从Nameserver上获取发消息的Topic信息
• Producer向提供服务的所有Master建立长连接，且定时向Master发送心跳
• Consumer通过NameServer集群获得Topic的路由信息
• Consumer会与所有的Master和所有的Slave都建立连接进行监听新消息

实现原理

RocketMQ由NameServer注册中心集群、Producer生产者集群、Consumer消费者集群和若干Broker（RocketMQ进程）组成，它的架构原理是这样的：

• Broker在启动的时候去向所有的NameServer注册，并保持长连接，每30s发送一次心跳
• Producer在发送消息的时候从NameServer获取Broker服务器地址，根据负载均衡算法选择一台服务器来发送消息
• Conusmer消费消息的时候同样从NameServer获取Broker地址，然后主动拉取消息来消费

核心概念

Message(消息)

消息载体。Message发送或者消费的时候必须指定Topic。Message有一个可选的Tag项用于过滤消息，还可以添加额外的键值对。

Topic(主题)

消息的逻辑分类，发消息之前必须要指定一个topic才能发，就是将这条消息发送到这个topic上。消费消息的时候指定这个topic进行消费。就是逻辑分类。

Queue(队列)

1个Topic会被分为N个Queue，数量是可配置的。message本身其实是存储到queue上的，消费者消费的也是queue上的消息。多说一嘴，比如1个topic4个queue，有5个Consumer都在消费这个topic，那么会有一个consumer浪费掉了，因为负载均衡策略，每个consumer消费1个queue，5>4，溢出1个，这个会不工作。

Tag(标签)

Tag 是 Topic 的进一步细分，顾名思义，标签。每个发送的时候消息都能打tag，消费的时候可以根据tag进行过滤，选择性消费。

消费模式(Message Model)

消息模型：集群（Clustering）和广播（Broadcasting）

集群模式(Clustering)

生产者往某个队列里面发送消息，一个队列可以存储多个生产者的消息，一个队列也可以有多个消费者，但是消费者之间是竞争关系，即每条消息只能被一个消费者消费。

• 每条消息只需要被处理一次，Broker只会把消息发送给消费集群中的一个消费者
• 在消息重投时，不能保证路由到同一台机器上
• 消费状态由Broker维护

广播模式(Broadcasting)

为了解决一条消息能被多个消费者消费的问题，发布/订阅模型就来了。该模型是将消息发往一个Topic即主题中，所有订阅了这个 Topic 的订阅者都能消费这条消息。

• 消费进度由Consumer维护
• 保证每个消费者都消费一次消息
• 消费失败的消息不会重投

Message Order(消息顺序)

消息顺序：顺序（Orderly）和并发（Concurrently）

顺序（Orderly）

并发（Concurrently）

Producer Group(生产组)

消息生产者组。标识发送同一类消息的Producer，通常发送逻辑一致。发送普通消息的时候，仅标识使用，并无特别用处。若事务消息，如果某条发送某条消息的producer-A宕机，使得事务消息一直处于PREPARED状态并超时，则broker会回查同一个group的其 他producer，确认这条消息应该commit还是rollback。但开源版本并不完全支持事务消息（阉割了事务回查的代码）。

Consumer Group(消费组)

消息消费者组。标识一类Consumer的集合名称，这类Consumer通常消费一类消息，且消费逻辑一致。同一个Consumer Group下的各个实例将共同消费topic的消息，起到负载均衡的作用。消费进度以Consumer Group为粒度管理，不同Consumer Group之间消费进度彼此不受影响，即消息A被Consumer Group1消费过，也会再给Consumer Group2消费。

注： RocketMQ要求同一个Consumer Group的消费者必须要拥有相同的注册信息，即必须要听一样的topic(并且tag也一样)。

Offset

在 Topic 的消费过程中，由于消息需要被不同的组进行多次消费，所以消费完的消息并不会立即被删除，这就需要 RocketMQ 为每个消费组在每个队列上维护一个消费位置（Consumer Offset），这个位置之前的消息都被消费过，之后的消息都没有被消费过，每成功消费一条消息，消费位置就加一。这个消费位置是非常重要的概念，我们在使用消息队列的时候，丢消息的原因大多是由于消费位置处理不当导致的。

核心设计

消息清理

Broker中的消息被消费后不会立即删除，每条消息都会持久化到CommitLog中，每个Consumer连接到Broker后会维持消费进度信息，当有消息消费后只是当前Consumer的消费进度（CommitLog的offset）更新了。默认48小时后会删除不再使用的CommitLog文件：

• 检查这个文件最后访问时间
• 判断是否大于过期时间
• 指定时间删除，默认凌晨4点

/**
 * {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#isTimeToDelete()}
 */
private boolean isTimeToDelete() {
    // when = "04";
    String when = DefaultMessageStore.this.getMessageStoreConfig().getDeleteWhen();
    // 是04点，就返回true
    if (UtilAll.isItTimeToDo(when)) {
        return true;
    }
 // 不是04点，返回false
    return false;
}

/**
 * {@link org.apache.rocketmq.store.DefaultMessageStore.CleanCommitLogService#deleteExpiredFiles()}
 */
private void deleteExpiredFiles() {
    // isTimeToDelete()这个方法是判断是不是凌晨四点，是的话就执行删除逻辑。
    if (isTimeToDelete()) {
        // 默认是72，但是broker配置文件默认改成了48，所以新版本都是48。
        long fileReservedTime = 48 * 60 * 60 * 1000;
        deleteCount = DefaultMessageStore.this.commitLog.deleteExpiredFile(72 * 60 * 60 * 1000, xx, xx, xx);
    }
}
                                                                       
/**
 * {@link org.apache.rocketmq.store.CommitLog#deleteExpiredFile()}
 */
public int deleteExpiredFile(xxx) {
    // 这个方法的主逻辑就是遍历查找最后更改时间+过期时间，小于当前系统时间的话就删了（也就是小于48小时）。
    return this.mappedFileQueue.deleteExpiredFileByTime(72 * 60 * 60 * 1000, xx, xx, xx);
}

push or pull

RocketMQ没有真正意义的push，都是pull，虽然有push类，但实际底层实现采用的是长轮询机制，即拉取方式。Broker端属性 longPollingEnable 标记是否开启长轮询，默认开启。源码如下：

// {@link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage()}

// 拉取消息，结果放到pullCallback里
this.pullAPIWrapper.pullKernelImpl(pullCallback);

为什么要主动拉取消息而不使用事件监听方式？

事件驱动方式是建立好长连接，由事件（发送数据）的方式来实时推送。如果broker主动推送消息的话有可能push速度快，消费速度慢的情况，那么就会造成消息在consumer端堆积过多，同时又不能被其他consumer消费的情况。而pull的方式可以根据当前自身情况来pull，不会造成过多的压力而造成瓶颈。所以采取了pull的方式。

负载均衡

RocketMQ通过Topic在多Broker中分布式存储实现。

Producer端

发送端指定message queue发送消息到相应的broker，来达到写入时的负载均衡

• 提升写入吞吐量，当多个producer同时向一个broker写入数据的时候，性能会下降
• 消息分布在多broker中，为负载消费做准备

默认策略是随机选择：

• producer维护一个index
• 每次取节点会自增
• index向所有broker个数取余
• 自带容错策略

其他实现：

• SelectMessageQueueByHash

• • hash的是传入的args

• SelectMessageQueueByRandom

• SelectMessageQueueByMachineRoom 没有实现

也可以自定义实现MessageQueueSelector接口中的select方法

MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);

Consumer端

采用的是平均分配算法来进行负载均衡。

其他负载均衡算法

• 平均分配策略(默认)(AllocateMessageQueueAveragely)
• 环形分配策略(AllocateMessageQueueAveragelyByCircle)
• 手动配置分配策略(AllocateMessageQueueByConfig)
• 机房分配策略(AllocateMessageQueueByMachineRoom)
• 一致性哈希分配策略(AllocateMessageQueueConsistentHash)
• 靠近机房策略(AllocateMachineRoomNearby)

当消费负载均衡Consumer和Queue不对等的时候会发生什么？

Consumer和Queue会优先平均分配，如果Consumer少于Queue的个数，则会存在部分Consumer消费多个Queue的情况，如果Consumer等于Queue的个数，那就是一个Consumer消费一个Queue，如果Consumer个数大于Queue的个数，那么会有部分Consumer空余出来，白白的浪费了。

最佳实践

Producer

• Topic：消息主题，通过Topic对不同的业务消息进行分类

• Tag：消息标签，用来进一步区分某个Topic下的消息分类，消息从生产者发出即带上的属性

• key：每个消息在业务层面的唯一标识码，要设置到 keys 字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引(哈希索引)，应用可以通过 topic，key来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证key 尽可能唯一，这样可以避免潜在的哈希冲突

• 日志：消息发送成功或者失败，要打印消息日志，务必要打印 send result 和key 字段

• send：send消息方法，只要不抛异常，就代表发送成功。但是发送成功会有多个状态，在sendResult里定义

  • SEND_OK：消息发送成功
  • FLUSH_DISK_TIMEOUT：消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  • FLUSH_SLAVE_TIMEOUT：消息发送成功，但是服务器同步到Slave时超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
  • SLAVE_NOT_AVAILABLE：消息发送成功，但是此时slave不可用，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失

• 订阅关系一致

  多个Group ID订阅了多个Topic，并且每个Group ID里的多个消费者实例的订阅关系保持了一致。

  

Consumer

• 消费幂等

  为了防止消息重复消费导致业务处理异常，消息队列RocketMQ版的消费者在接收到消息后，有必要根据业务上的唯一Key对消息做幂等处理。消息重复的场景如下：

  • 发送时消息重复
  • 投递时消息重复
  • 负载均衡时消息重复（包括但不限于网络抖动、Broker重启以及消费者应用重启）

• 日志：消费时记录日志，以便后续定位问题

• 批量消费：尽量使用批量方式消费方式，可以很大程度上提高消费吞吐量

事务消息

MQ与DB一致性原理（两方事务）

事务消息就是MQ提供的类似XA的分布式事务能力，通过事务消息可以达到分布式事务的最终一致性。半事务消息就是MQ收到了生产者的消息，但是没有收到二次确认，不能投递的消息。实现原理如下：

• 生产者先发送一条半事务消息到MQ
• MQ收到消息后返回ack确认
• 生产者开始执行本地事务
• 如果事务执行成功发送commit到MQ，失败发送rollback
• 如果MQ长时间未收到生产者的二次确认commit或者rollback，MQ对生产者发起消息回查
• 生产者查询事务执行最终状态
• 根据查询事务状态再次提交二次确认

如果MQ收到二次确认commit，就可以把消息投递给消费者，反之如果是rollback，消息会保存下来并且在3天后被删除。

保证顺序

RocketMQ的消息是存储到Topic的Queue里面的，Queue本身是FIFO（First Int First Out）先进先出队列。所以单个Queue是可以保证有序性的。

顺序消息（FIFO 消息）是 MQ 提供的一种严格按照顺序进行发布和消费的消息类型。顺序消息由两个部分组成：

• 顺序发布
• 顺序消费

顺序消息包含两种类型：

• 分区顺序：一个Partition内所有的消息按照先进先出的顺序进行发布和消费
• 全局顺序：一个Topic内所有的消息按照先进先出的顺序进行发布和消费

对于两个订单的消息的原始数据：a1、b1、b2、a2、a3、b3（绝对时间下发生的顺序）：

• 在发送时，a订单的消息需要保持a1、a2、a3的顺序，b订单的消息也相同，但是a、b订单之间的消息没有顺序关系，这意味着a、b订单的消息可以在不同的线程中被发送出去
• 在存储时，需要分别保证a、b订单的消息的顺序，但是a、b订单之间的消息的顺序可以不保证

保持顺序发送

消息被发送时保持顺序。

保持顺序发送存储

消息被存储时保持和发送的顺序一致。

保持顺序消费

消息被消费时保持和存储的顺序一致。

MQPullConsumer

MQPullConsumer由用户控制线程，主动从服务端获取消息，每次获取到的是一个MessageQueue中的消息。PullResult中的List msgFoundList自然和存储顺序一致，用户需要再拿到这批消息后自己保证消费的顺序。

MQPushConsumer

对于PushConsumer，由用户注册MessageListener来消费消息，在客户端中需要保证调用MessageListener时消息的顺序性。

消息不丢失

一条消息从生产到被消费，将会经历三个阶段：

• 生产阶段：Producer 新建消息，然后通过网络将消息投递给 MQ Broker
• 存储阶段：消息将会存储在 Broker 端磁盘中
• 消息阶段：Consumer 将会从 Broker 拉取消息

以上任一阶段都可能会丢失消息，我们只要找到这三个阶段丢失消息原因，采用合理的办法避免丢失，就可以彻底解决消息丢失的问题。

生产阶段

Producer通过网络将消息发送给Broker，这个发送可能会发生丢失，比如网络延迟不可达等。

失败会自动重试，即使重试N次也不行后，那客户端也会知道消息没成功，这也可以自己补偿等，不会盲目影响到主业务逻辑。再比如即使Broker挂了，那还有其他Broker再提供服务了，高可用，不影响。

总结：同步发送+自动重试机制+多个Master节点

同步发送

有三种send方法，同步发送、异步发送、单向发送，可以采取同步发送的方式进行发送消息。

• 同步发送：发消息时会同步阻塞等待broker返回的结果，如果没成功，则不会收到SendResult，这种是最可靠的
• 异步发送：在回调方法里可以得知是否发送成功
• 单向发送（OneWay）：最不靠谱的一种发送方式，我们无法保证消息真正可达

/**
 * {@link org.apache.rocketmq.client.producer.DefaultMQProducer}
 */

// 同步发送
public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {}
// 异步发送，sendCallback作为回调
public void send(Message msg,SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {}
// 单向发送，不关心发送结果，最不靠谱
public void sendOneway(Message msg) throws MQClientException, RemotingException, InterruptedException {}

失败重试

发送消息如果失败或者超时了，则会自动重试。默认是重试3次，可以根据api进行更改，比如改为10次：

producer.setRetryTimesWhenSendFailed(10);

底层源码逻辑如下：

/**
 * {@link org.apache.rocketmq.client.producer.DefaultMQProducer#sendDefaultImpl(Message, CommunicationMode, SendCallback, long)}
 */

// 自动重试次数，this.defaultMQProducer.getRetryTimesWhenSendFailed()默认为2，如果是同步发送，默认重试3次，否则重试1次
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
int times = 0;
for (; times < timesTotal; times++) {
      // 选择发送的消息queue
    MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
    if (mqSelected != null) {
        try {
            // 真正的发送逻辑，sendKernelImpl。
            sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
            switch (communicationMode) {
                case ASYNC:
                    return null;
                case ONEWAY:
                    return null;
                case SYNC:
                    // 如果发送失败了，则continue，意味着还会再次进入for，继续重试发送
                    if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
                        if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
                            continue;
                        }
                    }
                    // 发送成功的话，将发送结果返回给调用者
                    return sendResult;
                default:
                    break;
            }
        } catch (RemotingException e) {
            continue;
        } catch (...) {
            continue;
        }
    }
}

故障切换

假设Broker宕机了，但是生产环境一般都是多M多S的，所以还会有其他Master节点继续提供服务，这也不会影响到我们发送消息，我们消息依然可达。因为比如恰巧发送到broker的时候，broker宕机了，producer收到broker的响应发送失败了，这时候producer会自动重试，这时候宕机的broker就被踢下线了， 所以producer会换一台broker发送消息。

Broker存储阶段

若想很严格的保证Broker存储消息阶段消息不丢失，则需要如下配置，但是性能肯定远差于默认配置：

# master 节点配置
flushDiskType = SYNC_FLUSH
brokerRole=SYNC_MASTER

# slave 节点配置
brokerRole=slave
flushDiskType = SYNC_FLUSH

设置Broker同步刷盘策略

设置Broker同步刷盘策略。默认情况下，消息只要到了 Broker 端，将会优先保存到内存中，然后立刻返回确认响应给生产者。随后 Broker 定期批量的将一组消息从内存异步刷入磁盘。这种方式减少 I/O 次数，可以取得更好的性能，但是如果发生机器断电，异常宕机等情况，消息还未及时刷入磁盘，就会出现丢失消息的情况。

若想保证 Broker 端不丢消息，保证消息的可靠性，我们需要将消息保存机制修改为同步刷盘方式，即消息存储磁盘成功，才会返回响应。修改 Broker 端配置如下：

# 默认情况为 ASYNC_FLUSH 
flushDiskType = SYNC_FLUSH 

若 Broker 未在同步刷盘时间内（默认为 5s）完成刷盘，将会返回 SendStatus.FLUSH_DISK_TIMEOUT 状态给生产者。

等待Master和Slave刷盘完

等待Master和Slave刷盘完。即使Broker设置了同步刷盘策略，但是Broker刷完盘后磁盘坏了，这会导致盘上的消息全丢了。但是如果即使是1主1从了，但是Master刷完盘后还没来得及同步给Slave就磁盘坏了，这会导致盘上的消息全丢了。所以我们还可以配置不仅是等Master刷完盘就通知Producer，而是等Master和Slave都刷完盘后才去通知Producer说消息ok了。

# 默认为 ASYNC_MASTER
brokerRole=SYNC_MASTER

消费阶段

消费失败了其实也是消息丢失的一种变体。

只有当消费模式为 MessageModel.CLUSTERING(集群模式) 时，Broker 才会自动进行重试，对于广播消息是不会重试的。对于一直无法消费成功的消息，RocketMQ 会在达到最大重试次数之后，将该消息投递至死信队列。然后我们需要关注死信队列，并对该死信消息业务做人工的补偿操作。

手动ACK确认

消费者会先把消息拉取到本地，然后进行业务逻辑，业务逻辑完成后手动进行ack确认，这时候才会真正的代表消费完成。而不是说pull到本地后消息就算消费完了。举个例子

 consumer.registerMessageListener(new MessageListenerConcurrently() {
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
         try{
             for (MessageExt msg : msgs) {
             	String str = new String(msg.getBody());
             	System.out.println(str);
         	}
             
             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
         } catch(Throwable t){
             log.error("消费异常:{}", msgs, t);
             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
         }
     }
 });

消费异常自动重试

• 业务消费方返回 ConsumeConcurrentlyStatus.RECONSUME_LATER
• 业务消费方返回 null
• 业务消费方主动/被动抛出异常

针对以上3种情况下，Broker一般会进行重试（默认最大重试16次），RocketMQ 采用了“时间衰减策略”进行消息的重复投递，即重试次数越多，消息消费成功的可能性越小。我们可以在 RocketMQ 的 broker.conf 配置文件中配置 Consumer 侧重试次数及时间间隔（距离第1次发送的时间间隔）, 配置如下：

    messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

消费者客户端，首先判断消费端有没有显式设置最大重试次数 MaxReconsumeTimes， 如果没有，则设置默认重试次数为 16，否则以设置的最大重试次数为准。

private int getMaxReconsumeTimes() {
    // default reconsume times: 16
    if (this.defaultMQPushConsumer.getMaxReconsumeTimes() == -1) {
    	return 16;
    } else {
    	return this.defaultMQPushConsumer.getMaxReconsumeTimes();
    }
}

消费超时无线重试

如果是消费超时情况，MQ会无限制的发送给消费端。这种情况就是Consumer端没有返回ConsumeConcurrentlyStatus. CONSUME_SUCCESS，也没有返回ConsumeConcurrentlyStatus.RECONSUME_LATER。

死信队列

死信的处理逻辑：

• 首先判断消息当前重试次数是否大于等于 16，或者消息延迟级别是否小于 0
• 只要满足上述的任意一个条件，设置新的 topic（死信 topic）为：%DLQ%+consumerGroup
• 进行前置属性的添加
• 将死信消息投递到上述步骤 2 建立的死信 topic 对应的死信队列中并落盘，使消息持久化

最后单独启动一个死信队列的消费者进行消费，然后进行人工干预处理失败的消息。

消息幂等

在所有消息系统中消费消息有三种模式：at-most-once（最多一次）、at-least-once（最少一次）和 exactly-only-once（精确仅一次），分布式消息系统都是在三者间取平衡，前两者是可行的并且被广泛使用。

• at-most-once：消息投递后不论消息是否被消费成功，不会再重复投递，有可能会导致消息未被消费，RocketMQ 未使用该方式
• at-lease-once：消息投递后，消费完成后，向服务器返回 ACK，没有消费则一定不会返回 ACK 消息。由于网络异常、客户端重启等原因，服务器未能收到客户端返回的 ACK，服务器则会再次投递，这就会导致可能重复消费，RocketMQ 通过 ACK 来确保消息至少被消费一次
• exactly-only-once：在分布式系统环境下，如果要实现该模式，巨大的开销不可避免。RocketMQ 没有保证此特性，无法避免消息重复，由业务上进行幂等性处理。必须下面两个条件都满足，才能认为消息是"Exactly Only Once"：
  • 发送消息阶段，不允许发送重复消息
  • 消费消息阶段，不允许消费重复的消息

Zookeeper

**下载地址：**http://www.apache.org/dist/zookeeper/

ZK特性

Zookeeper主要靠其 分布式数据一致性 为集群提供 分布式协调服务，即指在集群的节点中进行可靠的消息传递，来协调集群的工作。主要具有如下特点：

• **最终一致性：**Client无论连接到哪个Server，展示给它都是同一个视图，这是zookeeper最重要的性能
• **可靠性：**如果一个消息或事物被一台Server接受，那么它将被所有的服务器接受
• 实时性：Zookeeper不能保证强一致性，只保证顺序一致性和最终一致性，因此称为伪实时性。由于网络延时等原因，Zookeeper不能保证两个客户端能同时得到刚更新的数据，如果需要最新数据，应该在读数据之前调用sync()接口
• **原子性：**更新只能成功或者失败，没有中间状态
• **顺序性：**包括 全序（Total order） 和 因果顺序（Causal order）
  • **全序：**如果消息a在消息b之前发送，则所有Server应该看到相同的结果
  • **因果顺序：**如果消息a在消息b之前发生（a导致了b），并被一起发送，则a始终在b之前被执行

ZK角色

• Server（服务端）
  • **Leader（领导者）：**负责对所有对ZK状态变更的请求，将状态更新请求进行排序与编号，以保证集群内部消息处理的有序性
  • Learner（学习者）
    • **Follower（追随者）：**用于接收客户请求并向客户端返回结果，在选举过程中参与投票
    • **Observer（观察者）：**其作用是为了扩展集群来提供读取速度，可以接收客户端连接，将写请求转发给Leader节点，但不参与投票，只同步Leader状态
• **Client（客户端）：**请求发起方

每个Server在工作过程中有三种状态：

• **LOOKING：**当前Server不知道Leader是谁，正在搜寻
• **LEADING：**当前Server即为选举出来的Leader
• **FOLLOWING：**Leader已经选举出来，当前Server与之同步

Zookeeper集群中，有Leader、Follower和Observer三种角色

• 领导者（Leader）：负责进行投票的发起和决议，更新系统状态

  Leader服务器是整个ZooKeeper集群工作机制中的核心，其主要工作：

  • 事务请求的唯一调度和处理者，保证集群事务处理的顺序性
  • 集群内部各服务的调度者

• 跟随者（Follower）：用于接收客户端请求并给客户端返回结果，在选主过程中进行投票

  Follower服务器是ZooKeeper集群状态的跟随者，其主要工作：

  • 处理客户端非事务请求，转发事务请求给Leader服务器
  • 参与事务请求Proposal的投票
  • 参与Leader选举投票

• 观察者（Observer）：可以接受客户端连接，将写请求转发给 leader，但是observer 不参加投票的过程，只是为了扩展系统，提高读取的速度

  Observer是3.3.0 版本开始引入的一个服务器角色，它充当一个观察者角色——观察ZooKeeper集群的最新状态变化并将这些状态变更同步过来。其工作：

  • 处理客户端的非事务请求，转发事务请求给 Leader 服务器
  • 不参与任何形式的投票

数据模型

Zookeeper 的数据模型：

• 层次化的目录结构，命名符合常规文件系统规范，类似于Linux
• 每个节点在zookeeper中叫做znode,并且其有一个唯一的路径标识
• 节点Znode可以包含数据和子节点，但是EPHEMERAL类型的节点不能有子节点
• Znode中的数据可以有多个版本，比如某一个路径下存有多个数据版本，那么查询这个路径下的数据就需要带上版本
• 客户端应用可以在节点上设置监视器
• 节点不支持部分读写，而是一次性完整读写

Server工作状态

服务器具有四种状态，分别是 LOOKING、FOLLOWING、LEADING、OBSERVING。

• LOOKING：寻找Leader状态。当服务器处于该状态时，它会认为当前集群中没有 Leader，因此需要进入 Leader 选举状态
• FOLLOWING：跟随者状态。表明当前服务器角色是Follower
• LEADING：领导者状态。表明当前服务器角色是Leader
• OBSERVING：观察者状态。表明当前服务器角色是Observer

运行模式

Zookeeper 有三种运行模式：单机模式、伪集群模式和集群模式。

• 单机模式：这种模式一般适用于开发测试环境，一方面我们没有那么多机器资源，另外就是平时的开发调试并不需要极好的稳定性。
• 集群模式：一个 ZooKeeper 集群通常由一组机器组成，一般 3 台以上就可以组成一个可用的 ZooKeeper 集群了。组成 ZooKeeper 集群的每台机器都会在内存中维护当前的服务器状态，并且每台机器之间都会互相保持通信。
• 伪集群模式：这是一种特殊的集群模式，即集群的所有服务器都部署在一台机器上。当你手头上有一台比较好的机器，如果作为单机模式进行部署，就会浪费资源，这种情况下，ZooKeeper 允许你在一台机器上通过启动不同的端口来启动多个 ZooKeeper 服务实例，以此来以集群的特性来对外服务。

Leader选举

服务器启动的Leader选举

zookeeper集群初始化阶段，服务器（myid=1-5）**「依次」**启动，开始zookeeper选举Leader~

• 服务器1（myid=1）启动，当前只有一台服务器，无法完成Leader选举

• 服务器2（myid=2）启动，此时两台服务器能够相互通讯，开始进入Leader选举阶段

  • 每个服务器发出一个投票

    服务器1 和 服务器2都将自己作为Leader服务器进行投票，投票的基本元素包括：服务器的myid和ZXID，我们以（myid，ZXID）形式表示。初始阶段，服务器1和服务器2都会投给自己，即服务器1的投票为（1,0），服务器2的投票为（2,0），然后各自将这个投票发给集群中的其他所有机器。

  • 接受来自各个服务器的投票

    每个服务器都会接受来自其他服务器的投票。同时，服务器会校验投票的有效性，是否本轮投票、是否来自LOOKING状态的服务器。

  • 处理投票

    收到其他服务器的投票，会将别人的投票跟自己的投票PK，PK规则如下：

    • 优先检查ZXID。ZXID比较大的服务器优先作为leader。
    • 如果ZXID相同的话，就比较myid，myid比较大的服务器作为leader。服务器1的投票是（1,0），它收到投票是（2,0），两者zxid都是0，因为收到的myid=2，大于自己的myid=1，所以它更新自己的投票为（2,0），然后重新将投票发出去。对于服务器2呢，即不再需要更新自己的投票，把上一次的投票信息发出即可。

  • 统计投票

    每次投票后，服务器会统计所有投票，判断是否有过半的机器接受到相同的投票信息。服务器2收到两票，少于3（n/2+1,n为总服务器5），所以继续保持LOOKING状态

• 服务器3（myid=3）启动，继续进入Leader选举阶段

  跟前面流程一致，服务器1和2先投自己一票，因为服务器3的myid最大，所以大家把票改投给它。此时，服务器为3票（大于等于n/2+1）,所以服务器3当选为Leader。服务器1，2更改状态为FOLLOWING，服务器3更改状态为LEADING；

• 服务器4启动，发起一次选举。

  此时服务器1，2，3已经不是LOOKING状态，不会更改选票信息。选票信息结果：服务器3为3票，服务器4为1票。服务器4并更改状态为FOLLOWING；

• 服务器5启动，发起一次选举。

  同理，服务器也是把票投给服务器3，服务器5并更改状态为FOLLOWING；

• 投票结束，服务器3当选为Leader

服务器运行期间的Leader选举

zookeeper集群的五台服务器（myid=1-5）正在运行中，突然某个瞬间，Leader服务器3挂了，这时候便开始Leader选举~

• 变更状态

  Leader 服务器挂了之后，余下的非Observer服务器都会把自己的服务器状态更改为LOOKING，然后开始进入Leader选举流程。

• 每个服务器发起投票

  每个服务器都把票投给自己，因为是运行期间，所以每台服务器的ZXID可能不相同。假设服务1,2,4,5的zxid分别为333,666,999,888，则分别产生投票（1,333），（2，666），（4,999）和（5,888），然后各自将这个投票发给集群中的其他所有机器。

• 接受来自各个服务器的投票

• 处理投票

  投票规则是跟Zookeeper集群启动期间一致的，优先检查ZXID，大的优先作为Leader，所以显然服务器zxid=999具有优先权。

• 统计投票

• 改变服务器状态

节点（znode）

① 节点组成

每个znode由4部分组成:

• **path：**即节点名称，用于存放简单可视化的数据
• **stat：**即状态信息，描述该znode的版本，权限等信息
• **data：**与该znode关联的数据
• **children：**该znode下的子节点

② 节点类型

• PERSISTENT（持久化目录节点）

  客户端与zookeeper断开连接后，该节点依旧存在

• PERSISTENT_SEQUENTIAL（持久化顺序编号目录节点）

  客户端与zookeeper断开连接后，该节点依旧存在，只是Zookeeper给该节点名称进行顺序编号

• EPHEMERAL（临时目录节点）

  客户端与zookeeper断开连接后，该节点被删除

• EPHEMERAL_SEQUENTIAL（临时顺序编号目录节点）

  客户端与zookeeper断开连接后，该节点被删除，只是Zookeeper给该节点名称进行顺序编号

关键词

• zxid（事务ID号）

  ZooKeeper状态的每一次改变，都对应着一个递增的Transaction id，该id称为zxid。由于zxid的递增性质，如果zxid1小于zxid2，那么zxid1肯定先于zxid2发生。创建任意节点、更新任意节点的数据、删除任意节点，都会导致Zookeeper状态发生改变，从而导致zxid的值增加。

• session（会话连接）

  在Client和Server通信之前，首先需要建立连接，该连接称为session。连接建立后，如果发生连接超时、授权失败或显式关闭连接，连接便处于CLOSED状态，此时session结束。

Watcher监听机制

Zookeeper 允许客户端向服务端的某个Znode注册一个Watcher监听，当服务端的一些指定事件触发了这个Watcher，服务端会向指定客户端发送一个事件通知来实现分布式的通知功能，然后客户端根据 Watcher通知状态和事件类型做出业务上的改变。

• ZooKeeper的Watcher机制主要包括客户端线程、客户端 WatcherManager、Zookeeper服务器三部分
• 客户端向ZooKeeper服务器注册Watcher的同时，会将Watcher对象存储在客户端的WatchManager中
• 当zookeeper服务器触发watcher事件后，会向客户端发送通知， 客户端线程从 WatcherManager 中取出对应的 Watcher 对象来执行回调逻辑

Watcher特性总结

• 「一次性:」 一个Watch事件是一个一次性的触发器。一次性触发，客户端只会收到一次这样的信息
• 「异步的：」 Zookeeper服务器发送watcher的通知事件到客户端是异步的，不能期望能够监控到节点每次的变化，Zookeeper只能保证最终的一致性，而无法保证强一致性
• 「轻量级：」 Watcher 通知非常简单，它只是通知发生了事件，而不会传递事件对象内容
• 「客户端串行：」 执行客户端 Watcher 回调的过程是一个串行同步的过程
• 注册 watcher用getData、exists、getChildren方法
• 触发 watcher用create、delete、setData方法

ZXID

ZXID有两部分组成：

• 任期：完成本次选举后，直到下次选举前，由同一Leader负责协调写入
• 事务计数器：单调递增，每生效一次写入，计数器加一

ZXID的低32位是计数器，所以同一任期内，ZXID是连续的，每个结点又都保存着自身最新生效的ZXID，通过对比新提案的ZXID与自身最新ZXID是否相差“1”，来保证事务严格按照顺序生效的。

工作流程

① 客户端发起的操作的主要流程

Leader可以执行增删改查操作，而Follower只能进行查询操作。所有的更新操作都会被转交给Leader来处理，Leader批准的任务，再发送给Follower去执行来保证和Leader的一致性。由于网络是不稳定的，为了保证执行顺序的一致，所有的任务都会被赋予一个唯一的顺序的编号，一定是按照这个编号来执行任务，保证任务顺序的一致性。

② 客户端的请求什么时候才能算处理成功？为什么说集群过半机器宕机后无法再工作？

Leader在收到客户端提交过来的任务后，会向集群中所有的Follower发送提案等待Follower的投票，Follower们收到这个提议后，会进行投票，同意或者不同意，Leader会回收Follower的投票，一旦受到过半的投票表示同意，则Leader认为这个提案通过，再发送命令要求所有的Follower都进行这个提案中的任务。由于需要过半的机器同意才能执行任务，所以一旦集群中过半的机器挂掉，整个集群就无法工作了。

• Leader工作流程

  Leader主要有三个功能：

  • 恢复数据
  • 维持与Learner的心跳，接收Learner请求并判断Learner的请求消息类型
  • Learner的消息类型主要有如下四种，根据不同的消息类型，进行不同的处理：
    • **PING消息：**指Learner的心跳信息
    • **REQUEST消息：**Follower发送的提议信息，包括写请求及同步请求
    • **ACK消息：**Follower的对提议的回复，超过半数的Follower通过，则commit该提议
    • **REVALIDATE消息：**用来延长SESSION有效时间

• Follower工作流程

  Follower主要有四个功能：

  • 向Leader发送消息请求（PING消息、REQUEST消息、ACK消息、REVALIDATE消息）
  • 接收Leader消息并进行处理
  • 接收Client的请求，如果为写请求，发送给Leader进行投票
  • 返回Client结果

  Follower的消息循环处理如下几种来自Leader的消息：

  • PING消息： 心跳消息

• **PROPOSAL消息：**Leader发起的提案，要求Follower投票

  • **COMMIT消息：**服务器端最新一次提案的信息
  • **UPTODATE消息：**表明同步完成
  • **REVALIDATE消息：**根据Leader的REVALIDATE结果，关闭待revalidate的session还是允许其接受消息
  • **SYNC消息：**返回SYNC结果到客户端，这个消息最初由客户端发起，用来强制得到最新的更新

• Observer工作流程

  对于Observer的流程不再叙述，Observer流程和Follower的唯一不同的地方就是Observer不会参加Leader发起的投票。

ZAB协议

ZooKeeper并没有完全采用Paxos算法，而是使用一种称为ZooKeeper Atomic Broadcast（ZAB，Zookeeper原子消息广播协议）的协议作为其数据一致性的核心算法，简称ZAB协议。ZAB协议分为两种模式：

• **崩溃恢复模式（Recovery）：**当服务初次启动或Leader节点挂了时，系统就会进入恢复模式，直到选出了有合法数量Follower的新Leader，然后新Leader负责将整个系统同步到最新状态
• **消息广播模式（Boardcast）：**ZAB协议中，所有的写请求都由Leader来处理。正常工作状态下，Leader接收请求并通过广播协议来处理（如：广播提议投票、广播命令）

① 崩溃恢复模式（Recovery）

为了使Leader挂了后系统能正常工作，需要解决以下两个问题：

• 已经被处理的消息不能丢
• 被丢弃的消息不能再次出现

② 消息广播模式（Boardcast）

广播的过程实际上是一个简化的二阶段提交过程：

• Leader 接收到消息请求后，将消息赋予一个全局唯一的 64 位自增 id，叫做：zxid，通过 zxid 的大小比较即可实现因果有序这一特性
• Leader 通过先进先出队列（通过 TCP 协议来实现，以此实现了全局有序这一特性）将带有 zxid 的消息作为一个提案（proposal）分发给所有 follower
• 当 follower 接收到 proposal，先将 proposal 写到硬盘，写硬盘成功后再向 leader 回一个 ACK
• 当 leader 接收到合法数量的 ACKs 后，leader 就向所有 follower 发送 COMMIT 命令，同事会在本地执行该消息
• 当 follower 收到消息的 COMMIT 命令时，就会执行该消息

总结

个人认为 Zab 协议设计的优秀之处有两点，一是简化二阶段提交，提升了在正常工作情况下的性能；二是巧妙地利用率自增序列，简化了异常恢复的逻辑，也很好地保证了顺序处理这一特性。值得注意的是，ZAB提交事务并不像2PC一样需要全部Follower都ACK，只需要得到quorum（超过半数的节点）的ACK就可以了。

③ ZAB 的四个阶段

• **Leader election（选举阶段）：**节点在一开始都处于选举阶段，只要有一个节点得到超半数节点的票数，它就可以当选准Leader（只有完成ZAB的四个阶段，准Leader才会成为真正的Leader）。本阶段的目的是就是为了选出一个准Leader，然后进入下一个阶段
• **Discovery（发现阶段）：**在次阶段，Followers跟准Leader进行通信，同步Followers最近接收的事务提议，这个一阶段的主要目的是发现当前大多数节点接收的最新提议
• **Synchronization（同步阶段）：**同步阶段主要是利用Leader前一阶段获得的最新提议历史，同步集群中所有的副本。只有当quorum都同步完成，准Leader才会成为真正的Leader。Follower只会接收zxid比自己的lastZxid大的提议
• **Broadcast（广播阶段）：**到了这个阶段，Zookeeper 集群才能正式对外提供事务服务，并且Leader可以进行消息广播。同时如果有新的节点加入，还需要对新节点进行同步

④ JAVA版ZAB协议

• Fast Leader Election：Fast Leader Election

  前面提到 FLE 会选举拥有最新提议历史（lastZixd最大）的节点作为 leader，这样就省去了发现最新提议的步骤。这是基于拥有最新提议的节点也有最新提交记录的前提。成为 leader 的条件：

  • 选epoch最大的
  • epoch相等，选 zxid 最大的
  • epoch和zxid都相等，选择server id最大的（即配置zoo.cfg中的myid）

• Recovery Phase：这一阶段Follower发送它们的 lastZixd 给Leader，Leader 根据 lastZixd 决定如何同步数据。这里的实现跟前面 Phase 2 有所不同：Follower 收到 TRUNC 指令会中止 L.lastCommittedZxid 之后的提议，收到 DIFF 指令会接收新的提议。

• Broadcast Phase：暂无

ZK选举过程

最开始集群启动时，会选择xzid最小的机器作为leader。

当Leader崩溃或者Leader失去大多数的Follower，这时候ZK进入恢复模式，恢复模式需要重新选举出一个新的Leader，让所有的Server都恢复到一个正确的状态。ZK的选举算法使用ZAB协议：

• 选举线程由当前Server发起选举的线程担任，其主要功能是对投票结果进行统计，并选出推荐的Server
• 选举线程首先向所有Server发起一次询问（包括自己）
• 选举线程收到回复后，验证是否是自己发起的询问(验证zxid是否一致)，然后获取对方的id(myid)，并存储到当前询问对象列表中，最后获取对方提议的leader相关信息(id,zxid)，并将这些信息存储到当次选举的投票记录表中
• 收到所有Server回复以后，就计算出zxid最大的那个Server，并将这个Server相关信息设置成下一次要投票的Server
• 线程将当前zxid最大的Server设置为当前Server要推荐的Leader，如果此时获胜的Server获得n/2 + 1的Server票数， 设置当前推荐的leader为获胜的Server，将根据获胜的Server相关信息设置自己的状态，否则，继续这个过程，直到leader被选举出来

分析结论

要使Leader获得多数Server的支持，则Server总数最好是奇数2n+1，且存活的Server的数目不得少于n+1。因为需要过半存活集群才能工作，所以2n个机器提供的集群可靠性其实和2n-1个机器提供的集群可靠性是一样的。

Zookeeper安装

单机模式

第一步：安装部署

# 下载解压
wget http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gz
tar -zxvf zookeeper-3.4.9.tar.gz

# 设置全局变量
vim ~/.bash_profile
# 最后一行加入
export ZOOKEEPER_HOME=/home/zookeeper/zookeeper-3.4.9
export PATH=$ZOOKEEPER_HOME/bin:$PATH
# 使之生效
source ~/.bash_profile

# 复制配置文件
cp /home/zookeeper/zookeeper-3.4.9/conf/zoo_sample.cfg /home/zookeeper/zookeeper-3.4.9/conf/zoo.cfg

第二步：配置信息

# 心跳间隔
tickTime=2000
# 保存数据目录
dataDir=/home/zookeeper/zookeeper-3.4.9/dataDir
# 保存日志目录
dataLogDir=/home/zookeeper/zookeeper-3.4.9/dataLogDir
# 客户端连接Zookeeper的端口
clientPort=2181
# 允许follower连接并同步到leader的初始化连接时间(心跳倍数)，超过则连接失败
initLimit=5
# 表示leader和follower之间发送消息时, 请求和应答的最大时间长度
syncLimit=2

集群模式

第一步：安装部署

安装方式参考单机模式。

第二步：配置信息

在dataDir根目录下新建myid文件，并向文件myid中写入值(如：1、2、3……)

# 1表示当前集群的id号
echo "1" > myid

在单机配置情况下，新增下述参数：

# 格式：server.X=A:B:C
# X表示myid(范围1~255)
# A是该server所在的IP地址
# B配置该server和集群中的leader交换消息所使用的端口,即数据同步端口
# C配置选举leader时所使用的端口
server.1=10.24.1.62:2888:3888
server.2=10.24.1.63:2888:3888
server.3=10.24.1.64:2888:3888

运维命令

# 启动ZK服务器
./zkServer.sh start
# 使用ZK Client连接指定服务器
./zkCli.sh -server 127.0.0.1:2181
# 查看ZK服务状态
./zkServer.sh status
# 停止ZK服务
./zkServer.sh stop
# 重启ZK服务
./zkServer.sh restart

zoo.cfg配置参数

# 客户端连接server的端口，默认值2181
clientPort=2181
# 存储快照文件snapshot的目录
dataDir=/User/lry/zookeeper/data
# ZK中的最小时间单元，单位为毫秒
tickTime=5000
# 事务日志输出目录
dataLogDir=/User/lry/zookeeper/datalog
# Server端最大允许的请求堆积数，默认值为1000  
globalOutstandingLimit=1000
# 每个事务日志文件的大小，默认值64M
preAllocSize=64
# 每进行snapCount次事务日志输出后，触发一次快照,默认值100000,实际代码中是随机范围触发，避免并发情况
snapCount=100000
# 单个客户端与单台服务器之间的连接数的限制，是ip级别的，默认是60，如果设置为0，那么表明不作任何限制
maxClientCnxns=60
# 对于多网卡的机器，可以为每个IP指定不同的监听端口。默认是所有IP都监听clientPort指定的端口
clientPortAddress=10.24.22.56
# Session超时时间限制，如果客户端设置的超时时间不在这个范围，那么会被强制设置为最大或最小时间。默认的Session超时时间是在2*tickTime~20*tickTime这个范围
minSessionTimeoutmaxSessionTimeout
# Follower在启动过程中，会从Leader同步所有最新数据，然后确定自己能够对外服务的起始状态。Leader允许F在 initLimit 时间内完成这个工作。
initLimit
# 在运行过程中，Leader负责与ZK集群中所有机器进行通信，例如通过一些心跳检测机制，来检测机器的存活状态。如果L发出心跳包在syncLimit之后，还没有从F那里收到响应，那么就认为这个F已经不在线了。
syncLimit
# 默认情况下，Leader是会接受客户端连接，并提供正常的读写服务。但是，如果你想让Leader专注于集群中机器的协调，那么可以将这个参数设置为no，这样一来，会大大提高写操作的性能
leaderServes=yes
# server.[myid]=[hostname]:[数据同步和其它通讯端口]:[选举投票通讯]
server.x=[hostname]:nnnnn[:nnnnn]
# 对机器分组和权重设置
group.x=nnnnn[:nnnnn]weight.x=nnnnn
# Leader选举过程中，打开一次连接的超时时间，默认是5s
cnxTimeout
# 每个节点最大数据量，是默认是1M
jute.maxbuffer

Netty

Netty流程

从功能上，流程可以分为服务启动、建立连接、读取数据、业务处理、发送数据、关闭连接以及关闭服务。整体流程如下所示(图中没有包含关闭的部分)：

处理事件

Netty中Reactor线程和worker线程所处理的事件：

1、Server端NioEventLoop处理的事件：

2、Client端NioEventLoop处理的事件

服务启动

服务启动时，以 example 代码中的 EchoServer 为例，启动的过程以及相应的源码类如下：

1. EchoServer#new NioEventLoopGroup(1)->NioEventLoop#provider.openSelector() : 创建 selector
2. EchoServer#b.bind(PORT).sync->AbstractBootStrap#doBind()->initAndRegister()-> channelFactory.newChannel() / init(channel) : 创建 serverSocketChannel 以及初始化
3. EchoServer#b.bind(PORT).sync->AbstractBootStrap#doBind()->initAndRegister()-> config().group().register(channel) ：从 boss group 中选择一个 NioEventLoop 开始注册 serverSocketChannel
4. EchoServer#b.bind(PORT).sync->AbstractBootStrap#doBind()->initAndRegister()->config().group().register(channel)->AbstractChannel#register0(promise)->AbstractNioChannel#javaChannel().register(eventLoop().unwrappedSelector(), 0, this) : 将 server socket channel 注册到选择的 NioEventLoop 的 selector
5. EchoServer#b.bind(PORT).sync()->AbstractBootStrap#doBind()->doBind0()->AbstractChannel#doBind(localAddress)->NioServerSocketChannel#javaChannel().bind(localAddress, config.getBacklog()) : 绑定地址端口开始启动
6. EchoServer#b.bind(PORT).sync()->AbstractBootStrap#doBind()->doBind0()->AbstractChannel#pipeline.fireChannelActive()->AbstractNioChannel#selectionKey.interestOps(interestOps|readInterestOp): 注册 OP_READ 事件

上述启动流程中，1、2、3 由我们自己的线程执行，即mainThread，4、5、6 是由Boss Thread执行。相应时序图如下：

建立连接

服务启动后便是建立连接的过程了，相应过程及源码类如下：

1. NioEventLoop#run()->processSelectedKey() NioEventLoop 中的 selector 轮询创建连接事件（OP_ACCEPT）
2. NioEventLoop#run()->processSelectedKey()->AbstractNioMessageChannel#read->NioServerSocketChannel#doReadMessages()->SocketUtil#accept(serverSocketChannel) 创建 socket channel
3. NioEventLoop#run()->processSelectedKey()->AbstractNioMessageChannel#fireChannelRead->ServerBootstrap#ServerBootstrapAcceptor#channelRead-> childGroup.register(child) 从worker group 中选择一个 NioEventLoop 开始注册 socket channel
4. NioEventLoop#run()->processSelectedKey()->AbstractNioMessageChannel#fireChannelRead->ServerBootstrap#ServerBootstrapAcceptor#channelRead-> childGroup.register(child)->AbstractChannel#register0(promise)-> AbstractNioChannel#javaChannel().register(eventLoop().unwrappedSelector(), 0, this) 将 socket channel 注册到选择的 NioEventLoop 的 selector
5. NioEventLoop#run()->processSelectedKey()->AbstractNioMessageChannel#fireChannelRead->ServerBootstrap#ServerBootstrapAcceptor#channelRead-> childGroup.register(child)->AbstractChannel#pipeline.fireChannelActive()-> AbstractNioChannel#selectionKey.interestOps(interestOps | readInterestOp) 注册 OP_ACCEPT 事件

同样，上述流程中 1、2、3 的执行仍由 Boss Thread 执行，直到 4、5 由具体的 Work Thread 执行。

读写与业务处理

连接建立完毕后是具体的读写，以及业务处理逻辑。以 EchoServerHandler 为例，读取数据后会将数据传播出去供业务逻辑处理，此时的 EchoServerHandler 代表我们的业务逻辑，而它的实现也非常简单，就是直接将数据写回去。我们将这块看成一个整条，流程如下：

1. NioEventLoop#run()->processSelectedKey() NioEventLoop 中的 selector 轮询创建读取事件（OP_READ）
2. NioEventLoop#run()->processSelectedKey()->AbstractNioByteChannel#read()nioSocketChannel 开始读取数据
3. NioEventLoop#run()->processSelectedKey()->AbstractNioByteChannel#read()->pipeline.fireChannelRead(byteBuf)把读取到的数据传播出去供业务处理
4. AbstractNioByteChannel#pipeline.fireChannelRead->EchoServerHandler#channelRead在这个例子中即 EchoServerHandler 的执行
5. EchoServerHandler#write->ChannelOutboundBuffer#addMessage 调用 write 方法
6. EchoServerHandler#flush->ChannelOutboundBuffer#addFlush 调用 flush 准备数据
7. EchoServerHandler#flush->NioSocketChannel#doWrite 调用 flush 发送数据

在这个过程中读写数据都是由 Work Thread 执行的，但是业务处理可以由我们自定义的线程池来处理，并且一般我们也是这么做的，默认没有指定线程的情况下仍然由 Work Thread 代为处理。

关闭连接

服务处理完毕后，单个连接的关闭是什么样的呢？

1. NioEventLoop#run()->processSelectedKey() NioEventLoop 中的 selector 轮询创建读取事件（OP_READ)，这里关闭连接仍然是读取事件
2. NioEventLoop#run()->processSelectedKey()->AbstractNioByteChannel#read()->closeOnRead(pipeline)当字节<0 时开始执行关闭 nioSocketChannel
3. NioEventLoop#run()->processSelectedKey()->AbstractNioByteChannel#read()->closeOnRead(pipeline)->AbstractChannel#close->AbstractNioChannel#doClose() 关闭 socketChannel
4. NioEventLoop#run()->processSelectedKey()->AbstractNioByteChannel#read()->closeOnRead(pipeline)->AbstractChannel#close->outboundBuffer.failFlushed/close 清理消息：不接受新信息，fail 掉所有 queue 中消息
5. NioEventLoop#run()->processSelectedKey()->AbstractNioByteChannel#read()->closeOnRead(pipeline)->AbstractChannel#close->fireChannelInactiveAndDeregister->AbstractNioChannel#doDeregister eventLoop().cancel(selectionKey()) 关闭多路复用器的 key

时序图如下：

关闭服务

最后是关闭整个 Netty 服务：

1. NioEventLoop#run->closeAll()->selectionKey.cancel/channel.close 关闭 channel，取消 selectionKey
2. NioEventLoop#run->confirmShutdown->cancelScheduledTasks 取消定时任务
3. NioEventLoop#cleanup->selector.close() 关闭 selector

时序图如下，为了好画将 NioEventLoop 拆成了 2 块：

长连接优化

更多连接

更高QPS

线程模型

Netty通过Reactor模型基于多路复用器接收并处理用户请求，内部实现了两个线程池，boss线程池和work线程池，其中boss线程池的线程负责处理请求的accept事件，当接收到accept事件的请求时，把对应的socket封装到一个NioSocketChannel中，并交给work线程池，其中work线程池负责请求的read和write事件，由对应的Handler处理。

单线程Reactor线程模型

下图演示了单线程reactor线程模型，之所以称之为单线程，还是因为只有一个accpet Thread接受任务，之后转发到reactor线程中进行处理。两个黄色框表示的是Reactor Thread Group，里面有多个Reactor Thread。一个Reactor Thread Group中的Reactor Thread功能都是相同的，例如第一个黄色框中的Reactor Thread都是处理拆分后的任务的第一阶段，第二个黄色框中的Reactor Thread都是处理拆分后的任务的第二步骤。任务具体要怎么拆分，要结合具体场景，下图只是演示作用。一般来说，都是以比较耗时的操作(例如IO)为切分点。

特别的，如果我们在任务处理的过程中，不划分为多个阶段进行处理的话，那么单线程reactor线程模型就退化成了并行工作和模型。事实上，可以认为并行工作者模型，就是单线程reactor线程模型的最简化版本。

多线程Reactor线程模型

所谓多线程reactor线程模型，无非就是有多个accpet线程，如下图中的虚线框中的部分。

混合型Reactor线程模型

混合型reactor线程模型，实际上最能体现reactor线程模型的本质：

• 将任务处理切分成多个阶段进行，每个阶段处理完自己的部分之后，转发到下一个阶段进行处理。不同的阶段之间的执行是异步的，可以认为每个阶段都有一个独立的线程池。
• 不同的类型的任务，有着不同的处理流程，划分时需要划分成不同的阶段。如下图蓝色是一种任务、绿色是另一种任务，两种任务有着不同的执行流程

Netty-Reactor线程模型

图中大致包含了5个步骤，而我们编写的服务端代码中可能并不能完全体现这样的步骤，因为Netty将其中一些步骤的细节隐藏了，笔者将会通过图形分析与源码分析相结合的方式帮助读者理解这五个步骤。这个五个步骤可以按照以下方式简要概括：

• 设置服务端ServerBootStrap启动参数
• 通过ServerBootStrap的bind方法启动服务端，bind方法会在parentGroup中注册NioServerScoketChannel，监听客户端的连接请求
• Client发起连接CONNECT请求，parentGroup中的NioEventLoop不断轮循是否有新的客户端请求，如果有，ACCEPT事件触发
• ACCEPT事件触发后，parentGroup中NioEventLoop会通过NioServerSocketChannel获取到对应的代表客户端的NioSocketChannel，并将其注册到childGroup中
• childGroup中的NioEventLoop不断检测自己管理的NioSocketChannel是否有读写事件准备好，如果有的话，调用对应的ChannelHandler进行处理

HashedWheelTimer

时间轮其实就是一种环形的数据结构，可以想象成时钟，分成很多格子，一个格子代表一段时间。并用一个链表保存在该格子上的计划任务，同时一个指针随着时间一格一格转动，并执行相应格子中的所有到期任务。任务通过时间取模决定放入那个格子。

在网络通信中管理上万的连接，每个连接都有超时任务，如果为每个任务启动一个Timer超时器，那么会占用大量资源。为了解决这个问题，可用Netty工具类HashedWheelTimer。

Netty 的时间轮 HashedWheelTimer 给出了一个粗略的定时器实现，之所以称之为粗略的实现是因为该时间轮并没有严格的准时执行定时任务，而是在每隔一个时间间隔之后的时间节点执行，并执行当前时间节点之前到期的定时任务。

当然具体的定时任务的时间执行精度可以通过调节 HashedWheelTimer 构造方法的时间间隔的大小来进行调节，在大多数网络应用的情况下，由于 IO 延迟的存在，并不会严格要求具体的时间执行精度，所以默认的 100ms 时间间隔可以满足大多数的情况，不需要再花精力去调节该时间精度。

HashedWheelTimer的特点

• 从源码分析可以看出，其实 HashedWheelTimer 的时间精度并不高，误差能够在 100ms 左右，同时如果任务队列中的等待任务数量过多，可能会产生更大的误差
• 但是 HashedWheelTimer 能够处理非常大量的定时任务，且每次定位到要处理任务的候选集合链表只需要 O(1) 的时间，而 Timer 等则需要调整堆，是 O(logN) 的时间复杂度
• HashedWheelTimer 本质上是模拟了时间的轮盘，将大量的任务拆分成了一个个的小任务列表，能够有效节省 CPU 和线程资源

源码解读

public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, 
                        int ticksPerWheel, boolean leakDetection, long maxPendingTimeouts) {
        ......
}

• threadFactory：自定义线程工厂，用于创建线程对象
• tickDuration：间隔多久走到下一槽（相当于时钟走一格）
• unit：定义tickDuration的时间单位
• ticksPerWheel：一圈有多个槽
• leakDetection：是否开启内存泄漏检测
• maxPendingTimeouts：最多待执行的任务个数。0或负数表示无限制

优缺点

• 优点
  • 可以添加、删除、取消定时任务
  • 能高效的处理大批定时任务
• 缺点
  • 对内存要求较高，占用较高的内存
  • 时间精度要求不高

定时任务方案

目前主流的一些定时任务方案：

• Timer
• ScheduledExecutorService
• ThreadPoolTaskScheduler（基于ScheduledExecutorService）
• Netty的schedule（用到了PriorityQueue）
• Netty的HashedWheelTimer（时间轮）
• Kafka的TimingWheel（层级时间轮）

使用案例

// 构造一个 Timer 实例
Timer timer = new HashedWheelTimer();

// 提交一个任务，让它在 5s 后执行
Timeout timeout1 = timer.newTimeout(new TimerTask() {
    @Override
    public void run(Timeout timeout) {
        System.out.println("5s 后执行该任务");
    }
}, 5, TimeUnit.SECONDS);

// 再提交一个任务，让它在 10s 后执行
Timeout timeout2 = timer.newTimeout(new TimerTask() {
    @Override
    public void run(Timeout timeout) {
        System.out.println("10s 后执行该任务");
    }
}, 10, TimeUnit.SECONDS);

// 取消掉那个 5s 后执行的任务
if (!timeout1.isExpired()) {
    timeout1.cancel();
}

// 原来那个 5s 后执行的任务，已经取消了。这里我们反悔了，我们要让这个任务在 3s 后执行
// 我们说过 timeout 持有上、下层的实例，所以下面的 timer 也可以写成 timeout1.timer()
timer.newTimeout(timeout1.task(), 3, TimeUnit.SECONDS);

ByteBuf

工作流程

ByteBuf维护两个不同的索引：读索引(readerIndex) 和 写索引(writerIndex) 。如下图所示:

• ByteBuf 维护了 readerIndex 和 writerIndex 索引
• 当 readerIndex > writerIndex 时，则抛出 IndexOutOfBoundsException
• ByteBuf容量 = writerIndex
• ByteBuf 可读容量 = writerIndex - readerIndex
• readXXX() 和 writeXXX() 方法将会推进其对应的索引。自动推进
• getXXX() 和 setXXX() 方法将对 writerIndex 和 readerIndex 无影响

使用模式

ByteBuf本质是一个由不同的索引分别控制读访问和写访问的字节数组。ByteBuf共有三种模式：堆缓冲区模式(Heap Buffer)、直接缓冲区模式(Direct Buffer) 和 复合缓冲区模式(Composite Buffer)。

堆缓冲区模式(Heap Buffer)

堆缓冲区模式又称为支撑数组(backing array)。将数据存放在JVM的堆空间，通过将数据存储在数组中实现。

• 优点：由于数据存储在Jvm堆中可以快速创建和快速释放，并且提供了数组直接快速访问的方法
• 缺点：每次数据与I/O进行传输时，都需要将数据拷贝到直接缓冲区

public static void heapBuffer() {
    // 创建Java堆缓冲区
    ByteBuf heapBuf = Unpooled.buffer(); 
    if (heapBuf.hasArray()) { // 是数组支撑
        byte[] array = heapBuf.array();
        int offset = heapBuf.arrayOffset() + heapBuf.readerIndex();
        int length = heapBuf.readableBytes();
        handleArray(array, offset, length);
    }
}

直接缓冲区模式(Direct Buffer)

Direct Buffer属于堆外分配的直接内存，不会占用堆的容量。适用于套接字传输过程，避免了数据从内部缓冲区拷贝到直接缓冲区的过程，性能较好。对于涉及大量I/O的数据读写，建议使用Direct Buffer。而对于用于后端的业务消息编解码模块建议使用Heap Buffer。

• 优点: 使用Socket传递数据时性能很好，避免了数据从Jvm堆内存拷贝到直接缓冲区的过程。提高了性能
• 缺点: 相对于堆缓冲区而言，Direct Buffer分配内存空间和释放更为昂贵

public static void directBuffer() {
    ByteBuf directBuf = Unpooled.directBuffer();
    if (!directBuf.hasArray()) {
        int length = directBuf.readableBytes();
        byte[] array = new byte[length];
        directBuf.getBytes(directBuf.readerIndex(), array);
        handleArray(array, 0, length);
    }
}

复合缓冲区模式(Composite Buffer)

Composite Buffer是Netty特有的缓冲区。本质上类似于提供一个或多个ByteBuf的组合视图，可以根据需要添加和删除不同类型的ByteBuf。

• Composite Buffer是一个组合视图。它提供一种访问方式让使用者自由组合多个ByteBuf，避免了拷贝和分配新的缓冲区
• Composite Buffer不支持访问其支撑数组。因此如果要访问，需要先将内容拷贝到堆内存中，再进行访问

下图是将两个ByteBuf：头部 + Body 组合在一起，没有进行任何复制过程。仅仅创建了一个视图：

public static void byteBufComposite() {
    // 复合缓冲区，只是提供一个视图
    CompositeByteBuf messageBuf = Unpooled.compositeBuffer();
    ByteBuf headerBuf = Unpooled.buffer(); // can be backing or direct
    ByteBuf bodyBuf = Unpooled.directBuffer();   // can be backing or direct
    messageBuf.addComponents(headerBuf, bodyBuf);
    messageBuf.removeComponent(0); // remove the header
    for (ByteBuf buf : messageBuf) {
        System.out.println(buf.toString());
    }
}

字节级操作

随机访问索引

ByteBuf的索引与普通的Java字节数组一样。第一个字节的索引是0，最后一个字节索引总是capacity()-1。访问方式如下：

• readXXX()和writeXXX()方法将会推进其对应的索引readerIndex和writerIndex。自动推进
• getXXX()和setXXX()方法用于访问数据，对writerIndex和readerIndex无影响

public static void byteBufRelativeAccess() {
    ByteBuf buffer = Unpooled.buffer(); // get reference form somewhere
    for (int i = 0; i < buffer.capacity(); i++) {
        byte b = buffer.getByte(i); // 不改变readerIndex值
        System.out.println((char) b);
    }
}

顺序访问索引

Netty的ByteBuf同时具有读索引和写索引，但JDK的ByteBuffer只有一个索引，所以JDK需要调用flip()方法在读模式和写模式之间切换。ByteBuf被读索引和写索引划分成3个区域：可丢弃字节区域、可读字节区域 和 可写字节区域 。

可丢弃字节区域

可丢弃字节区域是指：[0，readerIndex)之间的区域。可调用discardReadBytes()方法丢弃已经读过的字节。

• discardReadBytes()效果 ----- 将可读字节区域(CONTENT)[readerIndex, writerIndex)往前移动readerIndex位，同时修改读索引和写索引
• discardReadBytes()方法会移动可读字节区域内容(CONTENT)。如果频繁调用，会有多次数据复制开销，对性能有一定的影响

可读字节区域

可读字节区域是指:[readerIndex, writerIndex)之间的区域。任何名称以read和skip开头的操作方法，都会改变readerIndex索引。

可写字节区域

可写字节区域是指:[writerIndex, capacity)之间的区域。任何名称以write开头的操作方法都将改变writerIndex的值。

索引管理

• markReaderIndex()+resetReaderIndex() ----- markReaderIndex()是先备份当前的readerIndex，resetReaderIndex()则是将刚刚备份的readerIndex恢复回来。常用于dump ByteBuf的内容，又不想影响原来ByteBuf的readerIndex的值
• readerIndex(int) ----- 设置readerIndex为固定的值
• writerIndex(int) ----- 设置writerIndex为固定的值
• clear() ----- 效果是: readerIndex=0, writerIndex(0)。不会清除内存
• 调用clear()比调用discardReadBytes()轻量的多。仅仅重置readerIndex和writerIndex的值，不会拷贝任何内存，开销较小

查找操作(indexOf)

查找ByteBuf指定的值。类似于，String.indexOf("str")操作

• 最简单的方法 —— indexOf()
• 利用ByteProcessor作为参数来查找某个指定的值

public static void byteProcessor() {
    ByteBuf buffer = Unpooled.buffer(); //get reference form somewhere
    // 使用indexOf()方法来查找
    buffer.indexOf(buffer.readerIndex(), buffer.writerIndex(), (byte)8);
    // 使用ByteProcessor查找给定的值
    int index = buffer.forEachByte(ByteProcessor.FIND_CR);
}

派生缓冲——视图

派生缓冲区为ByteBuf提供了一个访问的视图。视图仅仅提供一种访问操作，不做任何拷贝操作。下列方法，都会呈现给使用者一个视图，以供访问:

• duplicate()
• slice()
• slice(int, int)
• Unpooled.unmodifiableBuffer(...)
• Unpooled.wrappedBuffer(...)
• order(ByteOrder)
• readSlice(int)

理解

• 上面的6中方法，都会返回一个新的ByteBuf实例，具有自己的读索引和写索引。但是，其内部存储是与原对象是共享的。这就是视图的概念
• 请注意：如果你修改了这个新的ByteBuf实例的具体内容，那么对应的源实例也会被修改，因为其内部存储是共享的
• 如果需要拷贝现有缓冲区的真实副本，请使用copy()或copy(int, int)方法
• 使用派生缓冲区，避免了复制内存的开销，有效提高程序的性能

public static void byteBufSlice() {
    Charset utf8 = Charset.forName("UTF-8");
    ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);
    ByteBuf sliced = buf.slice(0, 15);
    System.out.println(sliced.toString(utf8));
    buf.setByte(0, (byte)'J');
    assert buf.getByte(0) == sliced.getByte(0); // return true
}

public static void byteBufCopy() {
    Charset utf8 = Charset.forName("UTF-8");
    ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);
    ByteBuf copy = buf.copy(0, 15);
    System.out.println(copy.toString(utf8));
    buf.setByte(0, (byte)'J');
    assert buf.getByte(0) != copy.getByte(0); // return true
}

读/写操作

如上文所提到的，有两种类别的读/写操作:

• get()和set()操作 ----- 从给定的索引开始，并且保持索引不变
• read()和write()操作 ----- 从给定的索引开始，并且根据已经访问过的字节数对索引进行访问
• 下图给出get()操作API，对于set()操作、read()操作和write操作可参考书籍或API

更多操作

下面的两个方法操作字面意思较难理解，给出解释:

• hasArray()：如果ByteBuf由一个字节数组支撑，则返回true。通俗的讲：ByteBuf是堆缓冲区模式，则代表其内部存储是由字节数组支撑的
• array()：如果ByteBuf是由一个字节数组支撑泽返回数组，否则抛出UnsupportedOperationException异常。也就是，ByteBuf是堆缓冲区模式

ByteBuf分配

创建和管理ByteBuf实例的多种方式：按序分配(ByteBufAllocator)、Unpooled缓冲区 和 ByteBufUtil类。

按序分配：ByteBufAllocator接口

Netty通过接口ByteBufAllocator实现了(ByteBuf的)池化。Netty提供池化和非池化的ButeBufAllocator:

• ctx.channel().alloc().buffer()：本质就是ByteBufAllocator.DEFAULT
• ByteBufAllocator.DEFAULT.buffer()：返回一个基于堆或者直接内存存储的Bytebuf。默认是堆内存
• ByteBufAllocator.DEFAULT：有两种类型: UnpooledByteBufAllocator.DEFAULT(非池化)和PooledByteBufAllocator.DEFAULT(池化)。对于Java程序，默认使用PooledByteBufAllocator(池化)。对于安卓，默认使用UnpooledByteBufAllocator(非池化)
• 可以通过BootStrap中的Config为每个Channel提供独立的ByteBufAllocator实例

解释:

• 上图中的buffer()方法，返回一个基于堆或者直接内存存储的Bytebuf ----- 缺省是堆内存。源码: AbstractByteBufAllocator() { this(false); }
• ByteBufAllocator.DEFAULT ----- 可能是池化，也可能是非池化。默认是池化(PooledByteBufAllocator.DEFAULT)

Unpooled缓冲区——非池化

Unpooled提供静态的辅助方法来创建未池化的ByteBuf。

注意:

• 上图的buffer()方法，返回一个未池化的基于堆内存存储的ByteBuf
• wrappedBuffer() ----- 创建一个视图，返回一个包装了给定数据的ByteBuf。非常实用

创建ByteBuf代码:

 public void createByteBuf(ChannelHandlerContext ctx) {
    // 1. 通过Channel创建ByteBuf
    ByteBuf buf1 = ctx.channel().alloc().buffer();
    // 2. 通过ByteBufAllocator.DEFAULT创建
    ByteBuf buf2 =  ByteBufAllocator.DEFAULT.buffer();
    // 3. 通过Unpooled创建
    ByteBuf buf3 = Unpooled.buffer();
}

ByteBufUtil类

ByteBufUtil类提供了用于操作ByteBuf的静态的辅助方法: hexdump()和equals

• hexdump()：以十六进制的表示形式打印ByteBuf的内容。非常有价值
• equals()：判断两个ByteBuf实例的相等性

引用计数

Netty4.0版本中为ButeBuf和ButeBufHolder引入了引用计数技术。请区别引用计数和可达性分析算法(jvm垃圾回收)

• 谁负责释放：一般来说，是由最后访问(引用计数)对象的那一方来负责将它释放
• buffer.release()：引用计数减1
• buffer.retain()：引用计数加1
• buffer.refCnt()：返回当前对象引用计数值
• buffer.touch()：记录当前对象的访问位置，主要用于调试
• 引用计数并非仅对于直接缓冲区(direct Buffer)。ByteBuf的三种模式: 堆缓冲区(heap Buffer)、直接缓冲区(dirrect Buffer)和复合缓冲区(Composite Buffer)都使用了引用计数，某些时候需要程序员手动维护引用数值

public static void releaseReferenceCountedObject(){
    ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
    // 引用计数加1
    buffer.retain();
    // 输出引用计数
    buffer.refCnt();
    // 引用计数减1
    buffer.release();
}

Zero-Copy

Netty 的Zero-copy 体现在如下几个个方面:

• 通过CompositeByteBuf实现零拷贝：Netty提供了CompositeByteBuf 类，可以将多个ByteBuf合并为一个逻辑上的ByteBuf，避免了各个ByteBuf之间的拷贝
• 通过wrap操作实现零拷贝：通过wrap操作，可以将byte[]、ByteBuf、ByteBuffer等包装成一个Netty ByteBuf对象，进而避免了拷贝操作
• 通过slice操作实现零拷贝：ByteBuf支持slice操作，可以将ByteBuf分解为多个共享同一个存储区域的ByteBuf，避免了内存的拷贝
• 通过FileRegion实现零拷贝：通过 FileRegion 包装的FileChannel.tranferTo 实现文件传输，可以直接将文件缓冲区的数据发送到目标 Channel，避免了传统通过循环write方式导致的内存拷贝问题

零拷贝操作

通过CompositeByteBuf实现零拷贝

ByteBuf header = null;
ByteBuf body = null;
    
// 传统合并header和body：两次额外的数据拷贝
ByteBuf allBuf = Unpooled.buffer(header.readableBytes() + body.readableBytes());
allBuf.writeBytes(header);
allBuf.writeBytes(body);
    
// 合并header和body：内部这两个 ByteBuf 都是单独存在的, CompositeByteBuf 只是逻辑上是一个整体
CompositeByteBuf compositeByteBuf = Unpooled.compositeBuffer();
compositeByteBuf.addComponents(true, header, body);
// 底层封装了 CompositeByteBuf 操作
ByteBuf allByteBuf = Unpooled.wrappedBuffer(header, body);

通过wrap操作实现零拷贝

byte[] bytes = null;

// 传统方式：直接将byte[]数组拷贝到ByteBuf中
ByteBuf byteBuf = Unpooled.buffer();
byteBuf.writeBytes(bytes);

// wrap方式：将bytes包装成为一个UnpooledHeapByteBuf对象, 包装过程中, 是不会有拷贝操作的
// 即最后我们生成的生成的ByteBuf对象是和bytes数组共用了同一个存储空间, 对bytes的修改也会反映到ByteBuf对象中
ByteBuf byteBuf = Unpooled.wrappedBuffer(bytes);

通过slice操作实现零拷贝

slice 操作和 wrap 操作刚好相反,Unpooled.wrappedBuffer 可以将多个 ByteBuf 合并为一个, 而 slice 操作可以将一个 **ByteBuf **切片 为多个共享一个存储区域的 ByteBuf 对象. ByteBuf 提供了两个 slice 操作方法:

public ByteBuf slice();
public ByteBuf slice(int index, int length);

不带参数的slice方法等同于buf.slice(buf.readerIndex(), buf.readableBytes()) 调用, 即返回 buf 中可读部分的切片. 而 slice(int index, int length)方法相对就比较灵活了, 我们可以设置不同的参数来获取到 buf 的不同区域的切片。

用 slice 方法产生 header 和 body 的过程是没有拷贝操作的, header 和 body 对象在内部其实是共享了 byteBuf 存储空间的不同部分而已，即：

通过FileRegion实现零拷贝

Netty 中使用 FileRegion 实现文件传输的零拷贝, 不过在底层 FileRegion 是依赖于 Java NIO FileChannel.transfer 的零拷贝功能。当有了 FileRegion 后, 我们就可以直接通过它将文件的内容直接写入 Channel 中, 而不需要像传统的做法: 拷贝文件内容到临时 buffer, 然后再将 buffer 写入 Channel. 通过这样的零拷贝操作, 无疑对传输大文件很有帮助。

传统IO的流程

• 「第一步」：将文件通过 「DMA」 技术从磁盘中拷贝到内核缓冲区
• 「第二步」：将文件从内核缓冲区拷贝到用户进程缓冲区域中
• 「第三步」：将文件从用户进程缓冲区中拷贝到 socket 缓冲区中
• 「第四步」：将socket缓冲区中的文件通过 「DMA」 技术拷贝到网卡

零拷贝整体流程图

TCP粘包拆包

粘包拆包图解

假设客户端分别发送了两个数据包D1和D2给服务端，由于服务端一次读取到字节数是不确定的，故可能存在以下几种情况：

• 服务端分两次读取到两个独立的数据包，分别是D1和D2，没有粘包和拆包
• 服务端一次接收到了两个数据包，D1和D2粘在一起，发生粘包
• 服务端分两次读取到数据包，第一次读取到了完整D1包和D2包的部分内容，第二次读取到了D2包的剩余内容，发生拆包
• 服务端分两次读取到数据包，第一次读取到部分D1包，第二次读取到剩余的D1包和全部的D2包
• 当TCP缓存再小一点的话，会把D1和D2分别拆成多个包发送

产生原因

产生粘包和拆包问题的主要原因是，操作系统在发送TCP数据的时候，底层会有一个缓冲区，例如1024个字节大小：

• 如果一次请求发送的数据量比较小，没达到缓冲区大小，TCP则会将多个请求合并为同一个请求进行发送，这就形成了粘包问题
• 如果一次请求发送的数据量比较大，超过了缓冲区大小，TCP就会将其拆分为多次发送，这就是拆包，也就是将一个大的包拆分为多个小包进行发送

解决方案

固定长度

对于使用固定长度的粘包和拆包场景，可以使用：

• FixedLengthFrameDecoder：每次读取固定长度的消息，如果当前读取到的消息不足指定长度，那么就会等待下一个消息到达后进行补足。其使用也比较简单，只需要在构造函数中指定每个消息的长度即可。

 @Override
protected void initChannel(SocketChannel ch) throws Exception {
          // 这里将FixedLengthFrameDecoder添加到pipeline中，指定长度为20
          ch.pipeline().addLast(new FixedLengthFrameDecoder(20));
          // 将前一步解码得到的数据转码为字符串
          ch.pipeline().addLast(new StringDecoder());
          // 这里FixedLengthFrameEncoder是我们自定义的，用于将长度不足20的消息进行补全空格
          ch.pipeline().addLast(new FixedLengthFrameEncoder(20));
          // 最终的数据处理
          ch.pipeline().addLast(new EchoServerHandler());
}

指定分隔符

对于通过分隔符进行粘包和拆包问题的处理，Netty提供了两个编解码的类：

• LineBasedFrameDecoder：通过换行符，即\n或者\r\n对数据进行处理
• DelimiterBasedFrameDecoder：通过用户指定的分隔符对数据进行粘包和拆包处理

@Override
protected void initChannel(SocketChannel ch) throws Exception {
        String delimiter = "_$";
        // 将delimiter设置到DelimiterBasedFrameDecoder中，经过该解码一器进行处理之后，源数据将会
        // 被按照_$进行分隔，这里1024指的是分隔的最大长度，即当读取到1024个字节的数据之后，若还是未
        // 读取到分隔符，则舍弃当前数据段，因为其很有可能是由于码流紊乱造成的
        ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,
            Unpooled.wrappedBuffer(delimiter.getBytes())));
        // 将分隔之后的字节数据转换为字符串数据
        ch.pipeline().addLast(new StringDecoder());
        // 这是我们自定义的一个编码器，主要作用是在返回的响应数据最后添加分隔符
        ch.pipeline().addLast(new DelimiterBasedFrameEncoder(delimiter));
        // 最终处理数据并且返回响应的handler
        ch.pipeline().addLast(new EchoServerHandler());
}

数据包长度字段

处理粘拆包的主要思想是在生成的数据包中添加一个长度字段，用于记录当前数据包的长度。

• LengthFieldBasedFrameDecoder：按照参数指定的包长度偏移量数据对接收到的数据进行解码，从而得到目标消息体数据。解码过程如下图所示：

  

• LengthFieldPrepender：在响应的数据前面添加指定的字节数据，这个字节数据中保存了当前消息体的整体字节数据长度。编码过程如下图所示：

  

@Override
protected void initChannel(SocketChannel ch) throws Exception {
            // 这里将LengthFieldBasedFrameDecoder添加到pipeline的首位，因为其需要对接收到的数据
            // 进行长度字段解码，这里也会对数据进行粘包和拆包处理
            ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 2, 0, 2));
            // LengthFieldPrepender是一个编码器，主要是在响应字节数据前面添加字节长度字段
            ch.pipeline().addLast(new LengthFieldPrepender(2));
            // 对经过粘包和拆包处理之后的数据进行json反序列化，从而得到User对象
            ch.pipeline().addLast(new JsonDecoder());
            // 对响应数据进行编码，主要是将User对象序列化为json
            ch.pipeline().addLast(new JsonEncoder());
            // 处理客户端的请求的数据，并且进行响应
            ch.pipeline().addLast(new EchoServerHandler());
 }

自定义粘包拆包器

可以通过实现MessageToByteEncoder和ByteToMessageDecoder来实现自定义粘包和拆包处理的目的。

• MessageToByteEncoder：作用是将响应数据编码为一个ByteBuf对象
• ByteToMessageDecoder：将接收到的ByteBuf数据转换为某个对象数据

高性能

• IO线程模型：同步非阻塞，用最少的资源做更多的事
• 内存零拷贝：尽量减少不必要的内存拷贝，实现了更高效率的传输
• 内存池设计：申请的内存可以重用，主要指直接内存。内部实现是用一颗二叉查找树管理内存分配情况
• 串形化处理读写：避免使用锁带来的性能开销
• 高性能序列化协议：支持 protobuf 等高性能序列化协议

操作系统调优

文件描述符

• 设置系统最大文件句柄数

# 查看
cat /proc/sys/fs/file-max
# 修改
在/etc/sysctl.conf插入fs.file-max=1000000
# 配置生效
sysctl -p

• 设置单进程打开的最大句柄数

默认单进程打开的最大句柄数是 1024，通过 ulimit -a 可以查看相关参数，示例如下：

[root@test ~]# ulimit -a
core file size          (blocks, -c) 0
data seg size           (kbytes, -d) unlimited
scheduling priority             (-e) 0
file size               (blocks, -f) unlimited
pending signals                 (-i) 256324
max locked memory       (kbytes, -l) 64
max memory size         (kbytes, -m) unlimited
open files                      (-n) 1024
......

当并发接入的TCP连接数超过上限时，就会报“too many open files”，所有新的客户端接入将失败。通过 vi /etc/security/limits.conf 命令添加如下配置参数：

*  soft  nofile  1000000
*  hard  nofile  1000000

修改之后保存，注销当前用户，重新登录，通过 ulimit -a 查看修改的状态是否生效。

注意：尽管我们可以将单个进程打开的最大句柄数修改的非常大，但是当句柄数达到一定数量级之后，处理效率将出现明显下降，因此，需要根据服务器的硬件配置和处理能力进行合理设置。

TCP/IP相关参数

需要重点调优的TCP IP参数如下：

• net.ipv4.tcp_rmem：为每个TCP连接分配的读缓冲区内存大小。第一个值时socket接收缓冲区分配的最小字节数。

多网卡队列和软中断

• TCP缓冲区

  根据推送消息的大小，合理设置以下两个参数，对于海量长连接，通常 32K 是个不错的选择：

  • SO_SNDBUF：发送缓冲区大小
  • SO_RCVBUF：接收缓冲区大小

• 软中断

  使用命令 cat /proc/interrupts 查看网卡硬件中断的运行情况，如果全部被集中在CPU0上处理，则无法并行执行多个软中断。Linux kernel内核≥2.6.35的版本，可以开启RPS，网络通信能力提升20%以上，RPS原理是：根据数据包的源地址、目的地址和源端口等，计算出一个Hash值，然后根据Hash值来选择软中断运行的CPU，即实现每个链接和CPU绑定，通过Hash值来均衡软中断运行在多个CPU上。

Netty性能调优

设置合理的线程数

boss线程池优化

对于Netty服务端，通常只需要启动一个监听端口用于端侧设备接入，但是如果集群实例较少，甚至是单机部署，那么在短时间内大量设备接入时，需要对服务端的监听方式和线程模型做优化，即服务端监听多个端口，利用主从Reactor线程模型。由于同时监听了多个端口，每个ServerSocketChannel都对应一个独立的Acceptor线程，这样就能并行处理，加速端侧设备的接人速度，减少端侧设备的连接超时失败率，提高单节点服务端的处理性能。

work线程池优化(I/O工作线程池)

对于I/O工作线程池的优化，可以先采用系统默认值（cpu内核数*2）进行性能测试，在性能测试过程中采集I/O线程的CPU占用大小，看是否存在瓶颈，具体策略如下：

• 通过执行 ps -ef|grep java 找到服务端进程pid
• 执行top -Hp pid查询该进程下所有线程的运行情况，通过“shift+p”对CPU占用大小做排序，获取线程的pid及对应的CPU占用大小
• 使用printf'%x\n' pid将pid转换成16进制格式
• 通过jstack -f pid命令获取线程堆栈，或者通过jvisualvm工具打印线程堆栈，找到I/O work工作线程，查看他们的CPU占用大小及线程堆栈，关键词：SelectorImpl.lockAndDoSelect

分析

• 如果连续采集几次进行对比，发现线程堆栈都停留在SelectorImpl.lockAndDoSelect处，则说明I/O线程比较空闲，无需对工作线程数做调整
• 如果发现I/O线程的热点停留在读或写操作，或停留在ChannelHandler的执行处，则可以通过适当调大NioEventLoop线程的个数来提升网络的读写性能。调整方式有两种：
  • 接口API指定：在创建NioEventLoopGroup实例时指定线程数
  • 系统参数指定：通过-Dio.netty.eventLoopThreads来指定NioEventLoopGroup线程池（不建议）

心跳检测优化

心跳检测的目的就是确认当前链路是否可用，对方是否活着并且能够正常接收和发送消息。从技术层面看，要解决链路的可靠性问题，必须周期性地对链路进行有效性检测。目前最流行和通用的做法就是心跳检测。

海量设备接入的服务端心跳优化策略

• 要能够及时检测失效的连接，并将其剔除。防止无效的连接句柄积压，导致OOM等问题
• 设置合理的心跳周期。防止心跳定时任务积压，造成频繁的老年代GC（新生代和老年代都有导致STW的GC，不过耗时差异较大），导致应用暂停
• 使用Nety提供的链路空闲检测机制。不要自己创建定时任务线程池，加重系统的负担，以及增加潜在的并发安全问题

心跳检测机制分为三个层面

• TCP层的心跳检测：即TCP的 Keep-Alive机制，它的作用域是整个TCP协议栈
• 协议层的心跳检测：主要存在于长连接协议中，例如MQTT
• 应用层的心跳检测：它主要由各业务产品通过约定方式定时给对方发送心跳消息实现

心跳检测机制分类

• Ping-Pong型心跳：由通信一方定时发送Ping消息，对方接收到Ping消息后立即返回Pong答应消息给对方，属于“请求-响应型”心跳
• Ping-Ping型心跳：不区分心跳请求和答应，由通信双发按照约定时间向对方发送心跳Ping消息，属于”双向心跳“

心跳检测机制策略

• 心跳超时：连续N次检测都没有收到对方的Pong应答消息或Ping请求消息，则认为链路已经发生逻辑失效
• 心跳失败：在读取和发送心跳消息的时候，如果直接发生了IO异常，说明链路已经失效

链路空闲检测机制

• 读空闲：链路持续时间T没有读取到任何消息
• 写空闲：链路持续时间T没有发送任何消息
• 读写空闲：链路持续时间T没有接收或者发送任何消息

案例分析

由于移动无线网络的特点，推送服务的心跳周期并不能设置的太长，否则长连接会被释放，造成频繁的客户端重连，但是也不能设置太短，否则在当前缺乏统一心跳框架的机制下很容易导致信令风暴（例如微信心跳信令风暴问题）。具体的心跳周期并没有统一的标准，180S 也许是个不错的选择，微信为 300S。

在 Netty 中，可以通过在 ChannelPipeline 中增加 IdleStateHandler 的方式实现心跳检测，在构造函数中指定链路空闲时间，然后实现空闲回调接口，实现心跳的发送和检测。拦截链路空闲事件并处理心跳：

public class MyHandler extends ChannelHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            // 心跳处理
        }
    }
 }

心跳优化结论

• 对于百万级的服务器，一般不建议很长的心跳周期和超时时长
• 心跳检测周期通常不要超过60s，心跳检测超时通常为心跳检测周期的2倍
• 建议通过IdleStateHandler实现心跳，不要自己创建定时任务线程池，加重系统负担和增加潜在并发安全问题
• 发生心跳超时或心跳失败时，都需要关闭链路，由客户端发起重连操作，保证链路能够恢复正常
• 链路空闲事件被触发后并没有关闭链路，而是触发IdleStateEvent事件，用户订阅IdleStateEvent事件，用于自定义逻辑处理。如关闭链路、客户端发起重新连接、告警和日志打印等
• 链路空闲检测类库主要包括：IdleStateHandler、ReadTimeoutHandler、WriteTimeoutHandler

接收和发送缓冲区调优

对于长链接，每个链路都需要维护自己的消息接收和发送缓冲区，JDK 原生的 NIO 类库使用的是java.nio.ByteBuffer, 它实际是一个长度固定的byte[]，无法动态扩容。

场景：假设单条消息最大上限为10K，平均大小为5K，为满足10K消息处理，ByteBuffer的容量被设置为10K，这样每条链路实际上多消耗了5K内存，如果长链接链路数为100万，每个链路都独立持有ByteBuffer接收缓冲区，则额外损耗的总内存Total(M) =1000000×5K=4882M

Netty提供的ByteBuf支持容量动态调整，同时提供了两种接收缓冲区的内存分配器：

• FixedRecvByteBufAllocator：固定长度的接收缓冲区分配器，它分配的ByteBuf长度是固定大小的，并不会根据实际数据报大小动态收缩。但如果容量不足，支持动态扩展
• AdaptiveRecvByteBufAllocator：容量动态调整的接收缓冲区分配器，会根据之前Channel接收到的数据报大小进行计算，如果连续填充满接收缓冲区的可写空间，则动态扩展容量。如果连续2次接收到的数据报都小于指定值，则收缩当前的容量，以节约内存

相对于FixedRecvByteBufAllocator，使用AdaptiveRecvByteBufAllocator更为合理，可在创建客户端或者服务端的时候指定RecvByteBufAllocator。

Bootstrap b = new Bootstrap();
           b.group(group)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)

注意：无论是接收缓冲区还是发送缓冲区，缓冲区的大小建议设置为消息的平均大小，不要设置成最大消息的上限，这会导致额外的内存浪费。

合理使用内存池

每个NioEventLoop线程处理N个链路。链路处理流程：开始处理A链路→创建接收缓冲区(创建ByteBuf)→消息解码→封装成POJO对象→提交至后台线程成Task→释放接收缓冲区→开始处理B链路。

如果使用内存池，则当A链路接收到新数据报后，从NioEventLoop的内存池中申请空闲的ByteBuf，解码完成后，调用release将ByteBuf释放到内存池中，供后续B链路继续使用。使用内存池优化后，单个NioEventLoop的ByteBuf申请和GC次数从原来的N=1000000/64= 15625次减少为最少0次（假设每次申请都有可用的内存）。

Netty默认不使用内存池，需要在创建客户端或者服务端的时候进行指定：

Bootstrap b = new Bootstrap();
           b.group(group)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)

使用内存池之后，内存的申请和释放必须成对出现，即 retain() 和 release() 要成对出现，否则会导致内存泄露。值得注意的是，如果使用内存池，完成 ByteBuf 的解码工作之后必须显式的调用 ReferenceCountUtil.release(msg) 对接收缓冲区 ByteBuf 进行内存释放，否则它会被认为仍然在使用中，这样会导致内存泄露。

防止I/O线程被意外阻塞

通常情况不能在Netty的I/O线程上做执行时间不可控的操作，如访问数据库、调用第三方服务等，但有一些隐形的阻塞操作却容易被忽略，如打印日志。

生产环境中，一般需要实时打印接口日志，其它日志处于ERROR级别，当服务发生I/O异常后，会记录异常日志。如果磁盘的WIO比较高，可能会发生写日志文件操作被同步阻塞，阻塞时间无法预测，就会导致Netty的NioEventLoop线程被阻塞，Socket链路无法被及时管理，其它链路也无法进行读写操作等。

常用的log4j虽然支持异步写日志(AsyncAppender)，但当日志队列满之后，它会同步阻塞业务线程，直到日志队列有空闲位置可用。

类似问题具有极强的隐蔽性，往往WIO高的时间持续非常短，或是偶现的，在测试环境中很难模拟此类故障，问题定位难度大。

I/O线程与业务线程分离

• 如果服务端不做复杂业务逻辑操作，仅是简单内存操作和消息转发，则可通过调大NioEventLoop工作线程池的方式，直接在I/O线程中执行业务ChannelHandler，这样便减少了一次线上上下文切换，性能反而更高
• 如果有复杂的业务逻辑操作，则建议I/O线程和业务线程分离。
  • 对于I/O线程，由于互相之间不存在锁竞争，可以创建一个大的NioEventLoopGroup线程组，所有Channel都共享一个线程池
  • 对于后端的业务线程池，则建议创建多个小的业务线程池，线程池可以与I/O线程绑定，这样既减少了锁竞争，又提升了后端的处理性能

服务端并发连接数流控

无论服务端的性能优化到多少，都需要考虑流控功能。当资源成为瓶颈，或遇到端侧设备的大量接入，需要通过流控对系统做保护。一般Netty主要考虑并发连接数的控制。

案例分析

疑似内存泄漏

环境：8C16G的Linux

描述：boss为1，worker为6，其余分配给业务使用，保持10W用户长链接，2W用户并发做消息请求

分析：dump内存堆栈发现Netty的ScheduledFutureTask增加了9076%，达到110W个实例。通过业务代码分析发现用户使用了IdleStateHandler用于在链路空闲时进行业务逻辑处理，但空闲时间比较大，为15分钟。Netty 的 IdleStateHandler 会根据用户的使用场景，启动三类定时任务，分别是：ReaderIdleTimeoutTask、WriterIdleTimeoutTask 和 AllIdleTimeoutTask，它们都会被加入到 NioEventLoop 的 Task 队列中被调度和执行。由于超时时间过长，10W 个长链接链路会创建 10W 个 ScheduledFutureTask 对象，每个对象还保存有业务的成员变量，非常消耗内存。用户的持久代设置的比较大，一些定时任务被老化到持久代中，没有被 JVM 垃圾回收掉，内存一直在增长，用户误认为存在内存泄露，即小问题被放大而引出的问题。

解决：重新设计和反复压测之后将超时时间设置为45秒，内存可以实现正常回收。

当心CLOSE_WAIT

由于网络不稳定经常会导致客户端断连，如果服务端没有能够及时关闭 socket，就会导致处于 close_wait 状态的链路过多。close_wait 状态的链路并不释放句柄和内存等资源，如果积压过多可能会导致系统句柄耗尽，发生“Too many open files”异常，新的客户端无法接入，涉及创建或者打开句柄的操作都将失败。

close_wait 是被动关闭连接是形成的，根据 TCP 状态机，服务器端收到客户端发送的 FIN，TCP 协议栈会自动发送 ACK，链接进入 close_wait 状态。但如果服务器端不执行 socket 的 close() 操作，状态就不能由 close_wait 迁移到 last_ack，则系统中会存在很多 close_wait 状态的连接。通常来说，一个 close_wait 会维持至少 2 个小时的时间（系统默认超时时间的是 7200 秒，也就是 2 小时）。如果服务端程序因某个原因导致系统造成一堆 close_wait 消耗资源，那么通常是等不到释放那一刻，系统就已崩溃。

导致 close_wait 过多的可能原因如下：

• 程序处理Bug：导致接收到对方的 fin 之后没有及时关闭 socket，这可能是 Netty 的 Bug，也可能是业务层 Bug，需要具体问题具体分析
• 关闭socket不及时：例如 I/O 线程被意外阻塞，或者 I/O 线程执行的用户自定义 Task 比例过高，导致 I/O 操作处理不及时，链路不能被及时释放

解决方案

• 不要在 Netty 的 I/O 线程（worker线程）上处理业务（心跳发送和检测除外）
• 在I/O线程上执行自定义Task要当心
• IdleStateHandler、ReadTimeoutHandler和WriteTimeoutHandler使用要当

RabbitMQ

模式介绍

在 RabbitMQ 官网上提供了 6 中工作模式：简单模式、工作队列模式、发布/订阅模式、路由模式、主题模式 和 RPC 模式。本篇只对前 5 种工作方式进行介绍。

简单模式与工作队列模式

之所以将这两种模式合并在一起介绍，是因为它们工作原理非常简单，由 3 个对象组成：生产者、队列、消费者。

生产者负责生产消息，将消息发送到队列中，消费者监听队列，队列有消息就进行消费。

当有多个消费者时，消费者平均消费队列中的消息。代码演示：

生产者：

//1.获取连接
Connection connection = ConnectionUtil.getConnection();
//2.创建通道
Channel channel = connection.createChannel();
//3.申明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//4.发送消息
channel.basicPublish("", QUEUE_NAME, null, "hello simple".getBytes());

System.out.println("发送成功");
//5.释放连接
channel.close();
connection.close();

消费者：

// 1.获取连接
Connection connection = ConnectionUtil.getConnection();
// 2.创建通道
Channel channel = connection.createChannel();
// 3.申明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 4.监听消息
channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
            byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
        System.out.println("接收:" + message);
    }
});

发布/订阅、路由与主题模式

这 3 种模式都使用到交换机。生产者不直接与队列交互，而是将消息发送到交换机中，再由交换机将消息放入到已绑定该交换机的队列中给消费者消费。常用的交换机类型有 3 种：fanout、direct、topic。工作原理图如下：

fanout：不处理路由键。只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout 类型交换机转发消息是最快的。

其中，发布/订阅模式使用的是 fanout 类型的交换机。

direct：处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。如果一个队列绑定到该交换机上要求路由键 “dog”，则只有被标记为 “dog” 的消息才被转发，不会转发 dog.puppy，也不会转发 dog.guard，只会转发dog。

其中，路由模式使用的是 direct 类型的交换机。

topic：将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号 “#” 匹配一个或多个词，符号“*”匹配不多不少一个词。因此“audit.#” 能够匹配到“audit.irs.corporate”，但是“audit.*” 只会匹配到 “audit.irs”。

**其中，主题模式使用的是 topic 类型的交换机。**代码演示：

生产者：

// 1.获取连接
Connection connection = ConnectionUtil.getConnection();
// 2.创建通道
Channel channel = connection.createChannel();
// 3.申明交换机
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 4.发送消息
for (int i = 0; i < 100; i++) {
     channel.basicPublish(EXCHANGE_NAME, "", null, ("hello ps" + i + "").getBytes());
}

System.out.println("发送成功");
// 5.释放连接
channel.close();
connection.close();

多个消费者：

// 1.获取连接
Connection connection = ConnectionUtil.getConnection();
// 2.创建通道
Channel channel = connection.createChannel();
// 3.申明交换机
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 4.队列绑定交换机
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
// 5.消费消息
channel.basicQos(1);
channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
        System.out.println("recv1:" + message);

        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        channel.basicAck(envelope.getDeliveryTag(), false);
    }
});

Dubbo

Apache Dubbo 是一款高性能、轻量级的开源 Java 服务框架。Apache Dubbo 提供了六大核心能力：面向接口代理的高性能RPC调用，智能容错和负载均衡，服务自动注册和发现，高度可扩展能力，运行期流量调度，可视化的服务治理与运维。

• 面向接口代理的高性能RPC调用

  提供高性能的基于代理的远程调用能力，服务以接口为粒度，为开发者屏蔽远程调用底层细节。

• 智能负载均衡

  内置多种负载均衡策略，智能感知下游节点健康状况，显著减少调用延迟，提高系统吞吐量。

• 服务自动注册与发现

  支持多种注册中心服务，服务实例上下线实时感知。

• 高度可扩展能力

  遵循微内核+插件的设计原则，所有核心能力如Protocol、Transport、Serialization被设计为扩展点，平等对待内置实现和第三方实现。

• 运行期流量调度

  内置条件、脚本等路由策略，通过配置不同的路由规则，轻松实现灰度发布，同机房优先等功能。

• 可视化的服务治理与运维

  提供丰富服务治理、运维工具：随时查询服务元数据、服务健康状态及调用统计，实时下发路由策略、调整配置参数。

RPC

Remote Procedure Call Protocol 既 远程过程调用，一种能让我们像调用本地服务一样调用远程服务，可以让调用者对网络通信这些细节无感知，比如服务消费方在执行 helloWorldService.sayHello("sowhat") 时，实质上调用的是远端的服务。这种方式其实就是RPC，RPC思想在各大互联网公司中被广泛使用，如阿里巴巴的dubbo、当当的Dubbox 、Facebook 的 thrift、Google 的grpc、Twitter的finagle等。

框架设计

Dubbo 简介

Dubbo 是阿里巴巴研发开源工具，主要分为2.6.x 跟 2.7.x 版本。是一款分布式、高性能、透明化的 RPC 服务框架，提供服务自动注册、自动发现等高效服务治理方案，可以和Spring 框架无缝集成，它提供了6大核心能力：

• 面向接口代理的高性能RPC调用
• 智能容错和负载均衡
• 服务自动注册和发现
• 高度可扩展能力
• 运行期流量调度
• 可视化的服务治理与运维

调用过程：

• 服务提供者 Provider 启动然后向 Registry 注册自己所能提供的服务。
• 服务消费者 Consumer 向Registry订阅所需服务，Consumer 解析Registry提供的元信息，从服务中通过负载均衡选择 Provider调用。
• 服务提供方 Provider 元数据变更的话Registry会把变更推送给Consumer，以此保证Consumer获得最新可用信息。

注意点：

• Provider 跟 Consumer 在内存中记录调用次数跟时间，定时发送统计数据到Monitor，发送的时候是短连接。
• Monitor 跟 Registry 是可选的，可直接在配置文件中写好，Provider 跟 Consumer进行直连。
• Monitor 跟 Registry 挂了也没事， Consumer 本地缓存了 Provider 信息。
• Consumer 直接调用 Provider 不会经过 Registry。Provider、Consumer这俩到 Registry之间是长连接。

Dubbo框架分层

如上图，总的而言 Dubbo 分为三层。

• Busines层：由用户自己来提供接口和实现还有一些配置信息。
• RPC层：真正的RPC调用的核心层，封装整个RPC的调用过程、负载均衡、集群容错、代理。
• Remoting层：对网络传输协议和数据转换的封装。

如果每一层再细分下去，一共有十层：

1. 接口服务层（Service）：该层与业务逻辑相关，根据 provider 和 consumer 的业务设计对应的接口和实现。
2. 配置层（Config）：对外配置接口，以 ServiceConfig 和 ReferenceConfig 为中心初始化配置。
3. 服务代理层（Proxy）：服务接口透明代理，Provider跟Consumer都生成代理类，使得服务接口透明，代理层实现服务调用跟结果返回。
4. 服务注册层（Registry）：封装服务地址的注册和发现，以服务 URL 为中心。
5. 路由层（Cluster）：封装多个提供者的路由和负载均衡，并桥接注册中心，以Invoker 为中心，扩展接口为 Cluster、Directory、Router 和 LoadBlancce。
6. 监控层（Monitor）：RPC 调用次数和调用时间监控，以 Statistics 为中心，扩展接口为 MonitorFactory、Monitor 和 MonitorService。
7. 远程调用层（Protocal）：封装 RPC 调用，以 Invocation 和 Result 为中心，扩展接口为 Protocal、Invoker 和 Exporter。
8. 信息交换层（Exchange）：封装请求响应模式，同步转异步。以 Request 和Response 为中心，扩展接口为 Exchanger、ExchangeChannel、ExchangeClient 和 ExchangeServer。
9. 网络传输层（Transport）：抽象 mina 和 netty 为统一接口，以 Message 为中心，扩展接口为 Channel、Transporter、Client、Server 和 Codec。
10. 数据序列化层（Serialize）：可复用的一些工具，扩展接口为 Serialization、ObjectInput、ObjectOutput 和 ThreadPool。

他们之间的调用关系直接看下面官网图即可：

SPI机制

Dubbo 采用 微内核设计 + SPI 扩展技术来搭好核心框架，同时满足用户定制化需求。这里重点说下SPI。

SPI 含义

SPI 全称为 Service Provider Interface，是一种服务发现机制。它约定在ClassPath路径下的META-INF/services文件夹查找文件，自动加载文件里所定义的类。

Java SPI缺点

• 不能按需加载，Java SPI在加载扩展点的时候，会一次性加载所有可用的扩展点，很多是不需要的，会浪费系统资源。
• 获取某个实现类的方式不够灵活，只能通过 Iterator 形式获取，不能根据某个参数来获取对应的实现类。
• 不支持AOP与依赖注入，JAVA SPI可能会丢失加载扩展点异常信息，导致追踪问题很困难。

Dubbo SPI

JDK自带的不好用Dubbo 就自己实现了一个 SPI，该SPI 可以通过名字实例化指定的实现类，并且实现了 IOC 、AOP 与 自适应扩展 SPI 。

key = com.sowhat.value

Dubbo 对配置文件目录的约定，不同于 Java SPI ，Dubbo 分为了三类目录。

• META-INF/services/ ：该目录下 SPI 配置文件是为了用来兼容 Java SPI
• META-INF/dubbo/ ：该目录存放用户自定义的 SPI 配置文件
• META-INF/dubbo/internal/ ：该目录存 Dubbo 内部使用的 SPI 配置文件

Dubbo SPI源码追踪

ExtensionLoader.getExtension 方法的整个思路是 查找缓存是否存在，不存在则读取SPI文件，通过反射创建类，然后设置依赖注入这些东西，有包装类就包装下，执行流程如下图所示：

injectExtension IOC

查找 set 方法，根据参数找到依赖对象则注入。

WrapperClass AOP

包装类，Dubbo帮你自动包装，只需某个扩展类的构造函数只有一个参数，并且是扩展接口类型，就会被判定为包装类。

Activate

Active 有三个属性，group 表示修饰在哪个端，是 provider 还是 consumer，value 表示在 URL参数中出现才会被激活，order 表示实现类的顺序。

Adaptive自适应扩展

需求：根据配置来进行 SPI 扩展的加载后不想在启动的时候让扩展被加载，想根据请求时候的参数来动态选择对应的扩展。实现：Dubbo用代理机制实现了自适应扩展，为用户想扩展的接口 通过JDK 或者 Javassist 编译生成一个代理类，然后通过反射创建实例。实例会根据本来方法的请求参数得知需要的扩展类，然后通过 ExtensionLoader.getExtensionLoader(type.class).getExtension(name)来获取真正的实例来调用，看个官网样例。

public interface WheelMaker {
    Wheel makeWheel(URL url);
}
// WheelMaker 接口的自适应实现类
public class AdaptiveWheelMaker implements WheelMaker {
    public Wheel makeWheel(URL url) {
        if (url == null) {
            throw new IllegalArgumentException("url == null");
        }
     // 1. 调用 url 的 getXXX 方法获取参数值
        String wheelMakerName = url.getParameter("Wheel.maker");
        if (wheelMakerName == null) {
            throw new IllegalArgumentException("wheelMakerName == null");
        }
        // 2. 调用 ExtensionLoader 的 getExtensionLoader 获取加载器
        // 3. 调用 ExtensionLoader 的 getExtension 根据从url获取的参数作为类名称加载实现类
        WheelMaker wheelMaker = ExtensionLoader.getExtensionLoader(WheelMaker.class).getExtension(wheelMakerName);
        // 4. 调用实现类的具体方法实现调用。
        return wheelMaker.makeWheel(URL url);
    }
}

查看Adaptive注解源码可知该注解可用在类或方法上，Adaptive 注解在类上或者方法上有不同的实现逻辑。

Adaptive注解在类上

Adaptive 注解在类上时，Dubbo 不会为该类生成代理类，Adaptive 注解在类上的情况很少，在 Dubbo 中，仅有两个类被 Adaptive 注解了，分别是 AdaptiveCompiler 和 AdaptiveExtensionFactory，表示拓展的加载逻辑由人工编码完成，这不是我们关注的重点。

Adaptive注解在方法上

Adaptive注解在方法上时，Dubbo则会为该方法生成代理逻辑，表示拓展的加载逻辑需由框架自动生成，实现机制如下：

• 加载标注有 @Adaptive 注解的接口，如果不存在，则不支持 Adaptive 机制
• 为目标接口按照一定的模板生成子类代码，并且编译生成的代码，然后通过反射生成该类的对象
• 结合生成的对象实例，通过传入的URL对象，获取指定key的配置，然后加载该key对应的类对象，最终将调用委托给该类对象进行

@SPI("apple")
public interface FruitGranter {
  Fruit grant();
  @Adaptive
  String watering(URL url);
}
---
// 苹果种植者
public class AppleGranter implements FruitGranter {
  @Override
  public Fruit grant() {
    return new Apple();
  }
  @Override
  public String watering(URL url) {
    System.out.println("watering apple");
    return "watering finished";
  }
}
---
// 香蕉种植者
public class BananaGranter implements FruitGranter {
  @Override
  public Fruit grant() {
    return new Banana();
  }
  @Override
  public String watering(URL url) {
    System.out.println("watering banana");
    return "watering success";
  }
}

调用方法实现：

public class ExtensionLoaderTest {
  @Test
  public void testGetExtensionLoader() {
    // 首先创建一个模拟用的URL对象
    URL url = URL.valueOf("dubbo://192.168.0.1:1412?fruit.granter=apple");
    // 通过ExtensionLoader获取一个FruitGranter对象
    FruitGranter granter = ExtensionLoader.getExtensionLoader(FruitGranter.class)
      .getAdaptiveExtension();
    // 使用该FruitGranter调用其"自适应标注的"方法，获取调用结果
    String result = granter.watering(url);
    System.out.println(result);
  }
}

通过如上方式生成一个内部类。

服务暴露流程

服务暴露总览

Dubbo框架是以URL为总线的模式，运行过程中所有的状态数据信息都可以通过URL来获取，比如当前系统采用什么序列化，采用什么通信，采用什么负载均衡等信息，都是通过URL的参数来呈现的，所以在框架运行过程中，运行到某个阶段需要相应的数据，都可以通过对应的Key从URL的参数列表中获取。URL 具体的参数如下：

• protocol：指的是 dubbo 中的各种协议，如：dubbo thrift http
• username/password：用户名/密码
• host/port：主机/端口
• path：接口的名称
• parameters：参数键值对

protocol://username:password@host:port/path?k=v

服务暴露从代码流程看分为三部分：

• 检查配置，最终组装成 URL。
• 暴露服务到到本地服务跟远程服务。
• 服务注册至注册中心。

服务暴露从对象构建转换看分为两步：

• 将服务封装成Invoker。
• 将Invoker通过协议转换为Exporter。

服务暴露源码追踪

• 容器启动，Spring IOC刷新完毕后调用 onApplicationEvent 开启服务暴露，ServiceBean
• export 跟 doExport 来进行拼接构建URL，为屏蔽调用的细节，统一暴露出一个可执行体，通过ProxyFactory 获取到 invoker
• 调用具体 Protocol 将把包装后的 invoker 转换成 exporter，此处用到了SPI
• 然后启动服务器server，监听端口，使用NettyServer创建监听服务器
• 通过 RegistryProtocol 将URL注册到注册中心，使得consumer可获得provider信息

服务引用流程

Dubbo中一个可执行体就是一个invoker，所以 provider 跟 consumer 都要向 invoker 靠拢。通过上面demo可知为了无感调用远程接口，底层需要有个代理类包装 invoker。

服务的引入时机有两种

• 饿汉式

  通过实现 Spring 的 InitializingBean 接口中的 afterPropertiesSet 方法，容器通过调用 ReferenceBean的 afterPropertiesSet 方法时引入服务。

• 懒汉式(默认)

  懒汉式是只有当服务被注入到其他类中时启动引入流程。

服务引用的三种方式

• 本地引入：服务暴露时本地暴露，避免网络调用开销
• 直接连接引入远程服务：不启动注册中心，直接写死远程Provider地址 进行直连
• 通过注册中心引入远程服务：通过注册中心抉择如何进行负载均衡调用远程服务

服务引用流程

• 检查配置构建map ，map 构建 URL ，通过URL上的协议利用自适应扩展机制调用对应的 protocol.refer 得到相应的 invoker ，此处
• 想注册中心注册自己，然后订阅注册中心相关信息，得到provider的 ip 等信息，再通过共享的netty客户端进行连接。
• 当有多个 URL 时，先遍历构建出 invoker 然后再由 StaticDirectory 封装一下，然后通过 cluster 进行合并，只暴露出一个 invoker 。
• 然后再构建代理，封装 invoker 返回服务引用，之后 Comsumer 调用的就是这个代理类。

调用方式

• oneway：不关心请求是否发送成功。
• Async异步调用：Dubbo天然异步，客户端调用请求后将返回的 ResponseFuture 存到上下文中，用户可随时调用 future.get 获取结果。异步调用通过唯一ID 标识此次请求。
• Sync同步调用：在 Dubbo 源码中就调用了 future.get，用户感觉方法被阻塞了，必须等结果后才返回。

调用整体流程

调用之前你可能需要考虑这些事

• consumer 跟 provider 约定好通讯协议，dubbo支持多种协议，比如dubbo、rmi、hessian、http、webservice等。默认走dubbo协议，连接属于单一长连接，NIO异步通信。适用传输数据量很小(单次请求在100kb以内)，但是并发量很高。
• 约定序列化模式，大致分为两大类，一种是字符型(XML或json 人可看懂 但传输效率低)，一种是二进制流(数据紧凑，机器友好)。默认使用 hessian2作为序列化协议。
• consumer 调用 provider 时提供对应接口、方法名、参数类型、参数值、版本号。
• provider列表对外提供服务涉及到负载均衡选择一个provider提供服务。
• consumer 跟 provider 定时向monitor 发送信息。

调用大致流程

• 客户端发起请求来调用接口，接口调用生成的代理类。代理类生成RpcInvocation 然后调用invoke方法。
• ClusterInvoker获得注册中心中服务列表，通过负载均衡给出一个可用的invoker。
• 序列化跟反序列化网络传输数据。通过NettyServer调用网络服务。
• 服务端业务线程池接受解析数据，从exportMap找到invoker进行invoke。
• 调用真正的Impl得到结果然后返回。

调用方式

• oneway：不关心请求是否发送成功，消耗最小。
• sync同步调用：在 Dubbo 源码中就调用了 future.get，用户感觉方法被阻塞了，必须等结果后才返回。
• Async 异步调用：Dubbo天然异步，客户端调用请求后将返回的 ResponseFuture 存到上下文中，用户可以随时调用future.get获取结果。异步调用通过唯一ID标识此次请求。

集群容错负载均衡

Dubbo 引入了Cluster、Directory、Router、LoadBalance、Invoker模块来保证Dubbo系统的稳健性，关系如下图：

• 服务发现时会将多个多个远程调用放入Directory，然后通过Cluster封装成一个Invoker，该invoker提供容错功能
• 消费者代用的时候从Directory中通过负载均衡获得一个可用invoker，最后发起调用
• 你可以认为Dubbo中的Cluster对上面进行了大的封装，自带各种鲁棒性功能

集群容错

集群容错是在消费者端通过Cluster子类实现的，Cluster接口有10个实现类，每个Cluster实现类都会创建一个对应的ClusterInvoker对象。核心思想是让用户选择性调用这个Cluster中间层，屏蔽后面具体实现细节。

智能容错之负载均衡

Dubbo中一般有4种负载均衡策略。

• RandomLoadBalance：加权随机，它的算法思想简单。假设有一组服务器 servers = [A, B, C]，对应权重为 weights = [5, 3, 2]，权重总和为10。现把这些权重值平铺在一维坐标值上，[0, 5) 区间属于服务器 A，[5, 8) 区间属于服务器 B，[8, 10) 区间属于服务器 C。接下来通过随机数生成器生成一个范围在 [0, 10) 之间的随机数，然后计算这个随机数会落到哪个区间上。默认实现。
• LeastActiveLoadBalance：最少活跃数负载均衡，选择现在活跃调用数最少的提供者进行调用，活跃的调用数少说明它现在很轻松，而且活跃数都是从 0 加起来的，来一个请求活跃数+1，一个请求处理完成活跃数-1，所以活跃数少也能变相的体现处理的快。
• RoundRobinLoadBalance：加权轮询负载均衡，比如现在有两台服务器 A、B，轮询的调用顺序就是 A、B、A、B，如果加了权重，A 比B 的权重是2:1，那现在的调用顺序就是 A、A、B、A、A、B。
• ConsistentHashLoadBalance：一致性 Hash 负载均衡，将服务器的 IP 等信息生成一个 hash 值，将hash 值投射到圆环上作为一个节点，然后当 key 来查找的时候顺时针查找第一个大于等于这个 key 的 hash 值的节点。一般而言还会引入虚拟节点，使得数据更加的分散，避免数据倾斜压垮某个节点。如下图 Dubbo 默认搞了 160 个虚拟节点。

智能容错之服务目录

关于 服务目录Directory 你可以理解为是相同服务Invoker的集合，核心是RegistryDirectory类。具有三个功能。

• 从注册中心获得invoker列表
• 监控着注册中心invoker的变化，invoker的上下线
• 刷新invokers列表到服务目录

智能容错之服务路由

服务路由其实就是路由规则，它规定了服务消费者可以调用哪些服务提供者。条件路由规则由两个条件组成，分别用于对服务消费者和提供者进行匹配。比如有这样一条规则：

host = 10.20.153.14 => host = 10.20.153.12

该条规则表示 IP 为 10.20.153.14 的服务消费者只可调用 IP 为 10.20.153.12 机器上的服务，不可调用其他机器上的服务。条件路由规则的格式如下：

[服务消费者匹配条件] => [服务提供者匹配条件]

如果服务消费者匹配条件为空，表示不对服务消费者进行限制。如果服务提供者匹配条件为空，表示对某些服务消费者禁用服务。

设计RPC

一个RPC框架大致需要以下功能：

1. 服务的注册与发现
2. 用动态代理
3. 负载均衡（LoadBalance）
4. 通信协议
5. 序列化与反序列化
6. 网络通信(Netty)
7. Monitor

Nacos

基本架构

服务 (Service)

服务是指一个或一组软件功能（例如特定信息的检索或一组操作的执行），其目的是不同的客户端可以为不同的目的重用（例如通过跨进程的网络调用）。Nacos 支持主流的服务生态，如 Kubernetes Service、gRPC|Dubbo RPC Service 或者 Spring Cloud RESTful Service。

服务注册中心 (Service Registry)

服务注册中心，它是服务，其实例及元数据的数据库。服务实例在启动时注册到服务注册表，并在关闭时注销。服务和路由器的客户端查询服务注册表以查找服务的可用实例。服务注册中心可能会调用服务实例的健康检查 API 来验证它是否能够处理请求。

服务元数据 (Service Metadata)

服务元数据是指包括服务端点(endpoints)、服务标签、服务版本号、服务实例权重、路由规则、安全策略等描述服务的数据。

服务提供方 (Service Provider)

是指提供可复用和可调用服务的应用方。

服务消费方 (Service Consumer)

是指会发起对某个服务调用的应用方。

配置 (Configuration)

在系统开发过程中通常会将一些需要变更的参数、变量等从代码中分离出来独立管理，以独立的配置文件的形式存在。目的是让静态的系统工件或者交付物（如 WAR，JAR 包等）更好地和实际的物理运行环境进行适配。配置管理一般包含在系统部署的过程中，由系统管理员或者运维人员完成这个步骤。配置变更是调整系统运行时的行为的有效手段之一。

配置管理 (Configuration Management)

在数据中心中，系统中所有配置的编辑、存储、分发、变更管理、历史版本管理、变更审计等所有与配置相关的活动统称为配置管理。

名字服务 (Naming Service)

提供分布式系统中所有对象(Object)、实体(Entity)的“名字”到关联的元数据之间的映射管理服务，例如 ServiceName -> Endpoints Info, Distributed Lock Name -> Lock Owner/Status Info, DNS Domain Name -> IP List, 服务发现和 DNS 就是名字服务的2大场景。

配置服务 (Configuration Service)

在服务或者应用运行过程中，提供动态配置或者元数据以及配置管理的服务提供者。

逻辑架构

• 服务管理：实现服务CRUD，域名CRUD，服务健康状态检查，服务权重管理等功能
• 配置管理：实现配置管CRUD，版本管理，灰度管理，监听管理，推送轨迹，聚合数据等功能
• 元数据管理：提供元数据CURD 和打标能力
• 插件机制：实现三个模块可分可合能力，实现扩展点SPI机制
• 事件机制：实现异步化事件通知，sdk数据变化异步通知等逻辑
• 日志模块：管理日志分类，日志级别，日志可移植性（尤其避免冲突），日志格式，异常码+帮助文档
• 回调机制：sdk通知数据，通过统一的模式回调用户处理。接口和数据结构需要具备可扩展性
• 寻址模式：解决ip，域名，nameserver、广播等多种寻址模式，需要可扩展
• 推送通道：解决server与存储、server间、server与sdk间推送性能问题
• 容量管理：管理每个租户，分组下的容量，防止存储被写爆，影响服务可用性
• 流量管理：按照租户，分组等多个维度对请求频率，长链接个数，报文大小，请求流控进行控制
• 缓存机制：容灾目录，本地缓存，server缓存机制。容灾目录使用需要工具
• 启动模式：按照单机模式，配置模式，服务模式，dns模式，或者all模式，启动不同的程序+UI
• 一致性协议：解决不同数据，不同一致性要求情况下，不同一致性机制
• 存储模块：解决数据持久化、非持久化存储，解决数据分片问题
• Nameserver：解决namespace到clusterid的路由问题，解决用户环境与nacos物理环境映射问题
• CMDB：解决元数据存储，与三方cmdb系统对接问题，解决应用，人，资源关系
• Metrics：暴露标准metrics数据，方便与三方监控系统打通
• Trace：暴露标准trace，方便与SLA系统打通，日志白平化，推送轨迹等能力，并且可以和计量计费系统打通
• 接入管理：相当于阿里云开通服务，分配身份、容量、权限过程
• 用户管理：解决用户管理，登录，sso等问题
• 权限管理：解决身份识别，访问控制，角色管理等问题
• 审计系统：扩展接口方便与不同公司审计系统打通
• 通知系统：核心数据变更，或者操作，方便通过SMS系统打通，通知到对应人数据变更
• OpenAPI：暴露标准Rest风格HTTP接口，简单易用，方便多语言集成
• Console：易用控制台，做服务管理、配置管理等操作
• SDK：多语言sdk
• Agent：dns-f类似模式，或者与mesh等方案集成
• CLI：命令行对产品进行轻量化管理，像git一样好用

功能特性

Nacos 的关键特性包括：

• 服务发现和服务健康监测

• 动态配置服务

• 动态 DNS 服务

• 服务及其元数据管理

数据模型

Nacos 数据模型 Key 由三元组唯一确定, Namespace默认是空串，公共命名空间（public），分组默认是 DEFAULT_GROUP。

Nacos-SDK类视图

配置领域模型

围绕配置，主要有两个关联的实体，一个是配置变更历史，一个是服务标签（用于打标分类，方便索引），由 ID 关联。

安装部署

下载源码或者安装包

从 Github 上下载源码方式

git clone https://github.com/alibaba/nacos.git
cd nacos/
mvn -Prelease-nacos -Dmaven.test.skip=true clean install -U  
ls -al distribution/target/

// change the $version to your actual path
cd distribution/target/nacos-server-$version/nacos/bin

下载编译后压缩包方式

您可以从 最新稳定版本 下载 nacos-server-$version.zip 包。

  unzip nacos-server-$version.zip 或者 tar -xvf nacos-server-$version.tar.gz
  cd nacos/bin

启动服务器

Linux/Unix/Mac

启动命令(standalone代表着单机模式运行，非集群模式):

sh startup.sh -m standalone

如果您使用的是ubuntu系统，或者运行脚本报错提示符号找不到，可尝试如下运行：

bash startup.sh -m standalone

Windows

启动命令(standalone代表着单机模式运行，非集群模式):

startup.cmd -m standalone

服务测试

服务注册

curl -X POST 'http://127.0.0.1:8848/nacos/v1/ns/instance?serviceName=nacos.naming.serviceName&ip=20.18.7.10&port=8080'

服务发现

curl -X GET 'http://127.0.0.1:8848/nacos/v1/ns/instance/list?serviceName=nacos.naming.serviceName'

发布配置

curl -X POST "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=nacos.cfg.dataId&group=test&content=HelloWorld"

获取配置

curl -X GET "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=nacos.cfg.dataId&group=test"

关闭服务器

Linux/Unix/Mac

sh shutdown.sh

Windows

shutdown.cmd

或者双击 shutdown.cmd运行文件。

开源案例

Spring Boot

启动配置管理

启动了 Nacos server 后，您就可以参考以下示例代码，为您的 Spring Boot 应用启动 Nacos 配置管理服务了。

第一步：添加依赖

<dependency>
    <groupId>com.alibaba.boot</groupId>
    <artifactId>nacos-config-spring-boot-starter</artifactId>
    <version>${latest.version}</version>
</dependency>

注意：版本 0.2.x.RELEASE 对应的是 Spring Boot 2.x 版本，版本 0.1.x.RELEASE 对应的是 Spring Boot 1.x 版本。

第二步：在 application.properties 中配置 Nacos server 的地址：

nacos.config.server-addr=127.0.0.1:8848

第三步：使用 @NacosPropertySource 加载 dataId 为 example 的配置源，并开启自动更新：

@SpringBootApplication
@NacosPropertySource(dataId = "example", autoRefreshed = true)
public class NacosConfigApplication {
    public static void main(String[] args) {
        SpringApplication.run(NacosConfigApplication.class, args);
    }
}

第四步：通过 Nacos 的 @NacosValue 注解设置属性值。

@Controller
@RequestMapping("config")
public class ConfigController {

    @NacosValue(value = "${useLocalCache:false}", autoRefreshed = true)
    private boolean useLocalCache;

    @RequestMapping(value = "/get", method = GET)
    @ResponseBody
    public boolean get() {
        return useLocalCache;
    }
}

第五步：启动 NacosConfigApplication，调用 curl http://localhost:8080/config/get，返回内容是 false。

第六步：通过调用 Nacos Open API 向 Nacos server 发布配置：dataId 为example，内容为useLocalCache=true

curl -X POST "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=example&group=DEFAULT_GROUP&content=useLocalCache=true"

第七步：再次访问 http://localhost:8080/config/get，此时返回内容为true，说明程序中的useLocalCache值已经被动态更新了。

启动服务发现

本节演示如何在您的 Spring Boot 项目中启动 Nacos 的服务发现功能。

第一步：添加依赖

<dependency>
    <groupId>com.alibaba.boot</groupId>
    <artifactId>nacos-discovery-spring-boot-starter</artifactId>
    <version>${latest.version}</version>
</dependency>

注意：版本 0.2.x.RELEASE 对应的是 Spring Boot 2.x 版本，版本 0.1.x.RELEASE 对应的是 Spring Boot 1.x 版本。

第二步：在 application.properties 中配置 Nacos server 的地址：

nacos.discovery.server-addr=127.0.0.1:8848

第三步：使用 @NacosInjected 注入 Nacos 的 NamingService 实例：

@Controller
@RequestMapping("discovery")
public class DiscoveryController {

    @NacosInjected
    private NamingService namingService;

    @RequestMapping(value = "/get", method = GET)
    @ResponseBody
    public List<Instance> get(@RequestParam String serviceName) throws NacosException {
        return namingService.getAllInstances(serviceName);
    }
}

@SpringBootApplication
public class NacosDiscoveryApplication {

    public static void main(String[] args) {
        SpringApplication.run(NacosDiscoveryApplication.class, args);
    }
}