本文主要是介绍聊聊PowerJob Worker的ServerAddress,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
序
本文主要研究一下PowerJob Worker的ServerAddress
PowerJobAutoConfiguration
tech/powerjob/worker/autoconfigure/PowerJobAutoConfiguration.java
@Bean@ConditionalOnMissingBeanpublic PowerJobSpringWorker initPowerJob(PowerJobProperties properties) {PowerJobProperties.Worker worker = properties.getWorker();/** Address of PowerJob-server node(s). Do not mistake for ActorSystem port. Do not add* any prefix, i.e. http://.*/CommonUtils.requireNonNull(worker.getServerAddress(), "serverAddress can't be empty! " +"if you don't want to enable powerjob, please config program arguments: powerjob.worker.enabled=false");List<String> serverAddress = Arrays.asList(worker.getServerAddress().split(","));/** Create OhMyConfig object for setting properties.*/PowerJobWorkerConfig config = new PowerJobWorkerConfig();/** Configuration of worker port. Random port is enabled when port is set with non-positive number.*/if (worker.getPort() != null) {config.setPort(worker.getPort());} else {int port = worker.getAkkaPort();if (port <= 0) {port = NetUtils.getRandomPort();}config.setPort(port);}/** appName, name of the application. Applications should be registered in advance to prevent* error. This property should be the same with what you entered for appName when getting* registered.*/config.setAppName(worker.getAppName());config.setServerAddress(serverAddress);config.setProtocol(worker.getProtocol());/** For non-Map/MapReduce tasks, {@code memory} is recommended for speeding up calculation.* Map/MapReduce tasks may produce batches of subtasks, which could lead to OutOfMemory* exception or error, {@code disk} should be applied.*/config.setStoreStrategy(worker.getStoreStrategy());/** When enabledTestMode is set as true, PowerJob-worker no longer connects to PowerJob-server* or validate appName.*/config.setAllowLazyConnectServer(worker.isAllowLazyConnectServer());/** Max length of appended workflow context . Appended workflow context value that is longer than the value will be ignored.*/config.setMaxAppendedWfContextLength(worker.getMaxAppendedWfContextLength());config.setTag(worker.getTag());config.setMaxHeavyweightTaskNum(worker.getMaxHeavyweightTaskNum());config.setMaxLightweightTaskNum(worker.getMaxLightweightTaskNum());config.setHealthReportInterval(worker.getHealthReportInterval());/** Create PowerJobSpringWorker object and set properties.*/return new PowerJobSpringWorker(config);}
PowerJobAutoConfiguration读取powerjob.worker.server-address属性,解析逗号分隔,赋值给PowerJobWorkerConfig的serverAddress
PowerJobServerDiscoveryService
tech/powerjob/worker/background/discovery/PowerJobServerDiscoveryService.java
private String discovery() {// 只有允许延迟加载模式下,appId 才可能为空。每次服务发现前,都重新尝试获取 appInfo。由于是懒加载链路,此处完全忽略异常if (appInfo.getAppId() == null || appInfo.getAppId() < 0) {try {assertApp0();} catch (Exception e) {log.warn("[PowerDiscovery] assertAppName in discovery stage failed, msg: {}", e.getMessage());return null;}}if (ip2Address.isEmpty()) {config.getServerAddress().forEach(x -> ip2Address.put(x.split(":")[0], x));}String result = null;// 先对当前机器发起请求String currentServer = currentServerAddress;if (!StringUtils.isEmpty(currentServer)) {String ip = currentServer.split(":")[0];// 直接请求当前Server的HTTP服务,可以少一次网络开销,减轻Server负担String firstServerAddress = ip2Address.get(ip);if (firstServerAddress != null) {result = acquire(firstServerAddress);}}for (String httpServerAddress : config.getServerAddress()) {if (StringUtils.isEmpty(result)) {result = acquire(httpServerAddress);}else {break;}}if (StringUtils.isEmpty(result)) {log.warn("[PowerDiscovery] can't find any available server, this worker has been quarantined.");// 在 Server 高可用的前提下,连续失败多次,说明该节点与外界失联,Server已经将秒级任务转移到其他Worker,需要杀死本地的任务if (FAILED_COUNT++ > MAX_FAILED_COUNT) {log.warn("[PowerDiscovery] can't find any available server for 3 consecutive times, It's time to kill all frequent job in this worker.");List<Long> frequentInstanceIds = HeavyTaskTrackerManager.getAllFrequentTaskTrackerKeys();if (!CollectionUtils.isEmpty(frequentInstanceIds)) {frequentInstanceIds.forEach(instanceId -> {HeavyTaskTracker taskTracker = HeavyTaskTrackerManager.removeTaskTracker(instanceId);taskTracker.destroy();log.warn("[PowerDiscovery] kill frequent instance(instanceId={}) due to can't find any available server.", instanceId);});}FAILED_COUNT = 0;}return null;} else {// 重置失败次数FAILED_COUNT = 0;log.debug("[PowerDiscovery] current server is {}.", result);return result;}}
PowerJobServerDiscoveryService的discovery会遍历config.getServerAddress(),执行acquire(httpServerAddress),请求服务端获取该appName所负责的server地址,若获取不到则继续循环
小结
PowerJob的worker需要配置powerjob.worker.server-address属性,它可以配置多个地址和port,但是默认请求discovery的时候是按顺序遍历下来,这个后续可以优化为随机。另外请求server端获取该appName的currentServer的时候,也在一定程度上依赖了本机地址去负责该请求的worker,这个也是跟配置顺序有关系。官方是建议配置域名,通过域名负载均衡,或者是给不同app配置地址的时候,多个地址给随机化配置,不要按固定顺序。
这篇关于聊聊PowerJob Worker的ServerAddress的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!