手写分布式配置中心（四）增加实时刷新功能（长轮询）

上一篇文章中实现了短轮询，不过短轮询的弊端也很明显，如果请求的频率较高，那么就会导致服务端压力大（并发高）；如果请求的频率放低，那么客户端感知变更的及时性就会降低。所以我们来看另一种轮询方式，长轮询。
长轮询就是客户端发起请求，如果服务端的数据没有发生变更，那么就hold住请求，直到服务端的数据发生了变更，或者达到了一定的时间就会返回。这样就减少了客户端和服务端不断频繁连接和传递数据的过程，并且不会消耗服务端太多资源，而且客户端感知变更的及时性也会大大提高

代码在https://gitee.com/summer-cat001/config-center​​​​​​​

原理

要实现服务端长时间hold请求，就要利用到servlet异步的特性，因为web服务器会有一个线程池，每一个请求来了之后会提交给这个线程池去处理请求，如果一个任务很长时间都没完成的话就会一直占有这个线程，那么其他请求来了会发现线程池里没有可用的线程就会一直等，直到有空闲的线程，这样就会导致并发性大大的减少。所以需要采用异步响应的方式去实现，而比较方便实现异步http的方式就是Servlet3.0提供的AsyncContext 机制。asyncContext是为了把主线程返回给web服务器的线程池，不影响服务对其他客户端请求。会有线程专门处理这个长轮询，但并不是说每一个长轮询的http请求都要用一个线程阻塞在那。而是把长轮询的request的引用在一个集合中存起来，用一个或几个线程专门处理一批客户端的长轮询请求，这样就不需要为每一个长轮询单独分配线程阻塞在那了，从而大大降低了资源的消耗。注意，异步不是非阻塞，响应数据时还是要阻塞的。

服务端

服务端增加一个长轮询的接口

@PostMapping("/change/get/long")public Result<Void> getLongChangeConfig(@RequestBody Map<Long, Integer> configIdMap, HttpServletRequest request, HttpServletResponse response) {if (configIdMap == null || configIdMap.isEmpty()) {return Result.fail("配置参数错误");}response.setContentType("application/json;charset=UTF-8");AsyncContext asyncContext = request.startAsync();asyncContext.setTimeout(0);ConfigPolingTask configPolingTask = new ConfigPolingTask();configPolingTask.setAsyncContext(asyncContext);configPolingTask.setConfigPolingDataMap(configIdMap);configPolingTask.setEndTime(System.currentTimeMillis() + 28 * 1000);configService.configListener(configPolingTask);return null;}

主要就是把请求的配置id和版本的map、超时时间、asyncContext对象组装成一个任务，添加到任务池里，如有更新了配置，会去任务池里找是否有该配置id的任务，如果版本号大于任务的版本号，就将新配置返回给客户端。于此同时会有1个定时线程每1秒访问一下任务池，找到过期的任务，返回给客户端。客户端的请求过期时间是30秒，服务端过期时间定的是28秒，也就是配置没有改变的情况下，会hold请求28秒才返回，提前2秒返回是为了防止返回传输时间导致超过30秒，客户端断开链接。
 

