网站建设及运营服务流程,北京企业建站服务中企,省建设厅官网,17一起做网站童装上一篇文章中实现了短轮询#xff0c;不过短轮询的弊端也很明显#xff0c;如果请求的频率较高#xff0c;那么就会导致服务端压力大#xff08;并发高#xff09;#xff1b;如果请求的频率放低#xff0c;那么客户端感知变更的及时性就会降低。所以我们来看另一种轮询…上一篇文章中实现了短轮询不过短轮询的弊端也很明显如果请求的频率较高那么就会导致服务端压力大并发高如果请求的频率放低那么客户端感知变更的及时性就会降低。所以我们来看另一种轮询方式长轮询。 长轮询就是客户端发起请求如果服务端的数据没有发生变更那么就hold住请求直到服务端的数据发生了变更或者达到了一定的时间就会返回。这样就减少了客户端和服务端不断频繁连接和传递数据的过程并且不会消耗服务端太多资源而且客户端感知变更的及时性也会大大提高
代码在https://gitee.com/summer-cat001/config-center
原理
要实现服务端长时间hold请求就要利用到servlet异步的特性因为web服务器会有一个线程池每一个请求来了之后会提交给这个线程池去处理请求如果一个任务很长时间都没完成的话就会一直占有这个线程那么其他请求来了会发现线程池里没有可用的线程就会一直等直到有空闲的线程这样就会导致并发性大大的减少。所以需要采用异步响应的方式去实现而比较方便实现异步http的方式就是Servlet3.0提供的AsyncContext 机制。asyncContext是为了把主线程返回给web服务器的线程池不影响服务对其他客户端请求。会有线程专门处理这个长轮询但并不是说每一个长轮询的http请求都要用一个线程阻塞在那。而是把长轮询的request的引用在一个集合中存起来用一个或几个线程专门处理一批客户端的长轮询请求这样就不需要为每一个长轮询单独分配线程阻塞在那了从而大大降低了资源的消耗。注意异步不是非阻塞响应数据时还是要阻塞的。
服务端
服务端增加一个长轮询的接口
PostMapping(/change/get/long)public ResultVoid getLongChangeConfig(RequestBody MapLong, Integer configIdMap, HttpServletRequest request, HttpServletResponse response) {if (configIdMap null || configIdMap.isEmpty()) {return Result.fail(配置参数错误);}response.setContentType(application/json;charsetUTF-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 ResultVoid insertConfig(ConfigBO configBO) {ListConfigDO 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 ResultVoid 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 ResultVoid delConfig(long id, long updateUid) {configDAO.delConfig(id, updateUid);return Result.success(null);}Overridepublic ResultListConfigBO getAllValidConfig() {ListConfigDO configList configDAO.getAllValidConfig();return Result.success(configList.stream().map(this::ConfigDO2BO).collect(Collectors.toList()));}Overridepublic void configListener(ConfigPolingTask configPolingTask) {//先将任务加到待响应列表中然后再判断账号是否有改变防止并发问题//如先判断再加进去加入前如有变动任务里无法感知到空等到超时configPolingTasksHolder.addConfigTask(configPolingTask);ListConfigBO allValidConfig getAllValidConfig().getData();ListConfigVO changeConfigList getChangeConfigList(configPolingTask, allValidConfig);if (!changeConfigList.isEmpty()) {ListConfigPolingTask todoTask configPolingTasksHolder.getExecuteTaskList(configPolingTask::equals);if (!todoTask.isEmpty()) {doResponseTask(configPolingTask, Result.success(changeConfigList));}}}Overridepublic void onChangeConfigEvent(long configId) {ListConfigPolingTask todoTasks configPolingTasksHolder.getExecuteTaskList(configPolingTask - configPolingTask.getConfigPolingDataMap().containsKey(configId));if (!todoTasks.isEmpty()) {ListConfigBO configList Collections.singletonList(ConfigDO2BO(configDAO.getConfig(configId)));todoTasks.forEach(todoTask - {ListConfigVO changeConfigList getChangeConfigList(todoTask, configList);respExecutor.submit(() - doResponseTask(todoTask, Result.success(changeConfigList)));});}}private ListConfigVO getChangeConfigList(ConfigPolingTask configPolingTask, ListConfigBO configList) {MapLong, 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() {ListConfigPolingTask 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 ListConfigPolingTask configPolingTasks;public ConfigPolingTasksHolder() {configPolingTasks new ArrayList();}public synchronized void addConfigTask(ConfigPolingTask configPolingTask) {configPolingTasks.add(configPolingTask);}//将要处理的任务在任务列表中删除并将其放到外面执行防止锁的时间太长public synchronized ListConfigPolingTask getExecuteTaskList(PredicateConfigPolingTask predicate) {ListConfigPolingTask 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 MapLong, 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);MapLong, ListConfigDataBO 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;}MapString, 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);ListConfigVO configList httpResp2ConfigVOList(httpRespBO);if (configList.isEmpty()) {continue;}configList.forEach(configVO - {MapString, Object result new HashMap();DataTransUtil.buildFlattenedMap(result, configVO.getConfigData(), );ConfigBO configBO this.configMap.get(configVO.getId());configBO.setVersion(configVO.getVersion());ListConfigDataBO configDataList configBO.getConfigDataList();MapString, 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());ListRefreshFieldBO 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 ListConfigVO httpResp2ConfigVOList(HttpRespBO httpRespBO) {if (!httpRespBO.success()) {throw new IllegalArgumentException(获取配置失败code: httpRespBO.getCode() ,msg: httpRespBO.getMessage());}if (httpRespBO.getBody() null) {throw new IllegalArgumentException(获取配置失败 body is nullcode: 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 ListObject education;public ClientTest() throws NoSuchFieldException {ConfigCenterClient configCenterClient new ConfigCenterClient(http://localhost:8088);MapString, 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);}}
}
效果
