秒杀系统—3.第二版升级优化的技术文档一
大纲
1.秒杀系统的服务细分和服务定位
2.秒杀系统的同步锁定库存和延迟预热
3.秒杀系统的业务流程和技术要点以及高并发能力
4.秒杀系统的各服务功能简介
5.秒杀系统的缓存组件
6.秒杀系统的异步化组件
1.秒杀系统的服务细分和服务定位
(1)秒杀系统服务细分
(2)秒杀系统各个服务的定位
(1)秒杀系统服务细分
一.秒杀相关的服务
秒杀活动服务、秒杀库存服务、秒杀抢购服务、秒杀下单服务。
二.⻚⾯静态化相关的服务
页面渲染服务、页面聚合服务、页面发布服务。
三.依赖的其他服务
依赖的商品服务、依赖的订单服务。
(2)秒杀系统各个服务的定位
2.秒杀系统的同步锁定库存和延迟预热
当创建好秒杀活动后，其实没必要马上进行秒杀商品的库存分片。因为在秒杀活动开始前，运营可能会频繁修改秒杀活动及里面的商品。
比如运营刚创建好一个秒杀活动，此时距离活动开始可能还有几天时间。如果刚创建完秒杀活动就立刻对秒杀商品进行库存分片，那么运营频繁修改秒杀活动时就会出现频繁修改或释放秒杀商品的库存。
所以可以进行如下设计：
在秒杀活动开始前的1小时之内，禁止运营修改秒杀活动。因为此时用户可以查看秒杀活动及商品了。比如用户在前端可看到活动开始倒计时(1小时)，此时商品会开始预热。当秒杀商品开始预热时，就要去对秒杀商品进行库存分片，即延迟预热。为了避免库存分片时才发现库存不足导致秒杀失败，需要提前锁定库存。可以在向秒杀活动添加秒杀商品时，同步锁定秒杀商品的库存，不让商家或运营修改。
3.秒杀系统的业务流程和技术要点以及高并发能力
(1)业务流程的时序图
(2)秒杀系统的技术要点
(3)秒杀系统的高并发能力
(1)业务流程的时序图
(2)秒杀系统的技术要点
一.线程池
首先队列都使用SynchronousQueue队列，然后通过Semaphore来限制同时处理的Task数量，这样就不用考虑队列满的时候如何处理了。
二.分布式锁
通过封装一个Jedis组件来简单实现。
三.定时任务
没有引入分布式的定时任务，而是使用Spring自带的定时任务。通过Spring自带的定时任务 + 分布式锁，简单达到分布式定时任务的效果。
四.延时队列
使用RocketMQ提供的延时队列功能即可。
五.幂等控制
在需要保证幂等的地方，统一使用Redis来去重。
六.限流处理
首先在对外的接入层使用Nginx和Lua来进行限流，然后在内部的服务上并没有继续进行限流。如果秒杀服务里有需要限流的地方，可以使用Guava或信号量来单独限流。
(3)秒杀系统的高并发能力
一.TPS和QPS预期
TPS表示每秒系统处理的事务数，表示某台机器上的秒杀抢购接口每秒可以处理的事务数，即从开始访问接口到返回响应结果记1次TPS。
QPS表示每秒查询的数量，一次TPS会包含多次QPS访问，比如用户访问一次秒杀抢购接口至少会访问5次Redis进行数据读写。
监控TPS和QPS可以采用字节码插桩或AOP来实现，QPS预期就是每秒几十万，TPS的预期是每秒2000~4000。
二.4核8G的合理QPS和Tomcat优化
4核8G的机器，一般根据经验值，会开200~300个工作线程，所以给Tomcat线程池进行优化时给到的最大线程数为230。一个Tomcat线程每秒大概处理3~5个请求，这样4和8G的机器每秒就可以轻松抗下1000的QPS。
当然这个值不是越大越好，线程太多也会导致CPU负载太高，最后导致请求处理不过来，用户访问接口的延迟时间就会变大。
对Tomcat工作线程数调优时，需要注意：一般4核8G的机器最多开200~300个工作线程就可以了，再多很可能会造成机器CPU负载过高，反而造成接口访问延迟时间增加。
三.JVM性能优化
在秒杀场景下，瞬时流量是特别大的，JVM优化的原则是要尽量避免Full GC，优化后的JVM参数如下：
-XX:MetaspaceSize=256m -XX:MaxMetaspaceSize=256m -Xms6144m -Xmx6144m -Xmn5120m -Xss256k -XX:SurvivorRatio=6 -XX:+UseConcMarkSweepGC -XX:+UseParNewGC
在这个优化下，Full GC基本不会发生，Young GC会多一些。从压测采集到的每台机器的Young GC总耗时为2.5s左右，压测5分钟Young GC的总次数为60次左右，每次Young GC的时间大概是50ms左右。即使一个请求发出后刚好遇到JVM正在进行Young GC需要消耗50ms，那么在这50ms内请求会被STW，而在这50ms后请求会开始正常工作，基本一次请求还是可以控制在200ms内，所以Young GC多一点没关系。JVM GC监控可以采用jstat -gc [pid]来进行监控，pid是应用进程的id。
四.Redis集群优化
一个普通的Redis实例，保守估计每秒可以抗2W-3W的QPS。如果要支持每秒十几万的QPS，那么需要8个Redis分片实例。
五.并发能力分析
秒杀系统对Redis的操作主要就是基于Lua脚本进行库存扣减，加上Servlet 3.0异步化处理 + 请求先进入内存队列的优化。所以4核8G的机器，每秒可以扛的最高秒杀请求大概就是600-800。
如果要扛1w的秒杀请求，那么至少用15台4核8G的机器部署秒杀系统。虽然每个秒杀请求涉及多个Redis操作，但8个Redis节点可抗十几万QPS。一般而言，一台订单系统可以抗500下单请求。如果订单系统部署在4台机器上，可以每秒抗2000下单请求。
如果在几分钟内发起了超过10w人次的秒杀抢购，那么对于下单系统来说，压力不会太大。因为延迟5~10s，下单成功，也是可以的。
4.秒杀系统的各服务功能简介
(1)秒杀相关服务——秒杀活动服务
(2)秒杀相关服务——秒杀库存服务
(3)秒杀相关服务——秒杀抢购服务
(4)秒杀相关服务——秒杀下单服务
(5)页面静态化相关服务——页面渲染服务
(6)页面静态化相关服务——页面聚合服务
(7)页面静态化相关服务——页面发布服务
(8)依赖服务——商品系统
(9)依赖服务——订单系统
(1)秒杀相关服务——秒杀活动服务
一.秒杀活动管理
添加秒杀活动：提供给管理后台的接口 删除秒杀活动：提供给管理后台的接口 修改秒杀活动：提供给管理后台的接口 后台查询秒杀活动：提供给管理后台的接口 前台查询秒杀活动：提供给前台的接口，全部读内存缓存 查询秒杀商品的销售进度：提供给前台的接口，全部读内存缓存
二.秒杀商品管理
秒杀活动添加秒杀商品：提供给管理后台的接口 秒杀活动删除秒杀商品：提供给管理后台的接口 修改秒杀活动的秒杀商品信息：提供给管理后台的接口
三.秒杀页面预览
预览秒杀活动的商品列表页面：提供给管理后台的接口 预览秒杀商品的详情页面：提供给管理后台的接口
四.秒杀活动数据准备
渲染秒杀活动的商品列表页面：内部定时任务 + 秒杀活动开始前触发 渲染秒杀商品的详情页面：内部定时任务 + 秒杀活动开始前触发 进行库存分片并同步到Redis：内部定时任务 + 完成页面渲染后触发 计算秒杀商品的销售进度：内部定时任务 + 定时计算结果放入内存缓存
五.秒杀活动数据清理
清理秒杀活动相关的数据：内部定时任务 + 秒杀活动结束后清理页面 + 库存 统计秒杀活动相关的数据：内部定时任务
(2)秒杀相关服务——秒杀库存服务
一.库存管理
库存分片并同步到Redis：提供给秒杀活动服务的接口 查询秒杀商品的实时库存：提供给秒杀活动服务的接口
二.增减库存
消费支付成功消息：减少锁定库存 + 增加已销售库存 消费订单超时未支付消息：减少锁定库存 + 增加可销售库存 消费订单取消消息：减少锁定库存 + 增加可销售库存
(3)秒杀相关服务——秒杀抢购服务
秒杀抢购：提供给前台的接口 Servlet 3.0异步化 + 责任链 + 异步消息 + 直连订单服务
(4)秒杀相关服务——秒杀下单服务
消费秒杀下单消息： 幂等下单 + 消费速率控制 + 多线程下单
(5)页面静态化相关服务——页面渲染服务
一.模版文件管理
上传模版文件：提供给管理后台的接口 + 把模版文件内容存储到Redis 下载模版文件：提供给管理后台的接口 删除模版文件：提供给管理后台的接口 查询模版文件：提供给管理后台的接口
二.页面配置管理
添加页面配置：提供给管理后台的接口 删除页面配置：提供给管理后台的接口 修改页面配置：提供给管理后台的接口 查询页面配置：提供给管理后台的接口
三.页面渲染
消费页面渲染的消息：其中流程很长 使用多线程提升性能 + 使用异步化提升吞吐量
四.静态页面管理
上传渲染好的静态页面
五.消息生产
发送页面渲染完成消息 发送延时检查消息
六.消息消费
消费延时检查结果
(6)页面静态化相关服务——页面聚合服务
秒杀商品详情页数据聚合：提供给页面渲染服务的接口 秒杀活动商品列表页数据聚合：提供给页面渲染服务的接口
(7)页面静态化相关服务——页面发布服务
消费页面渲染完成的消息 消费页面发布延时检查的消息 发送页面延时检查结果的消息
(8)依赖服务——商品系统
生成商品数据 根据skuId查询sku信息 根据spuId查询spu信息
(9)依赖服务——订单系统
创建订单：提供给秒杀下单服务的接口 查询订单：提供给前台的接口 支付订单：提供给前台的接口
5.秒杀系统的缓存组件
(1)封装Jedis配置类
(2)封装Jedis连接池对象的管理器
(3)封装对Redis进行常规操作的组件
(4)封装基于Redis实现的简单分布式锁服务
(5)封装基于SpringBoot的自动装配类
(1)封装Jedis配置类
@ConfigurationProperties(prefix = "demo.jedis")
public class JedisConfig {
private Integer maxTotal;
private Integer maxIdle;
private Integer minIdle;
private List<String> redisAddrs;
public Integer getMaxTotal() {
return maxTotal;
}
public void setMaxTotal(Integer maxTotal) {
this.maxTotal = maxTotal;
}
public Integer getMaxIdle() {
return maxIdle;
}
public void setMaxIdle(Integer maxIdle) {
this.maxIdle = maxIdle;
}
public Integer getMinIdle() {
return minIdle;
}
public void setMinIdle(Integer minIdle) {
this.minIdle = minIdle;
}
public List<String> getRedisAddrs() {
return redisAddrs;
}
public void setRedisAddrs(List<String> redisAddrs) {
this.redisAddrs = redisAddrs;
}
}
(2)封装Jedis连接池对象的管理器
public class JedisManager implements DisposableBean {
private static final Logger LOGGER = LoggerFactory.getLogger(JedisManager.class);
private final List<JedisPool> jedisPools = new ArrayList<>();
public JedisManager(JedisConfig jedisConfig) {
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
jedisPoolConfig.setMaxTotal(jedisConfig.getMaxTotal());//Jedis连接池，最大有多少个连接实例
jedisPoolConfig.setMaxIdle(jedisConfig.getMaxIdle());
jedisPoolConfig.setMinIdle(jedisConfig.getMinIdle());
//加载和解析Redis集群地址
for (String addr : jedisConfig.getRedisAddrs()) {
String[] ipAndPort = addr.split(":");
String redisIp = ipAndPort[0];
int redisPort = Integer.parseInt(ipAndPort[1]);
//针对每个Redis实例，都会去建立一个JedisPool，每个Redis实例都需要一个连接池
JedisPool jedisPool = new JedisPool(jedisPoolConfig, redisIp, redisPort);
LOGGER.info("创建JedisPool, jedisPool={}", jedisPool);
//针对各个Redis实例，都有一个连接池
jedisPools.add(jedisPool);
}
}
public int getRedisCount() {
return jedisPools.size();
}
public Jedis getJedisByIndex(int index) {
return jedisPools.get(index).getResource();
}
public Jedis getJedisByHashKey(long hashKey) {
hashKey = Math.abs(hashKey);
int index = (int) (hashKey % getRedisCount());
return getJedisByIndex(index);
}
public Jedis getJedisByHashKey(int hashKey) {
hashKey = Math.abs(hashKey);
int index = hashKey % getRedisCount();
return getJedisByIndex(index);
}
@Override
public void destroy() throws Exception {
for (JedisPool jedisPool : jedisPools) {
LOGGER.info("关闭jedisPool, jedisPool={}", jedisPool);
jedisPool.close();
}
}
}
(3)封装对Redis进行常规操作的组件
因为JedisPool是⽤commons-pool2实现的，所以可以使用try resources语法。
//RedisCacheSupport类主要封装了对Redis集群的常规操作
//首先会根据key进行hash路由，获取对应Redis节点的Jedis实例
//然后再到对应的Redis节点执行操作
public class RedisCacheSupport implements CacheSupport {
private final JedisManager jedisManager;
public RedisCacheSupport(JedisManager jedisManager) {
this.jedisManager = jedisManager;
}
@Override
public int getRedisCount() {
return jedisManager.getRedisCount();
}
@Override
public Boolean exists(String key) {
try (Jedis jedis = jedisManager.getJedisByHashKey(key.hashCode())) {
return jedis.exists(key);
}
}
@Override
public Long expire(String key, int seconds) {
try (Jedis jedis = jedisManager.getJedisByHashKey(key.hashCode())) {
return jedis.expire(key, seconds);
}
}
@Override
public Long del(String key) {
try (Jedis jedis = jedisManager.getJedisByHashKey(key.hashCode())) {
return jedis.del(key);
}
}
@Override
public Long delOnAllRedis(String key) {
for (int i = 0; i < jedisManager.getRedisCount(); i++) {
try (Jedis jedis = jedisManager.getJedisByIndex(i)) {
jedis.del(key);
}
}
return 1L;
}
@Override
public String set(String key, String value) {
try (Jedis jedis = jedisManager.getJedisByHashKey(key.hashCode())) {
return jedis.set(key, value);
}
}
@Override
public String get(String key) {
try (Jedis jedis = jedisManager.getJedisByHashKey(key.hashCode())) {
return jedis.get(key);
}
}
@Override
public Long incr(String key) {
try (Jedis jedis = jedisManager.getJedisByHashKey(key.hashCode())) {
return jedis.incr(key);
}
}
@Override
public Object eval(Long hashKey, String script) {
try (Jedis jedis = jedisManager.getJedisByHashKey(hashKey)) {
return jedis.eval(script);
}
}
//由于一个商品的库存数据可能会分散在各个Redis节点上
//所以需要从各个Redis节点查询商品库存数据，然后合并起来才算是一份总的数据
@Override
public List<Map<String, String>> hgetAllOnAllRedis(String key) {
List<Map<String, String>> list = new ArrayList<>();
for (int i = 0; i < jedisManager.getRedisCount(); i++) {
try (Jedis jedis = jedisManager.getJedisByIndex(i)) {
list.add(jedis.hgetAll(key));
}
}
return list;
}
@Override
public void hsetOnAllRedis(String key, List<Map<String, String>> hashList) {
for (int i = 0; i < jedisManager.getRedisCount(); i++) {
//通过hset命令，向每个Redis节点写入库存分片数据
try (Jedis jedis = jedisManager.getJedisByIndex(i)) {
jedis.hset(key, hashList.get(i));
}
}
}
}
(4)封装基于Redis实现的简单分布式锁服务
//进行Redis加锁时，会对key进行hash路由到某个Redis节点，再执行具体的加锁逻辑
public class RedisLockService implements LockService {
private static final String DEL_KEY_BY_VALUE =
"if redis.call('get', '%s') == '%s'" +
"then" +
" redis.call('del','%s');" +
" return '1';" +
"else" +
" return '0'" +
"end";
private final JedisManager jedisManager;
public RedisLockService(JedisManager jedisManager) {
this.jedisManager = jedisManager;
}
@Override
public String tryLock(String lockKey, long expiration, TimeUnit timeUnit) {
int hashKey = lockKey.hashCode();
try (Jedis jedis = jedisManager.getJedisByHashKey(hashKey)) {
String lockToken = UUID.randomUUID().toString();
String result = jedis.set(lockKey, lockToken, SetParams.setParams().nx().px(timeUnit.toMillis(expiration)));
if ("OK".equals(result)) {
return lockToken;
}
}
return null;
}
@Override
public boolean release(String lockKey, String lockToken) {
int hashKey = lockKey.hashCode();
try (Jedis jedis = jedisManager.getJedisByHashKey(hashKey)) {
String script = String.format(DEL_KEY_BY_VALUE, lockKey, lockToken, lockKey);
String result = (String) jedis.eval(script);
if ("1".equals(result)) {
return true;
}
}
return false;
}
}
(5)封装基于SpringBoot的自动装配类
@Configuration
@EnableConfigurationProperties(JedisConfig.class)
public class JedisAutoConfiguration {
private final JedisConfig jedisConfig;
public JedisAutoConfiguration(JedisConfig jedisConfig) {
if (Objects.isNull(jedisConfig.getMaxTotal())) {
jedisConfig.setMaxTotal(8);
}
if (Objects.isNull(jedisConfig.getMaxIdle())) {
jedisConfig.setMaxIdle(8);
}
if (Objects.isNull(jedisConfig.getMinIdle())) {
jedisConfig.setMinIdle(0);
}
if (Objects.isNull(jedisConfig.getRedisAddrs()) || jedisConfig.getRedisAddrs().isEmpty()) {
jedisConfig.setRedisAddrs(Collections.singletonList("127.0.0.1:6379"));
}
this.jedisConfig = jedisConfig;
}
@Bean(destroyMethod = "destroy")
@ConditionalOnMissingBean
//封装了针对Redis各个实例的Jedis Pool连接池
public JedisManager jedisManager() {
return new JedisManager(jedisConfig);
}
@Bean(name = "redisCacheService")
@ConditionalOnMissingBean
public CacheSupport redisCacheService(JedisManager jedisManager) {
return new RedisCacheSupport(jedisManager);
}
@Bean(name = "redisLockService")
@ConditionalOnMissingBean
public LockService redisLockService(JedisManager jedisManager) {
return new RedisLockService(jedisManager);
}
}
基于SpringBoot的自动装配类在META-INF/spring.factories文件中的配置：
org.springframework.boot.autoconfigure.EnableAutoConfiguration=com.demo.starter.jedis.autoconfigure.JedisAutoConfiguration
6.秒杀系统的异步化组件
(1)异步化组件的架构图
(2)异步化组件的运行时序图
(3)异步化组件基于SpringBoot的自动装配
(4)异步化组件(自研异步框架)的双总线设计
(5)异步化组件(自研异步框架)的线程池
(6)异步化组件(自研异步框架)的事件监听器
(7)异步化组件(自研异步框架)的事件提交
(8)异步化组件(自研异步框架)的事件分发
(1)异步化组件的架构图
秒杀系统中，抢购商品、页面渲染、页面发布都可通过异步化提升性能，可以根据双总线 + Disruptor框架来实现一个通用的异步化组件。
Disruptor的内存队列使⽤了一种叫RingBuffer的数据结构。Disruptor⼀开始会创建好所有的元素对象，并且后⾯都会复⽤这些对象。Disruptor首先是⽆锁的，然后会通过填充缓存⾏技术消除伪共享来实现高效使⽤CPU缓存。所以Disruptor的性能很⾼，⽐JUC中的阻塞队列性能要⾼。
在如下的异步化组件的架构中：BossEventBus负责把事件分发给不同的WorkEventBus，每个WorkEventBus就是⼀个事件的通道Channel，每个Channel会关联不同的自定义的Listener。每个步骤都是⼀个事件Event，最终事件会交给对应的Listener来处理，整个过程是⽆锁的，Listener处理事件时还可以使用线程池来提升性能。
(2)异步化组件的运行时序图
(3)异步化组件基于SpringBoot的自动装配
application.yml配置文件里的配置如下：
demo:
async:
boss:
ringbufferSize: 4096
eventHandlerNum: 1
workers:
- channel: /channel/01/loadPageConfig
ringbufferSize: 4096
eventHandlerNum: 1
- channel: /channel/02/downloadTemplateFile
ringbufferSize: 4096
eventHandlerNum: 1
- channel: /channel/03/aggrData
ringbufferSize: 4096
eventHandlerNum: 1
- channel: /channel/04/renderPage
ringbufferSize: 4096
eventHandlerNum: 1
- channel: /channel/05/uploadStaticPage
ringbufferSize: 4096
eventHandlerNum: 1
- channel: /channel/06/sendRenderPageLogMessage
ringbufferSize: 4096
eventHandlerNum: 1
- channel: /channel/07/sendPublishMessage
ringbufferSize: 4096
eventHandlerNum: 1
executors:
- threadPool: /channel/01/loadPageConfig
threadCount: 1
- threadPool: /channel/02/downloadTemplateFile
threadCount: 1
- threadPool: /channel/03/aggrData
threadCount: 1
- threadPool: /channel/04/renderPage
threadCount: 1
- threadPool: /channel/05/uploadStaticPage
threadCount: 1
- threadPool: /channel/06/sendRenderPageLogMessage
threadCount: 1
- threadPool: /channel/07/sendPublishMessage
threadCount: 1
加载配置文件的信息：
@ConfigurationProperties(prefix = "demo.async.boss")
public class BossConfig {
private Integer ringbufferSize;
private Integer eventHandlerNum;
public Integer getRingbufferSize() {
return ringbufferSize;
}
public void setRingbufferSize(Integer ringbufferSize) {
this.ringbufferSize = ringbufferSize;
}
public Integer getEventHandlerNum() {
return eventHandlerNum;
}
public void setEventHandlerNum(Integer eventHandlerNum) {
this.eventHandlerNum = eventHandlerNum;
}
}
@ConfigurationProperties(prefix = "demo.async")
public class WorkerConfig {
private List<Config> workers = new ArrayList<>();
public List<Config> getWorkers() {
return workers;
}
public void setWorkers(List<Config> workers) {
this.workers = workers;
}
public static class Config {
private String channel;
private Integer ringbufferSize;
private Integer eventHandlerNum;
public String getChannel() {
return channel;
}
public void setChannel(String channel) {
this.channel = channel;
}
public Integer getRingbufferSize() {
return ringbufferSize;
}
public void setRingbufferSize(Integer ringbufferSize) {
this.ringbufferSize = ringbufferSize;
}
public Integer getEventHandlerNum() {
return eventHandlerNum;
}
public void setEventHandlerNum(Integer eventHandlerNum) {
this.eventHandlerNum = eventHandlerNum;
}
}
}
@ConfigurationProperties(prefix = "demo")
public class ExecutorConfig {
private List<Config> executors = new ArrayList<>();
public List<Config> getExecutors() {
return executors;
}
public void setExecutors(List<Config> executors) {
this.executors = executors;
}
public static class Config {
private String threadPool;
private Integer threadCount;
public String getThreadPool() {
return threadPool;
}
public void setThreadPool(String threadPool) {
this.threadPool = threadPool;
}
public Integer getThreadCount() {
return threadCount;
}
public void setThreadCount(Integer threadCount) {
this.threadCount = threadCount;
}
}
}
自动装配类的实现：
@Configuration
@EnableConfigurationProperties({BossConfig.class, WorkerConfig.class, ExecutorConfig.class})
public class AsyncAutoConfigure implements ApplicationListener<ContextRefreshedEvent>, ApplicationContextAware {
private final BossConfig bossConfig;
private final WorkerConfig workerConfig;
private final ExecutorConfig executorConfig;
private ApplicationContext applicationContext;
public AsyncAutoConfigure(BossConfig bossConfig, WorkerConfig workerConfig, ExecutorConfig executorConfig) {
this.bossConfig = bossConfig;
this.workerConfig = workerConfig;
this.executorConfig = executorConfig;
}
@Bean
@Conditional(EventBusCondition.class)
@ConditionalOnMissingBean
public BossEventBus bossEventBus() {
return new BossEventBus(bossConfig, workerConfig);
}
@Bean
@Conditional(ExecutorCondition.class)
@ConditionalOnMissingBean
public ExecutorService executorService() {
return new ExecutorService(executorConfig);
}
//当创建完一个BossEventBus + 多个WorkEventBus，应用系统启动后，就会执行如下onApplicationEvent()方法
//该方法就会把EventListener实例注册到对应的WorkEventBus实例中
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
Map<String, EventListener> eventListenerMap = applicationContext.getBeansOfType(EventListener.class);
WorkEventBusManager workEventBusManager = WorkEventBusManager.getSingleton();
for (EventListener eventListener : eventListenerMap.values()) {
//这里需要过滤一下代理类，比如使用Sentinel时可能会对Bean进行增强
//如果getClass获取到的是一个代理类，那么代理类上是拿不到Channel注解的
Class<?> realClazz = CglibUtils.filterCglibProxyClass(eventListener.getClass());
Channel channel = realClazz.getAnnotation(Channel.class);
if (channel != null && !channel.value().isEmpty()) {
//获取出来对应的WorkEventBus，则注册监听器
workEventBusManager.getWorkEventBus(channel.value()).register(eventListener);
}
}
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
}
自动装配类在META-INF/spring.factories文件中的配置：
org.springframework.boot.autoconfigure.EnableAutoConfiguration=com.demo.starter.async.autoconfigure.AsyncAutoConfigure
(4)异步化组件(自研异步框架)的双总线设计
public class BossEventBus {
private final Disruptor<BossEvent> bossRingBuffer;
public BossEventBus(BossConfig bossConfig, WorkerConfig workerConfig) {
//双总线架构设计：
//BossEventBus -> 事件会分发到不同的WorkEventBus -> 不同的线程池来进行并发处理
//Boss事件总线：即主事件总线，只有一个
//比如用来处理每一个秒杀请求
//Work事件总线：即工作任务事件总线，有多个，不同类型
//比如用来处理一个秒杀请求的每一个具体步骤
//每个步骤处理完之后又发送到Work事件总线处理下一个步骤
//所以先进来的请求可能不会先被处理完
//双总线架构的设计思想：其实就是对线程池不够快的进一步改进
//通过将一个请求拆分为多个步骤，当需要处理并发的多个请求时，就可以用多个线程池分别处理每个步骤，从而提升处理并发请求的速度
//因为一段代码的执行可能需要一定的时间，一个CPU时间片并不能执行完，需要很多个CPU时间片来执行，从而产生CPU时间片的等待
//如果将一段代码的执行拆分为几个步骤，那么一个步骤的执行可能一个CPU时间片就执行完了，不会产生比较多的CPU时间片等待
//如果觉得线程池不够快，那么就用双总线处理每个线程任务 ->
//如果觉得线程池 + 阻塞队列不够快，那么就用双总线 + Disruptor处理每个线程任务 ->
//如果觉得线程中加锁后不够快，那么就让线程自旋 ->
//如果觉得线程自旋不够快，那么就让出线程执行权自旋
//首先所有的Event先进入BossEventBus里的Disruptor
//然后BossEventBus.Disruptor里的线程会把Event交给BossEventHandler处理
//接着BossEventHandler再将这些Event分发到各自对应的WorkEventBus.Disruptor
//而WorkEventBus.Disruptor里的线程又会把Event拿出来交给WorkEventHandler处理
//最后WorkEventHandler则将Event交给监听的EventListener，由EventListener中的线程池来并发处理
//1.先准备好WorkEventBus
WorkEventBusManager workEventBusManager = WorkEventBusManager.getSingleton();
for (WorkerConfig.Config config : workerConfig.getWorkers()) {
//根据配置中的channel注册WorkEventBus
workEventBusManager.register(config);
}
//2.再准备好BossEventBus
//Disruptor是高性能的内存队列，其中的RingBuffer，可以看成是一个EventBus
//当一个一个的BossEvent写入到bossRingBuffer时，Disruptor内部的线程就会取出这些BossEvent
//把BossEvent交给BossEventHandler(handleEventsWithWorkerPool)来进行处理
bossRingBuffer = new Disruptor<>(BossEvent::new, bossConfig.getRingbufferSize(), NamedDaemonThreadFactory.getInstance("BossEventBus"));
BossEventHandler[] eventHandlers = new BossEventHandler[bossConfig.getEventHandlerNum()];
for (int i = 0; i < eventHandlers.length; i++) {
eventHandlers[i] = new BossEventHandler();
}
//传入的eventHandlers表示的是多消费者，通过线程池来让多个消费者进行消费，但默认的配置bossConfig.getEventHandlerNum() = 1
bossRingBuffer.handleEventsWithWorkerPool(eventHandlers);
bossRingBuffer.start();
}
public boolean publish(String channel, BaseEvent event, AsyncContext context) {
//EventTranslator就是把传入的参数，转换为Disruptor里面的Event对象
//下面的写法等价于：
//EventTranslator<BossEvent> translator = new EventTranslator<BossEvent>() {
// @Override
// public void translateTo(BossEvent e, long sequence) {
// e.channel = channel;
// e.event = event;
// e.context = context;
// }
//};
EventTranslator<BossEvent> translator = (e, s) -> {
e.channel = channel;
e.event = event;
e.context = context;
};
//把封装的BossEvent发布到Disruptor内存队列里
//发布成功后，Disruptor内部线程会消费和处理内存队列里的BossEvent
//也就是会把BossEvent交给BossEventHandler来进行处理
boolean success = bossRingBuffer.getRingBuffer().tryPublishEvent(translator);
if (!success) {
//如果异步发布Event对象到内存队列里失败了，可以抛出一个异常
}
return success;
}
}
public class BossEventHandler implements WorkHandler<BossEvent> {
@Override
public void onEvent(BossEvent event) throws Exception {
try {
dispatchBossEvent(event);
} finally {
event.clear();
}
}
//只做事件分发
@SuppressWarnings("unchecked")
private void dispatchBossEvent(BossEvent event) {
//1.根据channel获取到对应的WorkEventBus
WorkEventBus workEventBus = WorkEventBusManager.getSingleton().getWorkEventBus(event.channel);
//2.根据事件类型获取到对应的Listener，把之前注册的Listener拿出来
List<EventListener> eventListeners = workEventBus.getEventListeners(event.event);
//3.封装WorkEvent
//下面的写法等价于：
//EventTranslator<WorkEvent> translator = new EventTranslator<BossEvent>() {
// @Override
// public void translateTo(WorkEvent e, long sequence) {
// e.event = event;
// e.context = context;
// e.listeners = eventListeners;
// }
//};
EventTranslator<WorkEvent> translator = (e, s) -> {
e.event = event.event;//事件类型
e.context = event.context;//数据上下文
e.listeners = eventListeners;//注册到WorkEventBus里的Listener
};
//4.把Event分发到channel指定的WorkEventBus里
//WorkEvent会进入到内存队列里，内部会有一个线程，拿到WorkEvent，交给WorkEventHandler处理
boolean publish = workEventBus.publish(translator);
if (!publish) {
//如果发布到WorkEventBus时，遇到队列满的问题，那么publish就会为false
}
}
}
public class WorkEventBusManager {
private static final WorkEventBusManager SINGLETON = new WorkEventBusManager();
private final ConcurrentHashMap<String, WorkEventBus> BUFFER = new ConcurrentHashMap<>();
private WorkEventBusManager() {
}
public static WorkEventBusManager getSingleton() {
return SINGLETON;
}
//总的入口放到了BossEventBus的构造方法中
//在那里会先创建好所有的WorkEventBus，然后再创建BossEventBus
public void register(WorkerConfig.Config config) {
BUFFER.computeIfAbsent(config.getChannel(), k -> new WorkEventBus<>(
config.getRingbufferSize(),
config.getEventHandlerNum(),
WorkEvent::new,
WorkEventHandler::new)
);
}
public WorkEventBus getWorkEventBus(String channel) {
return BUFFER.get(channel);
}
}
public class WorkEventBus<E> {
private final Disruptor<E> workRingBuffer;
//用来存放WorkEventBus的所有Listener, 这些Listener可以按照处理的事件进行划分
private final List<EventListener> eventListeners = new ArrayList<>();
private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
@SuppressWarnings("unchecked")
public WorkEventBus(int ringBufferSize, int workerHandlerNum, EventFactory<E> eventFactory, Supplier<WorkHandler<E>> workHandlerSupplier) {
//Disruptor是高性能的内存队列
workRingBuffer = new Disruptor<>(eventFactory, ringBufferSize, NamedDaemonThreadFactory.getInstance("WorkEventBus"));
WorkHandler<E>[] workHandlers = new WorkHandler[workerHandlerNum]; // 1
for (int i = 0; i < workHandlers.length; i++) {
workHandlers[i] = workHandlerSupplier.get();
}
workRingBuffer.handleEventsWithWorkerPool(workHandlers);
workRingBuffer.start();
}
//注册Listener的方法
//可以在容器生命周期的某个阶段自行或者手动注册Listener，比如在自动装配类AsyncAutoConfigure中自动注册
public boolean register(EventListener eventListener) {
lock.writeLock().lock();
try {
if (eventListeners.contains(eventListener)) {
return false;
}
eventListeners.add(eventListener);
return true;
} finally {
lock.writeLock().unlock();
}
}
public List<EventListener> getEventListeners(BaseEvent event) {
//获取关注event事件的Listener
lock.readLock().lock();
try {
//把注册过的Listener查出来
return eventListeners.stream().filter(e -> e.accept(event)).collect(Collectors.toList());
} finally {
lock.readLock().unlock();
}
}
public boolean publish(EventTranslator<E> translator) {
return workRingBuffer.getRingBuffer().tryPublishEvent(translator);
}
}
public class WorkEventHandler implements WorkHandler<WorkEvent> {
@Override
public void onEvent(WorkEvent event) throws Exception {
try {
processWorkEvent(event);
} finally {
event.clear();
}
}
//这里才开始处理事件
@SuppressWarnings("unchecked")
private void processWorkEvent(WorkEvent event) {
for (EventListener listener : event.listeners) {
listener.onEvent(event.event, event.context);
}
}
}
(5)异步化组件(自研异步框架)的线程池
public class ExecutorService {
private static final ConcurrentHashMap<String, SafeThreadPool> BUFFER = new ConcurrentHashMap<>();
public ExecutorService(ExecutorConfig executorConfig) {
for (ExecutorConfig.Config config : executorConfig.getExecutors()) {
BUFFER.put(config.getThreadPool(), new SafeThreadPool(config.getThreadPool(), config.getThreadCount()));
}
}
public void execute(String channel, Runnable task) {
//Optional.ofNullable()方法的作用是将一个可能为null的值包装到Optional容器中
//如果该值为null，则返回一个空的Optional对象，否则返回一个包含该值的Optional对象
//使用Optional.ofNullable()可以有效地避免空指针异常
//因为它可以让我们在获取一个可能为null的对象时，先判断该对象是否为空，从而避免出现空指针异常
Optional.ofNullable(BUFFER.get(channel)).ifPresent(safeThreadPool -> safeThreadPool.execute(task));
}
}
public class SafeThreadPool {
private final Semaphore semaphore;
private final ThreadPoolExecutor threadPoolExecutor;
public SafeThreadPool(String name, int permits) {
//设置Semaphore信号量为线程数量
semaphore = new Semaphore(permits);
//根据线程数量封装一个线程池，其中最大线程数量maximum的大小就是线程数量permits * 2
//可以往这个线程池里提交最多maximumPoolSize个任务
threadPoolExecutor = new ThreadPoolExecutor(
0,
permits * 2,
60,//线程空闲了60秒就回收
TimeUnit.SECONDS,
new SynchronousQueue<>(),//无缓冲阻塞队列，用来在两个线程之间移交元素
NamedDaemonThreadFactory.getInstance(name)
);
}
public void execute(Runnable task) {
//每次往这个线程池提交任务时，都需要先获取一个信号量
//所以同一时刻，最多只能提交数量与信号量(线程数量)相同的任务到线程池里
//当有超过线程数量的任务提交时，便会在执行下面的代码"获取信号量"时，被阻塞住
semaphore.acquireUninterruptibly();
//虽然使用了semaphore去限制提交到线程池的线程任务数
//但是极端情况下，还是可能会有(信号量 * 2)个线程任务被提交到线程池
//这种极端情况就是：
//线程任务执行完任务并释放掉信号量时，还没释放自己被线程池回收，其他线程就获取到信号量提交到线程池了
threadPoolExecutor.submit(() -> {
try {
//执行任务
task.run();
} finally {
//释放信号量
semaphore.release();
}
//某线程执行到这里时，还没完全把自己释放出来，但信号量已释放，可能新的任务已经加入线程池
});
}
}
(6)异步化组件(自研异步框架)的事件监听器
当创建完一个BossEventBus + 多个WorkEventBus，且应用系统启动后，就会执行基于SpringBoot的自动装配类AsyncAutoConfigure的onApplicationEvent()方法，该方法会把EventListener实例注册到对应的WorkEventBus实例中。
@Configuration
@EnableConfigurationProperties({BossConfig.class, WorkerConfig.class, ExecutorConfig.class})
public class AsyncAutoConfigure implements ApplicationListener<ContextRefreshedEvent>, ApplicationContextAware {
...
//当创建完一个BossEventBus + 多个WorkEventBus，应用系统启动后，就会执行如下onApplicationEvent()方法
//该方法就会把EventListener实例注册到对应的WorkEventBus实例中
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
Map<String, EventListener> eventListenerMap = applicationContext.getBeansOfType(EventListener.class);
WorkEventBusManager workEventBusManager = WorkEventBusManager.getSingleton();
for (EventListener eventListener : eventListenerMap.values()) {
//这里需要过滤一下代理类，比如使用Sentinel时可能会对Bean进行增强
//如果getClass获取到的是一个代理类，那么代理类上是拿不到Channel注解的
Class<?> realClazz = CglibUtils.filterCglibProxyClass(eventListener.getClass());
Channel channel = realClazz.getAnnotation(Channel.class);
if (channel != null && !channel.value().isEmpty()) {
//获取出来对应的WorkEventBus，则注册监听器
workEventBusManager.getWorkEventBus(channel.value()).register(eventListener);
}
}
}
}
//该注解会被添加到自定义的XxxEventListener上, 指定这个Listener监听哪个Channel
@Documented
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface Channel {
String value();
}
//EventListener可以针对指定的Event事件进行监听
//当Event事件发布到WorkEventBus事件总线时，
//WorkEventHandler就会根据Event事件获取对应的EventListener进行事件处理
public interface EventListener<E extends BaseEvent> extends java.util.EventListener {
//BossEventHandler在分发事件时, 就会调用这个方法获取可以处理事件的Listener
//如果当前监听器EventListener确实监听某个Event事件，那么该方法就会返回true
boolean accept(BaseEvent event);
//实现处理具体Event事件的逻辑
//WorkEventHandler会调用监听器EventListener的这个onEvent()方法来处理具体的事件
void onEvent(E event, AsyncContext eventContext);
}
public class BossEventHandler implements WorkHandler<BossEvent> {
@Override
public void onEvent(BossEvent event) throws Exception {
try {
dispatchBossEvent(event);
} finally {
event.clear();
}
}
//只做事件分发
@SuppressWarnings("unchecked")
private void dispatchBossEvent(BossEvent event) {
//1.根据channel获取到对应的WorkEventBus
WorkEventBus workEventBus = WorkEventBusManager.getSingleton().getWorkEventBus(event.channel);
//2.根据事件类型获取到对应的Listener，把之前注册的Listener拿出来
List<EventListener> eventListeners = workEventBus.getEventListeners(event.event);
//3.封装WorkEvent
EventTranslator<WorkEvent> translator = (e, s) -> {
e.event = event.event;//事件类型
e.context = event.context;//数据上下文
e.listeners = eventListeners;//注册到WorkEventBus里的Listener
};
//4.把Event分发到channel指定的WorkEventBus里
//WorkEvent会进入到内存队列里，内部会有一个线程，拿到WorkEvent，交给WorkEventHandler处理
boolean publish = workEventBus.publish(translator);
if (!publish) {
//如果发布到WorkEventBus时，遇到队列满的问题，那么publish就会为false
}
}
}
public class WorkEventHandler implements WorkHandler<WorkEvent> {
@Override
public void onEvent(WorkEvent event) throws Exception {
try {
processWorkEvent(event);
} finally {
event.clear();
}
}
//这里才开始处理事件
@SuppressWarnings("unchecked")
private void processWorkEvent(WorkEvent event) {
//遍历注册的所有Listener，把发布到WorkEventBus的事件交给其Listener来处理
for (EventListener listener : event.listeners) {
listener.onEvent(event.event, event.context);
}
}
}
(7)异步化组件(自研异步框架)的事件提交
@Component
@RocketMQMessageListener(topic = QueueKey.QUEUE_RENDER_PAGE, consumerGroup = "rendPageConsumer")
public class RenderPageListener implements RocketMQListener<String> {
//异步框架BossEventBus
@Autowired
private BossEventBus bossEventBus;
@Override
public void onMessage(String messageString) {
log.info("收到渲染页面的消息, message={}", messageString);
try {
JSONObject message = JSONObject.parseObject(messageString);
PageRenderContext context = new PageRenderContext();
context.setPageCode(message.getString("pageCode"));
context.setBizData(message.getJSONObject("bizData"));
context.setParams(message.getJSONObject("params"));
context.setFileName(message.getString("fileName"));
context.setPageLog(new PageLog());
context.getPageLog().setStartTime(System.currentTimeMillis());
context.getPageLog().setBizData(JSON.toJSONString(context.getBizData()));
context.getPageLog().setOpType("render");
context.getPageLog().setFileName(context.getFileName());
context.getPageLog().setServerIps(BussinessConfig.getNginxServerIps());
context.getPageLog().setMsg(messageString);
//页面渲染的步骤：
//加载页面配置 -> 下载页面模板 -> 聚合数据 -> 渲染页面 -> 上传静态化页面 -> 保存页面渲染日志 -> 发布页面渲染成功的消息
bossEventBus.publish(ChannelKey.CHANNEL_01_LOAD_PAGE_CONFIG, PageRenderEventHolder.EVENT_01, context);
} catch (Exception ignore) {
}
}
}
public class BossEventBus {
private final Disruptor<BossEvent> bossRingBuffer;
...
public boolean publish(String channel, BaseEvent event, AsyncContext context) {
//EventTranslator就是把传入的参数，转换为Disruptor里面的Event对象
EventTranslator<BossEvent> translator = (e, s) -> {
e.channel = channel;
e.event = event;
e.context = context;
};
//把封装的BossEvent发布到Disruptor内存队列里
//发布成功后，Disruptor内部线程会消费和处理内存队列里的BossEvent
//也就是会把BossEvent交给BossEventHandler来进行处理
boolean success = bossRingBuffer.getRingBuffer().tryPublishEvent(translator);
if (!success) {
//如果异步发布Event对象到内存队列里失败了，可以抛出一个异常
}
return success;
}
}
(8)异步化组件(自研异步框架)的事件分发
public class BossEventHandler implements WorkHandler<BossEvent> {
@Override
public void onEvent(BossEvent event) throws Exception {
try {
dispatchBossEvent(event);
} finally {
event.clear();
}
}
//只做事件分发
@SuppressWarnings("unchecked")
private void dispatchBossEvent(BossEvent event) {
//1.根据channel获取到对应的WorkEventBus
WorkEventBus workEventBus = WorkEventBusManager.getSingleton().getWorkEventBus(event.channel);
//2.根据事件类型获取到对应的Listener，把之前注册的Listener拿出来
List<EventListener> eventListeners = workEventBus.getEventListeners(event.event);
//3.封装WorkEvent
EventTranslator<WorkEvent> translator = (e, s) -> {
e.event = event.event;//事件类型
e.context = event.context;//数据上下文
e.listeners = eventListeners;//注册到WorkEventBus里的Listener
};
//4.把Event分发到channel指定的WorkEventBus里
//WorkEvent会进入到内存队列里，内部会有一个线程，拿到WorkEvent，交给WorkEventHandler处理
boolean publish = workEventBus.publish(translator);
if (!publish) {
//如果发布到WorkEventBus时，遇到队列满的问题，那么publish就会为false
}
}
}
public class WorkEventHandler implements WorkHandler<WorkEvent> {
@Override
public void onEvent(WorkEvent event) throws Exception {
try {
processWorkEvent(event);
} finally {
event.clear();
}
}
//这里才开始处理事件
@SuppressWarnings("unchecked")
private void processWorkEvent(WorkEvent event) {
//遍历注册的所有Listener，把发布到WorkEventBus的事件交给其Listener来处理
for (EventListener listener : event.listeners) {
listener.onEvent(event.event, event.context);
}
}
}
后端技术栈的基础修养 文章被收录于专栏
详细介绍后端技术栈的基础内容，包括但不限于：MySQL原理和优化、Redis原理和应用、JVM和G1原理和优化、RocketMQ原理应用及源码、Kafka原理应用及源码、ElasticSearch原理应用及源码、JUC源码、Netty源码、zk源码、Dubbo源码、Spring源码、Spring Boot源码、SCA源码、分布式锁源码、分布式事务、分库分表和TiDB、大型商品系统、大型订单系统等
美团HR面2292人在聊