外观
JD-hotkey源码剖析
官方架构
如上图表述了hotkey在运行时的整体流程，hotkey可以对任意突发性的无法预先感知的热点数据，包括并不限于热点数据（如突发大量请求同一个商品）、热用户（如恶意爬虫刷子）、热接口（突发海量请求同一个接口）等，进行毫秒级精准探测到。然后对这些热数据、热用户等，推送到所有服务端JVM内存中，以大幅减轻对后端数据存储层的冲击。
那么它是怎么做的呢？下面从我看官方结构图，对各个节点产生的疑问进行分析
ps：源码上很多细节已经标明了注释，这里仅是串联各个端的作用。
整体链路图
清晰流程图可以查看：www.yuque.com/molost/lsho… 《JD-hotkey链路图》
一、dashboard服务端
规则信息变动做了什么处理
访问配置规则页面，可以看到发起了 /rule/list 请求，那么大胆猜测一下，保存也是在类似的路径 /rule/save
通过请求路径，直接定位到相应的代码，其实就是一个mvc架构的web工程。
整体流程执行如下：
com.jd.platform.hotkey.dashboard.biz.controller.RuleController#save
public Result save(Rules rules){
// 检验规则所属APP 信息
checkApp(rules.getApp());
// 校验热点规则格式
checkRule(rules.getRules());
rules.setUpdateUser(userName());
// 这里时重点，调用service保存对应的规则
int b = ruleService.save(rules);
return b == 0 ? Result.fail():Result.success();
}
com.jd.platform.hotkey.dashboard.biz.service.impl.RuleServiceImpl#save
@Override
public int save(Rules rules) {
String app = rules.getApp();
// 对应路径 /jd/rules/ + app 作为key，从etcd获取旧的热kye规则值
KeyValue kv = configCenter.getKv(ConfigConstant.rulePath + app);
// 存在就保存历史规则快照
String from = null;
if (kv != null) {
from = kv.getValue().toStringUtf8();
}
// 当前新规则快照
String to = JSON.toJSONString(rules);
// 对应路径 /jd/rules/ + app 作为key。向etcd直接设置 新的规则值，对旧值进行覆盖
configCenter.put(ConfigConstant.rulePath + app, rules.getRules());
// 根据新旧快照，保存操作日志记录
logMapper.insertSelective(new ChangeLog(app, 1, from, to, rules.getUpdateUser(), app, SystemClock.nowDate()));
return 1;
}可以看到规则变化后，仅更新ectd的值和记录操作日志，并没有主动通知worker和client
这里记录一下新的疑问？ 可以继续往下看，再回来回顾这两个问题。
- 为什么不主动通知worker和client？
- 减轻了dashboard的负担，特别是在大规模客户端的环境中，节约了dashboard的资源worker和client节点数量多的时候的交互成本
- 每当规则发生变化时，dashboard就需要向大量客户端发送更新信息。这可能导致网络拥塞和 dashboard 性能瓶颈
- 如果dashboard或网络发生故障导致推送失败，客户端可能无法获得最新的规则列表，这会影响热key的计算。而在拉取模式下，客户端如果在尝试拉取时失败，可以轻易实现重试逻辑，直到成功更新信息
- 简化dashboard设计，而不是重复造轮子。
- worker和client分别是怎么感知规则变化的？
- 参考下面内容
二、Client客户端
客户端接入的代码，从 startPipeline 方法开始看
/** 应用初始化 */
@PostConstruct
public void initHotkey() {
ClientStarter.Builder builder = new ClientStarter.Builder();
ClientStarter starter = builder.setAppName("test_hot_key").setEtcdServer("http://127.0.0.1:2379").build();
starter.startPipeline();
}
/**
* com.jd.platform.hotkey.client.ClientStarter#startPipeline
* 启动监听etcd
*/
public void startPipeline() {
JdLogger.info(getClass(), "etcdServer:" + etcdServer);
//设置caffeine的最大容量
Context.CAFFEINE_SIZE = caffeineSize;
//设置etcd地址
EtcdConfigFactory.buildConfigCenter(etcdServer);
//开始定时推送
PushSchedulerStarter.startPusher(pushPeriod);
PushSchedulerStarter.startCountPusher(10);
//开启worker重连器
WorkerRetryConnector.retryConnectWorkers();
registEventBus();
EtcdStarter starter = new EtcdStarter();
//与etcd相关的监听都开启
starter.start();
}
private void registEventBus() {
//netty连接器会关注WorkerInfoChangeEvent事件
EventBusCenter.register(new WorkerChangeSubscriber());
//热key探测回调关注热key事件
EventBusCenter.register(new ReceiveNewKeySubscribe());
//Rule的变化的事件
EventBusCenter.register(new KeyRuleHolder());
}
// com.jd.platform.hotkey.client.etcd.EtcdStarter#start
public void start() {
fetchWorkerInfo();
fetchRule();
startWatchRule();
//监听热key事件，只监听手工添加、删除的key
startWatchHotKey();
}1、怎么维护跟worker的关系和连接
关注如下代码部分
//开启worker重连器
WorkerRetryConnector.retryConnectWorkers();
//netty连接器会关注WorkerInfoChangeEvent事件
EventBusCenter.register(new WorkerChangeSubscriber());这里主要先看开启监听的 WorkerChangeSubscriber 订阅事件，可以看到一共订阅两个事件，一个是worker信息变动，一个是断连
public class WorkerChangeSubscriber {
/**
* 监听worker信息变动
*/
@Subscribe
public void connectAll(WorkerInfoChangeEvent event) {
List<String> addresses = event.getAddresses();
if (addresses == null) {
addresses = new ArrayList<>();
}
WorkerInfoHolder.mergeAndConnectNew(addresses);
}
/**
* 当client与worker的连接断开后，删除
*/
@Subscribe
public void channelInactive(ChannelInactiveEvent inactiveEvent) {
//获取断线的channel
Channel channel = inactiveEvent.getChannel();
InetSocketAddress socketAddress = (InetSocketAddress) channel.remoteAddress();
String address = socketAddress.getHostName() + ":" + socketAddress.getPort();
JdLogger.warn(getClass(), "this channel is inactive : " + socketAddress + " trying to remove this connection");
WorkerInfoHolder.dealChannelInactive(address);
}
}获取worker信息并连接
看导入的包，@Subscribe 注解是google的 eventbus 框架实现的：greenrobot.org/eventbus/
那么我们可以找对应的事件是哪里发出来的，idea直接搜索 事件类名，
往上追踪方法调用，可以看到是启动时开启了一个定时任务执行，从etcd获取相应的worker列表，然后发布一个变更事件。
com.jd.platform.hotkey.client.etcd.EtcdStarter#fetchWorkerInfo
/**
* 每隔30秒拉取worker信息
*/
private void fetchWorkerInfo() {
ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
//开启拉取etcd的worker信息，如果拉取失败，则定时继续拉取
scheduledExecutorService.scheduleAtFixedRate(() -> {
JdLogger.info(getClass(), "trying to connect to etcd and fetch worker info");
fetch();
}, 0, 30, TimeUnit.SECONDS);
}
private void fetch() {
IConfigCenter configCenter = EtcdConfigFactory.configCenter();
try {
//获取所有worker的ip
List<KeyValue> keyValues = configCenter.getPrefix(ConfigConstant.workersPath + Context.APP_NAME);
//worker为空，可能该APP没有自己的worker集群，就去连默认的，如果默认的也没有，就不管了，等着心跳
if (CollectionUtil.isEmpty(keyValues)) {
keyValues = configCenter.getPrefix(ConfigConstant.workersPath + "default");
}
//全是空，给个警告
if (CollectionUtil.isEmpty(keyValues)) {
JdLogger.warn(getClass(), "very important warn !!! workers ip info is null!!!");
}
List<String> addresses = new ArrayList<>();
if (keyValues != null) {
for (KeyValue keyValue : keyValues) {
//value里放的是ip地址
String ipPort = keyValue.getValue().toStringUtf8();
addresses.add(ipPort);
}
}
JdLogger.info(getClass(), "worker info list is : " + addresses + ", now addresses is "
+ WorkerInfoHolder.getWorkers());
//发布workerinfo变更信息 事件
notifyWorkerChange(addresses);
} catch (StatusRuntimeException ex) {
//etcd连不上
JdLogger.error(getClass(), "etcd connected fail. Check the etcd address!!!");
}
}
com.jd.platform.hotkey.client.etcd.EtcdStarter#notifyWorkerChange
private void notifyWorkerChange(List<String> addresses) {
EventBusCenter.getInstance().post(new WorkerInfoChangeEvent(addresses));
}订阅事件执行后，对不存在worker节点进行剔除和连接新增的worker节点，详细看下面的注释
public static void mergeAndConnectNew(List<String> allAddresses) {
// 移除那些在最新的worker地址集里没有的那些 下线的worker节点
removeNoneUsed(allAddresses);
//去连接那些在etcd里有，但是list里没有的
List<String> needConnectWorkers = newWorkers(allAddresses);
if (needConnectWorkers.size() == 0) {
return;
}
JdLogger.info(WorkerInfoHolder.class, "new workers : " + needConnectWorkers);
//再连接，连上后，value就有值了
NettyClient.getInstance().connect(needConnectWorkers);
Collections.sort(WORKER_HOLDER);
}
/**
* 底层保存worker集群其实就是一个 CopyOnWriteArrayList
* 保存worker的ip地址和Channel的映射关系，这是有序的。每次client发送消息时，都会根据该map的size进行hash
* 如key-1就发送到workerHolder的第1个Channel去，key-2就发到第2个Channel去
*/
private static final List<Server> WORKER_HOLDER = new CopyOnWriteArrayList<>();
/**
* 使用netty发起连接，连接成功或者失败，都保存到 WORKER_HOLDER 中，
* 区别就是连接成功的会保存 相应的连接信息到 Server中
*/
public synchronized boolean connect(List<String> addresses) {
boolean allSuccess = true;
for (String address : addresses) {
if (WorkerInfoHolder.hasConnected(address)) {
continue;
}
String[] ss = address.split(":");
try {
// 发起netty连接
ChannelFuture channelFuture = bootstrap.connect(ss[0], Integer.parseInt(ss[1])).sync();
Channel channel = channelFuture.channel();
// 连接成功保存到 WORKER_HOLDER 中
WorkerInfoHolder.put(address, channel);
} catch (Exception e) {
JdLogger.error(getClass(), "----该worker连不上----" + address);
// 连接失败的也保存到 WORKER_HOLDER 中，但是连接信息为空
WorkerInfoHolder.put(address, null);
allSuccess = false;
}
}
return allSuccess;
}失败重连
那么连接失败的后续怎么处理？
还记得开头说要关注的代码，有一个 WorkerRetryConnector 重连器，就是在这里处理连接失败的worker。
public class WorkerRetryConnector {
/**
* 定时去重连没连上的workers
*/
public static void retryConnectWorkers() {
@SuppressWarnings("PMD.ThreadPoolCreationRule")
ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("worker-retry-connector-service-executor", true));
//开启拉取etcd的worker信息，如果拉取失败，则定时继续拉取
scheduledExecutorService.scheduleAtFixedRate(WorkerRetryConnector::reConnectWorkers, 30, 30, TimeUnit.SECONDS);
}
/** 获取连接失败的Server, 然后重新连接 */
private static void reConnectWorkers() {
List<String> nonList = WorkerInfoHolder.getNonConnectedWorkers();
if (nonList.size() == 0) {
return;
}
JdLogger.info(WorkerRetryConnector.class, "trying to reConnect to these workers :" + nonList);
NettyClient.getInstance().connect(nonList);
}
}断开连接
跟连接类似，也是由监听的具体事件往上推导，可以看到是处理和Netty的断连事件，然后进行处理，其实就是获取到具体的地址，然后从本地的列表缓存中移除，细节可以自己跟一下具体的代码。
com.jd.platform.hotkey.client.netty.NettyClientHandler#channelInactive
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
//断线了，可能只是client和server断了，但都和etcd没断。也可能是client自己断网了，也可能是server断了
//发布断线事件。后续10秒后进行重连，根据etcd里的worker信息来决定是否重连，如果etcd里没了，就不重连。如果etcd里有，就重连
notifyWorkerChange(ctx.channel());
}
private void notifyWorkerChange(Channel channel) {
EventBusCenter.getInstance().post(new ChannelInactiveEvent(channel));
}2、如何感知变化规则变更
关注如下代码片段
com.jd.platform.hotkey.client.ClientStarter#registEventBus
//Rule的变化的事件
EventBusCenter.register(new KeyRuleHolder());
com.jd.platform.hotkey.client.etcd.EtcdStarter#start
fetchRule();
startWatchRule();整体思路和维护worker差不多，都是先获取全量的热key规则，然后发布变更事件。异步解耦执行规则变更逻辑，
但是这里有个差别点，就是获取规则的时机。
拉取规则的时机
- fetchRule()，启动定时任务获取全量的规则信息，如果获取成功，则会同步拉取已经存在的热key，同时关闭定时调度的线程池，这里主要是用于应用初始加载规则和热key缓存配置。
private void fetchRule() {
ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
//开启拉取etcd的worker信息，如果拉取失败，则定时继续拉取
scheduledExecutorService.scheduleAtFixedRate(() -> {
JdLogger.info(getClass(), "trying to connect to etcd and fetch rule info");
// 全量拉取rule信息
boolean success = fetchRuleFromEtcd();
if (success) {
//拉取已存在的热key
fetchExistHotKey();
// 然后关闭定时调度的线程池
scheduledExecutorService.shutdown();
}
}, 0, 5, TimeUnit.SECONDS);
}- startWatchRule()，开启etcd监听事件，监听应用下 所有的客户端规则的变动，这里主要是考虑规则变动的一个时效性。
/**
* 异步监听rule规则变化
*/
private void startWatchRule() {
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.submit(() -> {
JdLogger.info(getClass(), "--- begin watch rule change ----");
try {
IConfigCenter configCenter = EtcdConfigFactory.configCenter();
KvClient.WatchIterator watchIterator = configCenter.watch(ConfigConstant.rulePath + Context.APP_NAME);
//如果有新事件，即rule的变更，就重新拉取所有的信息
while (watchIterator.hasNext()) {
//这句必须写，next会让他卡住，除非真的有新rule变更
WatchUpdate watchUpdate = watchIterator.next();
List<Event> eventList = watchUpdate.getEvents();
JdLogger.info(getClass(), "rules info changed. begin to fetch new infos. rule change is " + eventList);
//全量拉取rule信息
fetchRuleFromEtcd();
}
} catch (Exception e) {
JdLogger.error(getClass(), "watch err");
}
});
}拉取规则的逻辑
根据appname 从etcd获取自己的rule，然后发布规则变更事件
- 如果规则为空，清空对应的本地规则缓存信息
- 不为空，直接替换整个规则集合
- 然后先清除掉那些在RULE_CACHE_MAP里存的，但是rule里已没有的
- 最后再将新增的rule，放到RULE_CACHE_MAP里面
private boolean fetchRuleFromEtcd() {
IConfigCenter configCenter = EtcdConfigFactory.configCenter();
try {
List<KeyRule> ruleList = new ArrayList<>();
//从etcd获取自己的rule
String rules = configCenter.get(ConfigConstant.rulePath + Context.APP_NAME);
// 如果规则为空
if (StringUtil.isNullOrEmpty(rules)) {
JdLogger.warn(getClass(), "rule is empty");
// 发布规则变更事件
notifyRuleChange(ruleList);
return true;
}
ruleList = FastJsonUtils.toList(rules, KeyRule.class);
// 发布规则变更事件
notifyRuleChange(ruleList);
return true;
} catch (StatusRuntimeException ex) {
//etcd连不上
JdLogger.error(getClass(), "etcd connected fail. Check the etcd address!!!");
return false;
} catch (Exception e) {
JdLogger.error(getClass(), "fetch rule failure, please check the rule info in etcd");
return true;
}
}
// 发布变更事件
private void notifyRuleChange(List<KeyRule> rules) {
EventBusCenter.getInstance().post(new KeyRuleInfoChangeEvent(rules));
}
// 订阅事件
@Subscribe
public void ruleChange(KeyRuleInfoChangeEvent event) {
JdLogger.info(getClass(), "new rules info is :" + event.getKeyRules());
List<KeyRule> ruleList = event.getKeyRules();
if (ruleList == null) {
return;
}
// 保存规则
putRules(ruleList);
}
/**
* 保存超时时间和caffeine的映射，key是超时时间，value是caffeine
*/
private static final ConcurrentHashMap<Integer, LocalCache> RULE_CACHE_MAP = new ConcurrentHashMap<>();
private static final List<KeyRule> KEY_RULES = new ArrayList<>();
/**
* 所有的规则，如果规则的超时时间变化了，会重建caffeine
*/
public static void putRules(List<KeyRule> keyRules) {
// 加锁执行
synchronized (KEY_RULES) {
//如果规则为空，清空规则表
if (CollectionUtil.isEmpty(keyRules)) {
KEY_RULES.clear();
RULE_CACHE_MAP.clear();
return;
}
KEY_RULES.clear();
KEY_RULES.addAll(keyRules);
Set<Integer> durationSet = keyRules.stream().map(KeyRule::getDuration).collect(Collectors.toSet());
for (Integer duration : RULE_CACHE_MAP.keySet()) {
//先清除掉那些在RULE_CACHE_MAP里存的，但是rule里已没有的
if (!durationSet.contains(duration)) {
RULE_CACHE_MAP.remove(duration);
}
}
//遍历所有的规则
for (KeyRule keyRule : keyRules) {
int duration = keyRule.getDuration();
if (RULE_CACHE_MAP.get(duration) == null) {
LocalCache cache = CacheFactory.build(duration);
RULE_CACHE_MAP.put(duration, cache);
}
}
}
}3、怎么上报数据给worker
看官方文档使用的API中，下面几个方法再使用过程中是会上报对应的key数据
/// 该方法会返回该key是否是热key，如果是返回true，如果不是返回false，并且会将key上报到探测集群进行数量计算。该方法通常用于判断只需要判断key是否热、不需要缓存value的场景，如刷子用户、接口访问频率等。
boolean JdHotKeyStore.isHotKey(String key)
// 该方法是一个整合方法，相当于isHotKey和get两个方法的整合，也是会上报信息
Object JdHotKeyStore.getValue(String key)这里我们官方demo入手，看getValue + smartSet 方法 是怎么实现上报数据的
/**
* 获取value，如果value不存在则发往netty
*/
public static Object getValue(String key, KeyType keyType) {
try {
//如果没有为该key配置规则，就不用上报key
if (!inRule(key)) {
return null;
}
Object userValue = null;
// 从本地缓存获取，是不是已经存在对应的值
ValueModel value = getValueSimple(key);
// 为空的话
if (value == null) {
// 上报访问数据
HotKeyPusher.push(key, keyType);
} else {
//不为空 临近过期了，也发
if (isNearExpire(value)) {
HotKeyPusher.push(key, keyType);
}
Object object = value.getValue();
//如果是默认值，也返回null
if (object instanceof Integer && Constant.MAGIC_NUMBER == (int) object) {
userValue = null;
} else {
userValue = object;
}
}
//统计计数
KeyHandlerFactory.getCounter().collect(new KeyHotModel(key, value != null));
return userValue;
} catch (Exception e) {
return null;
}
}判断有没有为该key配置规则，如果没有则直接返回
根据key从本地缓存获取对应的 ValueModel
- 如果为空，直接上报数据
- 不为空，且临近过期，也上报数据
- 不为空，构造返回值
统计计数，并返回结果值
下面看HotKeyPusher.push(key, keyType)，同时结合启动时开启的定时任务 PushSchedulerStarter.startPusher(pushPeriod) ，如何实现数据的实际推送
详细看代码注释 ，省略了部分这次分支无关的代码
// com.jd.platform.hotkey.client.core.key.HotKeyPusher#push
public static void push(String key, KeyType keyType) {
push(key, keyType, 1, false);
}
public static void push(String key, KeyType keyType, int count, boolean remove) {
// 数据校验跟设置初始值
....
// 这里使用 LongAdder 进行访问累计
LongAdder adderCnt = new LongAdder();
adderCnt.add(count);
// 构建一个keymodel,
HotKeyModel hotKeyModel = new HotKeyModel();
hotKeyModel.setAppName(Context.APP_NAME);
hotKeyModel.setKeyType(keyType);
hotKeyModel.setCount(adderCnt);
hotKeyModel.setRemove(remove);
hotKeyModel.setKey(key);
if (remove) {
......
} else {
//如果key是规则内的要被探测的key，就积累等待传送
if (KeyRuleHolder.isKeyInRule(key)) {
//TODO 重点看这里 积攒起来，等待每半秒发送一次
KeyHandlerFactory.getCollector().collect(hotKeyModel);
}
}
}
// KeyHandlerFactory.getCollector().collect(hotKeyModel); 跟踪 getCollector 往下，可以看到最终实现类为 TurnKeyCollector
private ConcurrentHashMap<String, HotKeyModel> map0 = new ConcurrentHashMap<>();
private ConcurrentHashMap<String, HotKeyModel> map1 = new ConcurrentHashMap<>();
private AtomicLong atomicLong = new AtomicLong(0);
// 累计需要上报key的计数 com.jd.platform.hotkey.client.core.key.TurnKeyCollector#collect
public void collect(HotKeyModel hotKeyModel) {
String key = hotKeyModel.getKey();
if (StrUtil.isEmpty(key)) {
return;
}
// 偶数时写入map0，奇数写入map1, 这里为什么要这么做?
if (atomicLong.get() % 2 == 0) {
//不存在时返回null并将key-value放入，已有相同key时，返回该key对应的value，并且不覆盖
HotKeyModel model = map0.putIfAbsent(key, hotKeyModel);
if (model != null) {
// 累计访问次数
model.add(hotKeyModel.getCount());
}
} else {
HotKeyModel model = map1.putIfAbsent(key, hotKeyModel);
if (model != null) {
model.add(hotKeyModel.getCount());
}
}
}
// 提供上报，获取对应需要上报的数据，主要是给下面的定时任务使用
public List<HotKeyModel> lockAndGetResult() {
//自增后，对应的map就会停止被写入，等待被读取
atomicLong.addAndGet(1);
List<HotKeyModel> list;
if (atomicLong.get() % 2 == 0) {
list = get(map1);
map1.clear();
} else {
list = get(map0);
map0.clear();
}
return list;
}
// 实际上报key的动作 com.jd.platform.hotkey.client.core.key.PushSchedulerStarter#startPusher
public static void startPusher(Long period) {
/ period 周期时间，默认为500毫秒
if (period == null || period <= 0) {
period = 500L;
}
// 开启定时任务
ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("hotkey-pusher-service-executor", true));
scheduledExecutorService.scheduleAtFixedRate(() -> {
// 获取 TurnKeyCollector 执行类
IKeyCollector<HotKeyModel, HotKeyModel> collectHK = KeyHandlerFactory.getCollector();
// 获取对应需要上报的数据
List<HotKeyModel> hotKeyModels = collectHK.lockAndGetResult();
// 如果不为空
if(CollectionUtil.isNotEmpty(hotKeyModels)){
// NettyKeyPusher
KeyHandlerFactory.getPusher().send(Context.APP_NAME, hotKeyModels);
// 预留扩展实现,
collectHK.finishOnce();
}
},0, period, TimeUnit.MILLISECONDS);
}
public void sendCount(String appName, List<KeyCountModel> list) {
long now = System.currentTimeMillis();
// 分组 key -> netty通道 , value -> 对应上报的key集合
Map<Channel, List<KeyCountModel>> map = new HashMap<>();
for(KeyCountModel model : list) {
model.setCreateTime(now);
// 从本地缓存的worker列表，选择需要发送的发送的worker节点的netty通道，hash轮询算法
Channel channel = WorkerInfoHolder.chooseChannel(model.getRuleKey());
if (channel == null) {
continue;
}
// 不存在和赋值为空集合，存在则取出对应分组的列表
List<KeyCountModel> newList = map.computeIfAbsent(channel, k -> new ArrayList<>());
// 新增key
newList.add(model);
}
// 遍历 并推送key的累计信息
for (Channel channel : map.keySet()) {
try {
List<KeyCountModel> batch = map.get(channel);
HotKeyMsg hotKeyMsg = new HotKeyMsg(MessageType.REQUEST_HIT_COUNT, Context.APP_NAME);
hotKeyMsg.setKeyCountModels(batch);
channel.writeAndFlush(hotKeyMsg).sync();
} catch (Exception e) {
try {
InetSocketAddress insocket = (InetSocketAddress) channel.remoteAddress();
JdLogger.error(getClass(),"flush error " + insocket.getAddress().getHostAddress());
} catch (Exception ex) {
JdLogger.error(getClass(),"flush error");
}
}
}
}在累计上报数据的时候，为什么会根据atomicLong进行判断，偶数时写入map0，奇数写入map1？
主要是为了定时任务实际上报数据的时候，不阻塞写入操作。试想一下，如果使用一个map集合进行存储的话，定时上传为了保证数据的准确定，肯定需要进行加锁，这样的话在上报频率快的时候，效率就会大大下降，对于一个高性能的探测框架来说，肯定是不允许的。
整体思路如下
4、接收热key的流程
应用初始化时，注册热key探测回调关注热key事件，同worker管理，也是先找对应发布变更事件的地方，
//热key探测回调关注热key事件
EventBusCenter.register(new ReceiveNewKeySubscribe());热key的数据来源
a、启动时，拉取全量规则后，获取已经存在的热key
private void fetchExistHotKey() {
JdLogger.info(getClass(), "--- begin fetch exist hotKey from etcd ----");
IConfigCenter configCenter = EtcdConfigFactory.configCenter();
try {
//获取所有热key
List<KeyValue> handKeyValues = configCenter.getPrefix(ConfigConstant.hotKeyPath + Context.APP_NAME);
for (KeyValue keyValue : handKeyValues) {
String key = keyValue.getKey().toStringUtf8().replace(ConfigConstant.hotKeyPath + Context.APP_NAME + "/", "");
HotKeyModel model = new HotKeyModel();
model.setRemove(false);
model.setKey(key);
// 组装后发布变更事件
EventBusCenter.getInstance().post(new ReceiveNewKeyEvent(model));
}
} catch (StatusRuntimeException ex) {
//etcd连不上
JdLogger.error(getClass(), "etcd connected fail. Check the etcd address!!!");
}
}b、异步监听 服务端手工添加和删除key
/**
* 异步开始监听热key变化信息，该目录里只有手工添加的key信息
* 开启异步线程，循环监听
*/
private void startWatchHotKey() {
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.submit(() -> {
JdLogger.info(getClass(), "--- begin watch hotKey change ----");
IConfigCenter configCenter = EtcdConfigFactory.configCenter();
try {
KvClient.WatchIterator watchIterator = configCenter.watchPrefix(ConfigConstant.hotKeyPath + Context.APP_NAME);
//如果有新事件，即新key产生或删除
while (watchIterator.hasNext()) {
WatchUpdate watchUpdate = watchIterator.next();
List<Event> eventList = watchUpdate.getEvents();
KeyValue keyValue = eventList.get(0).getKv();
Event.EventType eventType = eventList.get(0).getType();
try {
String key = keyValue.getKey().toStringUtf8().replace(ConfigConstant.hotKeyPath + Context.APP_NAME + "/", "");
//如果是删除key，就立刻删除
if (Event.EventType.DELETE == eventType) {
HotKeyModel model = new HotKeyModel();
model.setRemove(true);
model.setKey(key);
// 这里发布删除key事件
EventBusCenter.getInstance().post(new ReceiveNewKeyEvent(model));
} else {
HotKeyModel model = new HotKeyModel();
model.setRemove(false);
String value = keyValue.getValue().toStringUtf8();
//新增热key
JdLogger.info(getClass(), "etcd receive new key : " + key + " --value:" + value);
//如果这是一个删除指令，就什么也不干
if (Constant.DEFAULT_DELETE_VALUE.equals(value)) {
continue;
}
//手工创建的value是时间戳
model.setCreateTime(Long.valueOf(keyValue.getValue().toStringUtf8()));
model.setKey(key);
// 新增事件
EventBusCenter.getInstance().post(new ReceiveNewKeyEvent(model));
}
} catch (Exception e) {
JdLogger.error(getClass(), "new key err ：" + keyValue);
}
}
} catch (Exception e) {
JdLogger.error(getClass(), "watch err");
}
});
}c、worker端推送
netty处理，包含心跳 跟 新的key事件
com.jd.platform.hotkey.client.netty.NettyClientHandler#channelRead0
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, HotKeyMsg msg) {
if (MessageType.PONG == msg.getMessageType()) {
JdLogger.info(getClass(), "heart beat");
return;
}
if (MessageType.RESPONSE_NEW_KEY == msg.getMessageType()) {
JdLogger.info(getClass(), "receive new key : " + msg);
if (CollectionUtil.isEmpty(msg.getHotKeyModels())) {
return;
}
for (HotKeyModel model : msg.getHotKeyModels()) {
// 发布事件
EventBusCenter.getInstance().post(new ReceiveNewKeyEvent(model));
}
}
}怎么处理热key
订阅回调关注热key事件
public void newKeyComing(ReceiveNewKeyEvent event) {
HotKeyModel hotKeyModel = event.getModel();
if (hotKeyModel == null) {
return;
}
//收到新key推送
if (receiveNewKeyListener != null) {
// 实际执行逻辑, 本质上操作的还是 Caffeine
receiveNewKeyListener.newKey(hotKeyModel);
}
}可以看到，接收到的事件内容主要包含如下，所属appname，是否删除，类型，创建事件，key名称，和count
com.jd.platform.hotkey.client.callback.DefaultNewKeyListener#newKey
public void newKey(HotKeyModel hotKeyModel) {
long now = System.currentTimeMillis();
//如果key到达时已经过去1秒了，记录一下。手工删除key时，没有CreateTime
if (hotKeyModel.getCreateTime() != 0 && Math.abs(now - hotKeyModel.getCreateTime()) > 1000) {
JdLogger.warn(getClass(), "the key comes too late : " + hotKeyModel.getKey() + " now " +
+now + " keyCreateAt " + hotKeyModel.getCreateTime());
}
if (hotKeyModel.isRemove()) {
//如果是删除事件，就直接删除
deleteKey(hotKeyModel.getKey());
return;
}
//已经是热key了，又推过来同样的热key，做个日志记录，并刷新一下
if (JdHotKeyStore.isHot(hotKeyModel.getKey())) {
JdLogger.warn(getClass(), "receive repeat hot key ：" + hotKeyModel.getKey() + " at " + now);
}
addKey(hotKeyModel.getKey());
}
private void addKey(String key) {
ValueModel valueModel = ValueModel.defaultValue(key);
if (valueModel == null) {
//不符合任何规则
deleteKey(key);
return;
}
//如果原来该key已经存在了，那么value就被重置，过期时间也会被重置。如果原来不存在，就新增的热key
JdHotKeyStore.setValueDirectly(key, valueModel);
}
private void deleteKey(String key) {
CacheFactory.getNonNullCache(key).delete(key);
}这里变更通知的仅是key，那么对应的value是怎么赋值的？
回顾demo的使用案例。还需要执行一次原本的操作，获取实际值，才会对key进行赋值。
为什么不在上报的时候，直接上报对应的value？后续变更时也是直接推送即可？
1、时效性，多次变更不一样，还需要额外操作保证
2、存储成本
方法给热key赋值value，如果是热key，该方法才会赋值，非热key，什么也不做
void smartSet(String key, Object value)
public static void smartSet(String key, Object value) {
// 判断是否是热key，如果是热key，则给value赋值
if (isHot(key)) {
ValueModel valueModel = getValueSimple(key);
if (valueModel == null) {
return;
}
// 设置value值。
valueModel.setValue(value);
}
}三、worker计算节点
猜想一下，clent端初始是通过@PostConstruct注解，在bean初始化后做相应的逻辑处理，那么worker端是不是也是类似，通过搜索可以看到有多个逻辑初始化动作，接下来看看，worker节点是如何实现相应逻辑的。
1、如何感知变化规则变更
规则在dashboard端是存放到etcd的，那么直接看EtcdStarter的相应实现，有四个初始化动作，
根据代码可以看到 EtcdStarter#watch 方法是开启了 对 /jd/rules/ 路径下资源的监听，那么就是在该方法作为对应的入口。详细解析看下面代码注释
@PostConstruct
public void watch() {
// 开启一个异步线程
AsyncPool.asyncDo(() -> {
KvClient.WatchIterator watchIterator;
// 判断是否仅为指定app服务，根据启动时候的 application.yml 文件的workerPath配置
if (isForSingle()) {
// 是的话，读取指定app的规则
watchIterator = configCenter.watch(ConfigConstant.rulePath + workerPath);
} else {
// 读取全量的规则
watchIterator = configCenter.watchPrefix(ConfigConstant.rulePath);
}
while (watchIterator.hasNext()) {
WatchUpdate watchUpdate = watchIterator.next();
List<Event> eventList = watchUpdate.getEvents();
// 取出对应变化的规则
KeyValue keyValue = eventList.get(0).getKv();
logger.info("rule changed : " + keyValue);
try {
// 变更规则
ruleChange(keyValue);
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
// 变更规则的实际逻辑，实际就是放到本地缓存，ConcurrentHashMap
private synchronized void ruleChange(KeyValue keyValue) {
String appName = keyValue.getKey().toStringUtf8().replace(ConfigConstant.rulePath, "");
if (StrUtil.isEmpty(appName)) {
return;
}
String ruleJson = keyValue.getValue().toStringUtf8();
List<KeyRule> keyRules = FastJsonUtils.toList(ruleJson, KeyRule.class);
// 放到本地缓存
KeyRuleHolder.put(appName, keyRules);
}
/**
* key就是appName，value是rule
*/
private static final Map<String, List<KeyRule>> RULE_MAP = new ConcurrentHashMap<>();2、怎么接收Client上报的消息
client上报的时候是通过netty发送的，那么直接查看启动netty时配置的监听事件即可，代码入口 NodesServerStarter，
com.jd.platform.hotkey.worker.starters.NodesServerStarter#start
@PostConstruct
public void start() {
AsyncPool.asyncDo(() -> {
logger.info("netty server is starting");
NodesServer nodesServer = new NodesServer();
nodesServer.setClientChangeListener(iClientChangeListener);
nodesServer.setMessageFilters(messageFilters);
try {
// 启动netty服务端
nodesServer.startNettyServer(port);
} catch (Exception e) {
e.printStackTrace();
}
});
}
/// 启动netty
public void startNettyServer(int port) throws Exception {
//boss单线程
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(CpuNum.workerCount());
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.option(ChannelOption.SO_BACKLOG, 1024)
//保持长连接
.childOption(ChannelOption.SO_KEEPALIVE, true)
//出来网络io事件，如记录日志、对消息编解码等 重点看这里，绑定了处理的逻辑
.childHandler(new ChildChannelHandler());
.............
} catch (Exception e) {
.......
} finally {
........
}
}
// 消息接收后逻辑
private class ChildChannelHandler extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) {
NodesServerHandler serverHandler = new NodesServerHandler();
serverHandler.setClientEventListener(clientChangeListener);
serverHandler.addMessageFilters(messageFilters);
ByteBuf delimiter = Unpooled.copiedBuffer(Constant.DELIMITER.getBytes());
ch.pipeline()
// 拆包 基于指定字符串
.addLast(new DelimiterBasedFrameDecoder(Constant.MAX_LENGTH, delimiter))
// 消息解码编码
.addLast(new MsgDecoder())
.addLast(new MsgEncoder())
// 实际处理逻辑
.addLast(serverHandler);
}
}
com.jd.platform.hotkey.worker.netty.server.NodesServerHandler#channelRead0
protected void channelRead0(ChannelHandlerContext ctx, HotKeyMsg msg) {
if (msg == null) {
return;
}
// 遍历执行 过滤器 责任链模式
for (INettyMsgFilter messageFilter : messageFilters) {
boolean doNext = messageFilter.chain(msg, ctx);
if (!doNext) {
return;
}
}
}可以看到实现有多个，分别是appName相关，心跳包，热key，和 key数据统计的，这里我们只关注热key相关的HotKeyFilter
@Component
// 注入引用集合的排序，为第三
@Order(3)
public class HotKeyFilter implements INettyMsgFilter {
@Resource
private KeyProducer keyProducer;
public static AtomicLong totalReceiveKeyCount = new AtomicLong();
private Logger logger = LoggerFactory.getLogger(getClass());
@Override
public boolean chain(HotKeyMsg message, ChannelHandlerContext ctx) {
// 判断是否对应的事件类型，是的话责任链不再往下执行，其余Filter也是一样
if (MessageType.REQUEST_NEW_KEY == message.getMessageType()) {
// totalReceiveKeyCount 纪元 +1
totalReceiveKeyCount.incrementAndGet();
// 实际处理逻辑
publishMsg(message, ctx);
return false;
}
return true;
}
private void publishMsg(HotKeyMsg message, ChannelHandlerContext ctx) {
//老版的用的单个HotKeyModel，新版用的数组
List<HotKeyModel> models = message.getHotKeyModels();
long now = SystemClock.now();
if (CollectionUtil.isEmpty(models)) {
return;
}
for (HotKeyModel model : models) {
//白名单key不处理
if (WhiteListHolder.contains(model.getKey())) {
continue;
}
long timeOut = now - model.getCreateTime();
if (timeOut > 1000) {
if (EtcdStarter.LOGGER_ON) {
logger.info("key timeout " + timeOut + ", from ip : " + NettyIpUtil.clientIp(ctx));
}
}
// 放到队列里面
keyProducer.push(model, now);
}
}
}
// com.jd.platform.hotkey.worker.keydispatcher.KeyProducer#push
public void push(HotKeyModel model, long now) {
if (model == null || model.getKey() == null) {
return;
}
//5秒前的过时消息就不处理了
if (now - model.getCreateTime() > InitConstant.timeOut) {
expireTotalCount.increment();
return;
}
try {
QUEUE.put(model);
totalOfferCount.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 最总放到一个阻塞队列里面
public static BlockingQueue<HotKeyModel> QUEUE = new LinkedBlockingQueue<>(2000000);3、如何探测为热key并推送
从第二步知道，上报的key数据时存放到 LinkedBlockingQueue 阻塞队列里面的，那么消费的时候，肯定是从这里取值，那么直接搜索 QUEUE.take() 就能找到对应处理的逻辑在哪里，详细看下面逻辑，涉及到相应的滑动窗口计算方法可以自行了解，源码已经写好详细的注释
public class KeyConsumer {
private IKeyListener iKeyListener;
public void setKeyListener(IKeyListener iKeyListener) {
this.iKeyListener = iKeyListener;
}
public void beginConsume() {
// 死循环处理
while (true) {
try {
// 这里从队列取出，没有值的时候阻塞
HotKeyModel model = QUEUE.take();
// 然后判断时删除，还是新增
if (model.isRemove()) {
iKeyListener.removeKey(model, KeyEventOriginal.CLIENT);
} else {
// 新增，实际看这里的逻辑
iKeyListener.newKey(model, KeyEventOriginal.CLIENT);
}
//处理完毕，将数量加1
totalDealCount.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
// 处理新key
public void newKey(HotKeyModel hotKeyModel, KeyEventOriginal original) {
//构造cache里的key
String key = buildKey(hotKeyModel);
//如果在缓存存在，说明已经是热key 不进行后续处理
Object o = hotCache.getIfPresent(key);
if (o != null) {
return;
}
// 获取对应滑动窗口
SlidingWindow slidingWindow = checkWindow(hotKeyModel, key);
// 在滑动窗口内，看看hot没
boolean hot = slidingWindow.addCount(hotKeyModel.getCount());
if (!hot) {
//如果没hot，重新put，cache会自动刷新过期时间
CaffeineCacheHolder.getCache(hotKeyModel.getAppName()).put(key, slidingWindow);
} else {
// 已经时热key后，添加到缓存中
hotCache.put(key, 1);
//删掉该key
CaffeineCacheHolder.getCache(hotKeyModel.getAppName()).invalidate(key);
//开启推送
hotKeyModel.setCreateTime(SystemClock.now());
//当开关打开时，打印日志。大促时关闭日志，就不打印了
if (EtcdStarter.LOGGER_ON) {
logger.info(NEW_KEY_EVENT + hotKeyModel.getKey());
}
//分别推送到各client和etcd
for (IPusher pusher : iPushers) {
// 其实也是放到阻塞队列，
pusher.push(hotKeyModel);
}
}
}
//com.jd.platform.hotkey.worker.netty.pusher.AppServerPusher#push
public void push(HotKeyModel model) {
hotKeyStoreQueue.offer(model);
}
// AppServerPusher 开头说明的 @PostConstruct 注解
@PostConstruct
public void batchPushToClient() {
AsyncPool.asyncDo(() -> {
while (true) {
try {
List<HotKeyModel> tempModels = new ArrayList<>();
//每10ms推送一次 从待推送队列取出具体需要推送的key
Queues.drain(hotKeyStoreQueue, tempModels, 10, 10, TimeUnit.MILLISECONDS);
if (CollectionUtil.isEmpty(tempModels)) {
continue;
}
Map<String, List<HotKeyModel>> allAppHotKeyModels = new HashMap<>();
//拆分出每个app的热key集合，按app分堆
for (HotKeyModel hotKeyModel : tempModels) {
List<HotKeyModel> oneAppModels = allAppHotKeyModels.computeIfAbsent(hotKeyModel.getAppName(), (key) -> new ArrayList<>());
oneAppModels.add(hotKeyModel);
}
//遍历所有app，进行推送
for (AppInfo appInfo : ClientInfoHolder.apps) {
List<HotKeyModel> list = allAppHotKeyModels.get(appInfo.getAppName());
if (CollectionUtil.isEmpty(list)) {
continue;
}
HotKeyMsg hotKeyMsg = new HotKeyMsg(MessageType.RESPONSE_NEW_KEY);
hotKeyMsg.setHotKeyModels(list);
//整个app全部发送，通过netty
appInfo.groupPush(hotKeyMsg);
}
allAppHotKeyModels = null;
} catch (Exception e) {
e.printStackTrace();
}
}
});
}上面这段代码存在一个问题，官方已经明显写出注释，所以这里直接贴出来。
//********** watch here ************//
//该方法会被InitConstant.threadCount个线程同时调用，存在多线程问题
//下面的那句addCount是加了锁的，代表给Key累加数量时是原子性的，不会发生多加、少加的情况，到了设定的阈值一定会hot
//譬如阈值是2，如果多个线程累加，在没hot前，hot的状态肯定是对的，譬如thread1 加1，thread2加1，那么thread2会hot返回true，开启推送
//但是极端情况下，譬如阈值是10，当前是9，thread1走到这里时，加1，返回true，thread2也走到这里，加1，此时是11，返回true，问题来了
//该key会走下面的else两次，也就是2次推送。
//所以出现问题的原因是hotCache.getIfPresent(key)这一句在并发情况下，没return掉，放了两个key+1到addCount这一步时，会有问题
//测试代码在TestBlockQueue类，直接运行可以看到会同时hot
//那么该问题用解决吗，NO，不需要解决，1 首先要发生的条件极其苛刻，很难触发，以京东这样高的并发量，线上我也没见过触发连续2次推送同一个key的
//2 即便触发了，后果也是可以接受的，2次推送而已，毫无影响，客户端无感知。但是如果非要解决，就要对slidingWindow实例加锁了，必然有一些开销
//所以只要保证key数量不多计算就可以，少计算了没事。因为热key必然频率高，漏计几次没事。但非热key，多计算了，被干成了热key就不对了四、总结
dashboard 是标准的web项目，通过controller入口，保存规则到etcd
client在初始化的时候，会开启各项定时任务，包含获取规则，worker节点信息保存到本地缓存
- 通过wokrker信息，创建netty客户端跟worker节点建立连接
- 使用对应的api时会根据热key规则进行累计
- 定时通过netty推送统计数据到worker节点进行计算
worker初始化时也会开启各项定时任务，获取规则，跟启动netty服务端，等待client的连接
- 监听client的netty事件，进行心跳，appname，热key计算，统计等处理
- 根据时间滑动窗口算法，判定为热key，则推送对应的数据给对应的client端
上面结合各个端介绍了热key的规则配置跟怎么统计生效，部分细节可能描述的不是很准确。可以自行下载官方源码查看，其实很多细节官方已经标明了注释，最后感谢各位看到这里。
原文链接: https://juejin.cn/post/7353456468094287909