聊聊PowerJob Worker的ServerAddress

2024-02-08 04:12

本文主要是介绍聊聊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的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/689866

相关文章

java线程深度解析(四)——并发模型(Master-Worker)

http://blog.csdn.net/daybreak1209/article/details/51372929 二、Master-worker ——分而治之      Master-worker常用的并行模式之一,核心思想是由两个进程协作工作,master负责接收和分配任务,worker负责处理任务,并把处理结果返回给Master进程,由Master进行汇总,返回给客

【HarmonyOS】-TaskPool和Worker的对比实践

ArkTS提供了TaskPool与Worker两种多线程并发方案,下面我们将从其工作原理、使用效果对比两种方案的差异,进而选择适用于ArkTS图片编辑场景的并发方案。 TaskPool与Worker工作原理 TaskPool与Worker两种多线程并发能力均是基于 Actor并发模型实现的。Worker主、子线程通过收发消息进行通信;TaskPool基于Worker做了更多场景化的功能封装,例

聊聊说话的习惯

1 在日常生活中,每个人都有固定的说话习惯。心理学研究表明,通过一个人的说话习惯,也可以分析出他的性格特点。对于每一个人来讲,说话习惯已经融为他们生活中的一部分。在社交活动中,一些不良的说话习惯很可能会给他们带来麻烦。因此,了解说话习惯对心理活动的影响是十分有必要的。 2 具有顺畅的说话习惯的人,大多思路清晰、语速适中、用词准确并且声声人耳,是典型的顺畅型说话方式这种类型的人要么不说话,要么

聊聊分布式,再讨论分布式解决方案

前言 最近很久没有写博客了,一方面是因为公司事情最近比较忙,另外一方面是因为在进行 CAP 的下一阶段的开发工作,不过目前已经告一段落了。 接下来还是开始我们今天的话题,说说分布式事务,或者说是我眼中的分布式事务,因为每个人可能对其的理解都不一样。 分布式事务是企业集成中的一个技术难点,也是每一个分布式系统架构中都会涉及到的一个东西,特别是在微服务架构中,几乎可以说是无法避免,本文就分布式事

聊聊资源调度

资源调度 般分为两个阶段: 是实现物理资源的虚拟化(即资源的抽象)于当前机器的性能越来越好,硬件配置越来越高,直接用物理机跑业务比较浪费,所以将物理机分割成更小单位的虚拟机,这样可以显著提升机器的利用效率,在公司内部一般采用容器技术来隔离资源 是将资源虚拟化后进 步在时间和空间上实现更细粒度的编排 ,优化资源的使用。 1 .一些数据 如果公司的几万台机器都是物理机,那么资源的使用率稍低: CP

聊聊Spark中的宽依赖和窄依赖

开门见山,本文就针对一个点,谈谈Spark中的宽依赖和窄依赖,这是Spark计算引擎划分Stage的根源所在,遇到宽依赖,则划分为多个stage,针对每个Stage,提交一个TaskSet: 上图:一张网上的图: 基于此图,分析下这里为什么前面的流程都是窄依赖,而后面的却是宽依赖: 我们仔细看看,map和filter算子中,对于父RDD来说,一个分区内的数据,有且仅有一个子RDD的分区来

php think worker:gateway

php think worker:gateway

聊聊灰度发布

有没有在北京面试java的小伙伴,每家公司面试问的问题都不一样,昨天面试官问到了灰度发布,一脸懵,好像在哪儿听说过,毕竟我都没发布过,之前都是项目组长在干这些事儿,所以聊聊,了解一下 什么是灰度发布 全量发布:把旧服务kill掉,把新服务启动,这个过程就可以理解为全量发布 回滚周期长 如果我们更新完应用之后,我们做线上回归测试的时候发现有BUG,这个时候就要做回滚,过程就是把新服

聊聊随机测试和猴子测试

目录 随机测试的特点 1.不可预测性 2.缺乏针对性 3.自动化 4.资源密集型 猴子测试 随机测试 (Random Testing) 猴子测试 (Monkey Testing) 特点: 区别 1.控制程度 2.目标差异 3.实现方式 在我们测试的过程中,通常会使用到随机测试和猴子测试,其中随机测试侧重于人工测试,猴子测试侧重于借助工具执行命令进行测试。 随机测试

【聊聊经济社会】论阶级跨越

为什么要在市场中寻求自由,在市场中寻求洒脱,原因不胜其数,其中便有一条,现实生活中多是xx,可能社会属性本身就具备党同伐异,像是一股意志,平庸一切不平庸,中和一切特立独行,最终以达到一种变态的稳定. 消其意志,断其未来,耗其钱财 ,而我称之为阶级壁垒 阶级之所以难以跨越,主要也在于这三点 一:没有这样的志向,像那种羡慕有钱,或者羡慕有权,权当做梦。这样的志向,正常人只停留于羡慕的层次,而一旦受到丁