@Slf4j
@Service
public class ConfigServiceImpl implements ConfigService {private ConfigDAO configDAO;private ConfigSyncService configSyncService;@Autowiredprivate LocalConfigDAO localConfigDAO;@Autowiredprivate LocalConfigSyncServiceImpl localConfigSyncService;@Value("${config.center.mode:0}")private int configCenterMode;private int respThreadNum;private final ExecutorService respExecutor;private final ConfigPolingTasksHolder configPolingTasksHolder;public ConfigServiceImpl() {configPolingTasksHolder = new ConfigPolingTasksHolder();//构建用于响应长轮询的线程池respExecutor = new ThreadPoolExecutor(100, 5000,0, TimeUnit.SECONDS,new ArrayBlockingQueue<>(102400),this::newRespThread,new ThreadPoolExecutor.CallerRunsPolicy());//每1秒轮询执行一次任务超时检测ScheduledExecutorService timeoutCheckExecutor = new ScheduledThreadPoolExecutor(1, this::newCheckThread);timeoutCheckExecutor.scheduleAtFixedRate(this::responseTimeoutTask, 0, 1, TimeUnit.SECONDS);}@PostConstructpublic void init() {ConfigCenterModeEnum configCenterModeEnum = ConfigCenterModeEnum.getEnum(configCenterMode);if (configCenterModeEnum == null) {throw new IllegalArgumentException("配置config.center.mode错误");}if (configCenterModeEnum == ConfigCenterModeEnum.STANDALONE) {this.configDAO = localConfigDAO;this.configSyncService = localConfigSyncService;}}@Overridepublic Result<Void> insertConfig(ConfigBO configBO) {List<ConfigDO> configList = configDAO.getAllValidConfig();if (configList.stream().anyMatch(c -> c.getName().equals(configBO.getName()))) {return Result.fail("配置名重复");}ConfigDO configDO = new ConfigDO();configDO.setName(configBO.getName());configDO.setConfigData(configBO.getConfigData().toJSONString());configDAO.insertConfigDO(configDO);return Result.success(null);}@Overridepublic Result<Void> updateConfig(ConfigBO configBO) {ConfigDO configDO = new ConfigDO();configDO.setId(configBO.getId());configDO.setName(configBO.getName());configDO.setConfigData(configBO.getConfigData().toJSONString());configDAO.updateConfig(configDO);configSyncService.publish(configBO.getId());return Result.success(null);}@Overridepublic Result<Void> delConfig(long id, long updateUid) {configDAO.delConfig(id, updateUid);return Result.success(null);}@Overridepublic Result<List<ConfigBO>> getAllValidConfig() {List<ConfigDO> configList = configDAO.getAllValidConfig();return Result.success(configList.stream().map(this::ConfigDO2BO).collect(Collectors.toList()));}@Overridepublic void configListener(ConfigPolingTask configPolingTask) {//先将任务加到待响应列表中，然后再判断账号是否有改变，防止并发问题//如先判断再加进去，加入前如有变动，任务里无法感知到，空等到超时configPolingTasksHolder.addConfigTask(configPolingTask);List<ConfigBO> allValidConfig = getAllValidConfig().getData();List<ConfigVO> changeConfigList = getChangeConfigList(configPolingTask, allValidConfig);if (!changeConfigList.isEmpty()) {List<ConfigPolingTask> todoTask = configPolingTasksHolder.getExecuteTaskList(configPolingTask::equals);if (!todoTask.isEmpty()) {doResponseTask(configPolingTask, Result.success(changeConfigList));}}}@Overridepublic void onChangeConfigEvent(long configId) {List<ConfigPolingTask> todoTasks = configPolingTasksHolder.getExecuteTaskList(configPolingTask -> configPolingTask.getConfigPolingDataMap().containsKey(configId));if (!todoTasks.isEmpty()) {List<ConfigBO> configList = Collections.singletonList(ConfigDO2BO(configDAO.getConfig(configId)));todoTasks.forEach(todoTask -> {List<ConfigVO> changeConfigList = getChangeConfigList(todoTask, configList);respExecutor.submit(() -> doResponseTask(todoTask, Result.success(changeConfigList)));});}}private List<ConfigVO> getChangeConfigList(ConfigPolingTask configPolingTask, List<ConfigBO> configList) {Map<Long, Integer> configPolingDataMap = configPolingTask.getConfigPolingDataMap();return configList.stream().filter(configBO -> configPolingDataMap.containsKey(configBO.getId())).filter(configBO -> configBO.getVersion() > configPolingDataMap.get(configBO.getId())).map(ConfigServiceImpl::configBO2ConfigVO).collect(Collectors.toList());}private ConfigBO ConfigDO2BO(ConfigDO configDO) {ConfigBO configBO = new ConfigBO();configBO.setId(configDO.getId());configBO.setName(configDO.getName());configBO.setVersion(configDO.getVersion());configBO.setCreateTime(configDO.getCreateTime());configBO.setConfigData(JSON.parseObject(configDO.getConfigData()));return configBO;}//响应超时未改变的任务private void responseTimeoutTask() {List<ConfigPolingTask> timeoutTasks = configPolingTasksHolder.getExecuteTaskList(configPolingTask -> System.currentTimeMillis() >= configPolingTask.getEndTime());timeoutTasks.forEach(timeoutTask -> respExecutor.submit(() ->doResponseTask(timeoutTask, Result.success(new ArrayList<>()))));}private void doResponseTask(ConfigPolingTask configPolingTask, Result<?> result) {AsyncContext asyncContext = configPolingTask.getAsyncContext();try (PrintWriter writer = asyncContext.getResponse().getWriter()) {writer.write(JSON.toJSONString(result));writer.flush();} catch (Exception e) {log.error("doResponseTimeoutTask error,task:{}", configPolingTask, e);} finally {asyncContext.complete();}}private Thread newCheckThread(Runnable r) {Thread t = new Thread(r);t.setDaemon(true);t.setName("ConfigLongPollingTimeoutCheckExecutor");return t;}private Thread newRespThread(Runnable r) {Thread t = new Thread(r);t.setDaemon(true);t.setName("ConfigLongPollingTimeoutRespExecutor-" + respThreadNum++);return t;}public static ConfigVO configBO2ConfigVO(ConfigBO configBO) {ConfigVO configVO = new ConfigVO();configVO.setId(configBO.getId());configVO.setName(configBO.getName());configVO.setVersion(configBO.getVersion());configVO.setConfigData(configBO.getConfigData());configVO.setCreateTime(DateUtil.date2str1(configBO.getCreateTime()));return configVO;}
}

public class ConfigPolingTasksHolder {private final List<ConfigPolingTask> configPolingTasks;public ConfigPolingTasksHolder() {configPolingTasks = new ArrayList<>();}public synchronized void addConfigTask(ConfigPolingTask configPolingTask) {configPolingTasks.add(configPolingTask);}//将要处理的任务在任务列表中删除，并将其放到外面执行，防止锁的时间太长public synchronized List<ConfigPolingTask> getExecuteTaskList(Predicate<ConfigPolingTask> predicate) {List<ConfigPolingTask> resultTasks = new ArrayList<>();configPolingTasks.removeIf(configPolingTask -> {boolean res = predicate.test(configPolingTask);if (res) {resultTasks.add(configPolingTask);}return res;});return resultTasks;}
}

@Data
public class ConfigPolingTask {/*** 截止时间*/private long endTime;/*** 异步请求*/private AsyncContext asyncContext;/*** 配置轮询数据（配置id，版本）*/private Map<Long, Integer> configPolingDataMap;
}

客户端

客户端就很简单了，只要循环发一个超时时间是30秒的http请求就行

public void startLongPolling() {polling("/config/change/get/long", null, 30000);}public void polling(String uri, Runnable runnable, int readTimeout) {Thread thread = new Thread(() -> {while (!Thread.interrupted()) {try {Optional.ofNullable(runnable).ifPresent(Runnable::run);Map<Long, List<ConfigDataBO>> refreshConfigMap = new HashMap<>();configMap.values().forEach(configBO -> {Optional.ofNullable(configBO.getConfigDataList()).ifPresent(cdList -> cdList.stream().filter(cd -> cd.getRefreshFieldList() != null && !cd.getRefreshFieldList().isEmpty()).forEach(refreshConfigMap.computeIfAbsent(configBO.getId(), k1 -> new ArrayList<>())::add));});if (refreshConfigMap.isEmpty()) {return;}Map<String, Integer> configIdMap = refreshConfigMap.keySet().stream().collect(Collectors.toMap(String::valueOf, configId -> configMap.get(configId).getVersion()));HttpRespBO httpRespBO = HttpUtil.httpPostJson(url + uri, JSON.toJSONString(configIdMap), readTimeout);List<ConfigVO> configList = httpResp2ConfigVOList(httpRespBO);if (configList.isEmpty()) {continue;}configList.forEach(configVO -> {Map<String, Object> result = new HashMap<>();DataTransUtil.buildFlattenedMap(result, configVO.getConfigData(), "");ConfigBO configBO = this.configMap.get(configVO.getId());configBO.setVersion(configVO.getVersion());List<ConfigDataBO> configDataList = configBO.getConfigDataList();Map<String, ConfigDataBO> configDataMap = configDataList.stream().collect(Collectors.toMap(ConfigDataBO::getKey, Function.identity()));result.forEach((key, value) -> {ConfigDataBO configDataBO = configDataMap.get(key);if (configDataBO == null) {configDataList.add(new ConfigDataBO(key, value.toString()));} else {configDataBO.setValue(value.toString());List<RefreshFieldBO> refreshFieldList = configDataBO.getRefreshFieldList();if (refreshFieldList == null) {refreshFieldList = new ArrayList<>();configDataBO.setRefreshFieldList(refreshFieldList);}refreshFieldList.forEach(refreshFieldBO -> {try {Field field = refreshFieldBO.getField();field.setAccessible(true);field.set(refreshFieldBO.getBean(), value.toString());} catch (Exception e) {log.error("startShortPolling set Field error", e);}});}});});} catch (Exception e) {log.error("startShortPolling error", e);}}});thread.setName("startShortPolling");thread.setDaemon(true);thread.start();}private List<ConfigVO> httpResp2ConfigVOList(HttpRespBO httpRespBO) {if (!httpRespBO.success()) {throw new IllegalArgumentException("获取配置失败：code:" + httpRespBO.getCode() + ",msg:" + httpRespBO.getMessage());}if (httpRespBO.getBody() == null) {throw new IllegalArgumentException("获取配置失败 body is null：code:" + httpRespBO.getCode() + ",msg:" + httpRespBO.getMessage());}Result<?> result = JSON.parseObject(new String(httpRespBO.getBody(), StandardCharsets.UTF_8), Result.class);if (result.failed()) {throw new IllegalArgumentException("获取配置失败 result:" + result);}return JSON.parseArray(JSON.toJSONString(result.getData()), ConfigVO.class);}

public class ClientTest {private String userName;private String userAge;private List<Object> education;public ClientTest() throws NoSuchFieldException {ConfigCenterClient configCenterClient = new ConfigCenterClient("http://localhost:8088");Map<String, String> configProperty = configCenterClient.getConfigProperty();this.userName = configProperty.get("user.name");this.userAge = configProperty.get("user.age");this.education = new ArrayList<>();int i = 0;while (configProperty.containsKey("user.education[" + i + "]")) {education.add(configProperty.get("user.education[" + (i++) + "]"));}configCenterClient.addRefreshField("user.name", new RefreshFieldBO(this, ClientTest.class.getDeclaredField("userName")));configCenterClient.startLongPolling();}public String toString() {return "姓名:" + userName + ",年龄:" + userAge + ",教育经历:" + education;}public static void main(String[] args) throws NoSuchFieldException, InterruptedException {ClientTest clientTest = new ClientTest();while (!Thread.interrupted()) {System.out.println(clientTest);Thread.sleep(1000);}}
}

效果