北京网站建设qq群,工作牌,全球十大猎头公司排名,网站结构 扁平结构 树状结构这是一篇让你受益匪浅的文章#xff0c;代码即使人生。 worker启动比server启动要复杂一些#xff0c;毕竟worker是要实际干活的#xff0c;工欲善其事必先利其器#xff0c;所以需要准备的工具还是不能少的#xff0c;server对于powerjob来说#xff0c;只是一个调度用的… 这是一篇让你受益匪浅的文章代码即使人生。 worker启动比server启动要复杂一些毕竟worker是要实际干活的工欲善其事必先利其器所以需要准备的工具还是不能少的server对于powerjob来说只是一个调度用的说白了就是管理worker做什么工作的只需要给他一个流程让他按照流程上的内容一次告诉worker去工作至于怎么做只有worker知道server当然不会知道的也没有必要知道。
worker的启动大概分为以下这么几个步骤 判断是否重复初始化 获取默认配置 校验appName 获取IP地址和端口这一步和server端是一样的在这里就不赘述了 初始化定时线程 连接server 初始化Akka 初始化日志系统 初始化存储 初始化定时任务
步骤是蛮多的但是其实都不是非常的复杂 由于worker的启动源码过于多了就不全贴出来了。
开胃菜
首先因为该worker包是需要被依赖的所以并没有spring的启动类但是却有启动spring时添加其配置的内容在worker包里面的PowerJobWorker类是实现了ApplicationContextAware, InitializingBean, DisposableBean这三个类这三个类默认有三个方法分别是 public void setApplicationContext(ApplicationContext applicationContext) throws BeansException public void afterPropertiesSet() throws Exception public void destroy() throws Exception 99.9%的初始化工作都是在afterPropertiesSet这个方法里完成的看名字大概也能猜出这个方法的意思就是字面意思。 判断是否重复初始化
if (!initialized.compareAndSet(false, true)) {log.warn([PowerJobWorker] please do not repeat the initialization);return;
}
这段代码意思就是一个initialized的变量代表的意思是是否初始化一开始的时候是false因为还没有开始初始化然后compareAndSet后面跟着两个参数第一个参数是预期值如果预期值和当前的变量值一样则将当前变量更新为第二个参数的值。
如果initialized的值是false 和预期值一样则compareAndSet方法返回的是true跳出if条件语句并且initialized值变成了true。
如果initialized的·值是true和预期值不一样则compareAndSet返回的是false进入条件语句打印告警日志并且不再有后续的初始化操作此时initialized的值不变依旧是true。 获取默认配置
PowerJobWorkerConfig config workerRuntime.getWorkerConfig();//下面这些代码都是在之后的初始化操作中进行赋值的
workerRuntime.setWorkerAddress(workerAddress);workerRuntime.setServerDiscoveryService(serverDiscoveryService);workerRuntime.setActorSystem(actorSystem);workerRuntime.setOmsLogHandler(omsLogHandler);workerRuntime.setTaskPersistenceService(taskPersistenceService);
这个WorkerRuntime类是worker.common包里面的一个Bean类记录了一些worker运行时的参数和环境里面有的有默认值有的没有默认值需要在初始化的时候进行赋值。比如上面代码中后面set的值
校验appName
我将里面有关打印日志的部分全部拿掉了通过appName去server请求appId如果请求不到则说明配置文件里面的“powerjob.worker.app-name”配置的有问题所有appName都是需要注册的所以名字是不会重复的。
private void assertAppName() {//获取到appNamePowerJobWorkerConfig config workerRuntime.getWorkerConfig();String appName config.getAppName();Objects.requireNonNull(appName, appName cant be empty!);//调用server端的服务String url http://%s/server/assert?appName%s;for (String server : config.getServerAddress()) {
//获取到server的请求地址String realUrl String.format(url, server, appName);try {
//请求服务返回结果String resultDTOStr CommonUtils.executeWithRetry0(() - HttpUtils.get(realUrl));
//解析返回结果ResultDTO resultDTO JsonUtils.parseObject(resultDTOStr, ResultDTO.class);if (resultDTO.isSuccess()) {
//将appId设置到运行环境里Long appId Long.valueOf(resultDTO.getData().toString());workerRuntime.setAppId(appId);return;}else {throw new PowerJobException(resultDTO.getMessage());}}catch (PowerJobException oe) {throw oe;}catch (Exception ignore) {}}throw new PowerJobException(no server available!);
}
连接Server serverDiscoveryService.start(timingPool); 最主要的就是上面这一行代码这个代码里面主要流程如下 将配置文件里面的服务器地址存入内存。 当前服务地址如果不为空调用server端的/acquire服务获取结果。 如果经过第二步没有结果返回则遍历配置文件中所有的server地址来获取结果。 如果依旧没有结果说明连接不到server需要将所有的本地任务停止。 如果得到结果则将结果返回。 private String discovery() {
//1.将配置文件里面的服务器地址存入内存。if (ip2Address.isEmpty()) {config.getServerAddress().forEach(x - ip2Address.put(x.split(:)[0], x));}String result null;
//2.当前服务地址如果不为空调用server端的/acquire服务获取结果。String currentServer currentServerAddress;if (!StringUtils.isEmpty(currentServer)) {String ip currentServer.split(:)[0];String firstServerAddress ip2Address.get(ip);if (firstServerAddress ! null) {result acquire(firstServerAddress);}}
//3.如果经过第二步没有结果返回则遍历配置文件中所有的server地址来获取结果。for (String httpServerAddress : config.getServerAddress()) {if (StringUtils.isEmpty(result)) {result acquire(httpServerAddress);}else {break;}}if (StringUtils.isEmpty(result)) {
//4.如果依旧没有结果说明连接不到server需要将所有的本地任务停止。if (FAILED_COUNT MAX_FAILED_COUNT) {ListLong frequentInstanceIds TaskTrackerPool.getAllFrequentTaskTrackerKeys();if (!CollectionUtils.isEmpty(frequentInstanceIds)) {frequentInstanceIds.forEach(instanceId - {TaskTracker taskTracker TaskTrackerPool.remove(instanceId);taskTracker.destroy();});}FAILED_COUNT 0;}return null;} else {
//5.如果得到结果则将结果返回。FAILED_COUNT 0;return result;}
} 初始化日志系统 OmsLogHandler omsLogHandler new OmsLogHandler(workerAddress, actorSystem, serverDiscoveryService); 这个日志系统的主要作用就是将本地的日志上报的server上从传进的参数就能看出都是和通讯相关的内容。 这个日志系统的提交也是异步单独占用一个线程在之前开启的线程中其中就有一个线程是用来提交日志的该线程会在worker启动的最后开启代码段如下 timingPool.scheduleWithFixedDelay(omsLogHandler.logSubmitter, 0, 5, TimeUnit.SECONDS); 固定每5秒提交一次日志。 初始化存储 worker使用的是本地的H2数据库持久化的策略分为磁盘和内存在worker停止的时候会将本地的数据文件全部销毁。其主要的初始化代码在worker.persistence包里面的ConnectionFactory类中源代码如下 private final String DISK_JDBC_URL String.format(jdbc:h2:file:%spowerjob_worker_db;DB_CLOSE_DELAY-1;DATABASE_TO_UPPERfalse, H2_PATH);
private final String MEMORY_JDBC_URL String.format(jdbc:h2:mem:%spowerjob_worker_db;DB_CLOSE_DELAY-1;DATABASE_TO_UPPERfalse, H2_PATH);public synchronized void initDatasource(StoreStrategy strategy) {strategy strategy null ? StoreStrategy.DISK : strategy;HikariConfig config new HikariConfig();config.setDriverClassName(Driver.class.getName());config.setJdbcUrl(strategy StoreStrategy.DISK ? DISK_JDBC_URL : MEMORY_JDBC_URL);config.setAutoCommit(true);// 池中最小空闲连接数量config.setMinimumIdle(2);// 池中最大连接数量config.setMaximumPoolSize(32);dataSource new HikariDataSource(config);try {FileUtils.forceDeleteOnExit(new File(H2_PATH));}catch (Exception ignore) {}
} HikariCP 是一个高性能的 JDBC 连接池组件HikariConfig 就是其相关的配置类。 总结 worker工作起来确实不是很容易需要找到自己的上级还需要记录自己工作的日志需要一个人干好多任务还需要再不耽误正常任务的同时向自己的上级汇报工作汇报自己的身体状态。简直就是我们底层程序员的真实写照啊。里面使用了很多经典的技术也有比较新的技术对于日志系统做的还是让我学到了很多。
