序
本文主要研究一下PowerJob Worker的ServerAddress
PowerJobAutoConfiguration
tech/powerjob/worker/autoconfigure/PowerJobAutoConfiguration.java
代码语言：javascript复制 @Bean
@ConditionalOnMissingBean
public 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
代码语言：javascript复制 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配置地址的时候，多个地址给随机化配置，不要按固定顺序。
