datax(11):源码解读 ContainerCommunicator

datax(11):源码解读 ContainerCommunicator前面看了datax的通讯类communication,现在看看在他之上包装的一个容器通信类ContainerCommunicator一、抽象基类AbstractContainerCommunicatordataX中提供了一个基类 AbstractContainerCommunicator来处理JobContainer、TaskGroupContainer和Task的通讯。AbstractContainerCommunicator提供了注册、收集信息等接口,信息的单位是Communication.

大家好,又见面了,我是你们的朋友全栈君。

前面看了datax的 通讯类communication,现在看看在他之上包装的一个容器通信类ContainerCommunicator


一、抽象基类AbstractContainerCommunicator

dataX中提供了一个基类 AbstractContainerCommunicator来处理JobContainer、TaskGroupContainer和Task的通讯。AbstractContainerCommunicator提供了注册、收集信息等接口,信息的单位是Communication。

类继承关系
在这里插入图片描述

类的主要方法

在这里插入图片描述


二、AbstractContainerCommunicator的主要两个属性

AbstractContainerCommunicator主要将其功能委托给2个属性:
private AbstractCollector collector;
private AbstractReporter reporter;

1、AbstractCollector collector

Collector负责管理下级注册到上级,搜集并合并下级所有的信息。 dataX提供一个基类AbstractCollector和一个实现类ProcessInnerCollector

在这里插入图片描述

1.1 AbstractCollector同时包含将Task注册到TaskGroupContainer(registerTaskCommunication方法)和将TaskGroupContainer注册到JobContainer(registerTGCommunication方法)的功能。具体如下:
  • taskCommunicationMap属性用于保存Task注册到TaskGroupContainer,当Task注册到TaskGroupContainer的时候将TaskId和新建的Communication对象保存进taskCommunicationMap即可(在registerTaskCommunication方法中)。

  /** * Task注册到TaskGroupContainer * * @param taskConfigurationList List<Configuration> */
  public void registerTaskCommunication(List<Configuration> taskConfigurationList) { 
   
    for (Configuration taskConfig : taskConfigurationList) { 
   
      int taskId = taskConfig.getInt(CoreConstant.TASK_ID);
      this.taskCommunicationMap.put(taskId, new Communication());
    }
  }
  • TaskGroupContainer注册到JobContainer注册信息则是保存在全局变量LocalTGCommunicationManager中,便于全局访问(在registerTGCommunication方法中)。
  /** * 将TaskGroupContainer注册到JobContainer * * @param taskGroupConfigurationList List<Configuration> */
  public void registerTGCommunication(List<Configuration> taskGroupConfigurationList) { 
   
    for (Configuration config : taskGroupConfigurationList) { 
   
      int taskGroupId = config.getInt(
          CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_ID);
      LocalTGCommunicationManager.registerTaskGroupCommunication(taskGroupId, new Communication());
    }
  }

此外AbstractCollector#collectFromTask提供搜集所有任务信息的功能;


    /** * 搜集所有任务信息的功能 * @return Communication */
  public Communication collectFromTask() { 
   
    Communication communication = new Communication();
    communication.setState(State.SUCCEEDED);

    for (Communication taskCommunication :
        this.taskCommunicationMap.values()) { 
   
      communication.mergeFrom(taskCommunication);
    }
    return communication;
  }

1.2 实现类ProcessInnerCollector只实现 了一个方法collectFromTaskGroup,collectFromTaskGroup提供搜集所有TaskGroupContainer的信息。
    @Override
    public Communication collectFromTaskGroup() { 
   
        return LocalTGCommunicationManager.getJobCommunication();
    }


2、AbstractReporter reporter

Reporter的主要功能是将收集到的信息上报给上级。dataX提供一个基类AbstractReporter和一个实现类ProcessInnerCollector.

类继承关系
在这里插入图片描述

主要方法
在这里插入图片描述

2.1 ProcessInnerCollector#reportJobCommunication将job信息汇报给上级,job在dataX中是最上级,所以该方法没有操作。
  • ProcessInnerCollector#reportTGCommunication将TaskGroupContianer的信息汇报给上级,操作也很简单直接更新注册时分配给该TaskGroup的Communication(Map中的值)
public class ProcessInnerReporter extends AbstractReporter { 
   

  @Override
  public void reportJobCommunication(Long jobId, Communication communication) { 
   
    // do nothing
  }

  /** * 将TaskGroupContianer的信息汇报给上级,操作也很简单直接更新注册时分配给该TaskGroup的Communication(Map中的值) * * @param taskGroupId Integer * @param communication Communication */
  @Override
  public void reportTGCommunication(Integer taskGroupId, Communication communication) { 
   
    LocalTGCommunicationManager.updateTaskGroupCommunication(taskGroupId, communication);
  }
}

三、 AbstractContainerCommunicator的实现类

1、StandAloneJobContainerCommunicator

StandAloneJobContainerCommunicator是AbstractContainerCommunicator一个实现类,主要处理JobContainer和TaskGroupContainer之间的信息传递。

  1. 每个TaskGroupContainer通过StandAloneJobContainerCommunicator#registerCommunication注册
  2. 注册之后TaskGroupContainer每隔一段时间通过StandAloneJobContainerCommunicator#Reporter#report向JobContainer发送自己的状态。
  3. JobContainer每隔一段时间通过StandAloneJobContainerCommunicator#collect获取TaskGroup的信息。最后调用StandAloneJobContainerCommunicator#report向上级汇报,这里JobContainer已经是最上级了,向日志中输出先关信息即可。
/** * 主要处理JobContainer和TaskGroupContainer之间的信息传递 */
public class StandAloneJobContainerCommunicator extends AbstractContainerCommunicator { 
   

  private static final Logger LOG = LoggerFactory
      .getLogger(StandAloneJobContainerCommunicator.class);

  public StandAloneJobContainerCommunicator(Configuration cfg) { 
   
    super(cfg);
    super.setCollector(new ProcessInnerCollector(cfg.getLong(DATAX_CORE_CONTAINER_JOB_ID)));
    super.setReporter(new ProcessInnerReporter());
  }

  @Override
  public void registerCommunication(List<Configuration> configurationList) { 
   
    super.getCollector().registerTGCommunication(configurationList);
  }

  /** * JobContainer每隔一段时间 主动 获取TaskGroup的信息。最后调用本类的#report向上级汇报, * 这里JobContainer已经是最上级了,向日志中输出先关信息即可 * @return */
  @Override
  public Communication collect() { 
   
    return super.getCollector().collectFromTaskGroup();
  }

  @Override
  public State collectState() { 
   
    return this.collect().getState();
  }

  /** * 和 DistributeJobContainerCollector 的 report 实现一样 * 每隔一段时间向JobContainer 主动 发送自己的状态 */
  @Override
  public void report(Communication communication) { 
   
    super.getReporter().reportJobCommunication(super.getJobId(), communication);
    LOG.info(CommunicationTool.Stringify.getSnapshot(communication));
    reportVmInfo();
  }

  @Override
  public Communication getCommunication(Integer taskGroupId) { 
   
    return super.getCollector().getTGCommunication(taskGroupId);
  }

  @Override
  public Map<Integer, Communication> getCommunicationMap() { 
   
    return super.getCollector().getTGCommunicationMap();
  }
}

2、AbstractTGContainerCommunicator

AbstractTGContainerCommunicator是AbstractContainerCommunicator的另一个抽象实现类,

/** * 该类是用于处理 taskGroupContainer 的 communication 的收集汇报的父类 * 主要是 taskCommunicationMap 记录了 taskExecutor 的 communication 属性 * 主要处理TaskGroupContainer和Task之间的信息 */
public abstract class AbstractTGContainerCommunicator extends AbstractContainerCommunicator { 
   

    protected long jobId;

    /** * 由于taskGroupContainer是进程内部调度 * 其registerCommunication(),getCommunication(), * getCommunications(),collect()等方法是一致的 * 所有TG的Collector都是ProcessInnerCollector */
    protected int taskGroupId;

    public AbstractTGContainerCommunicator(Configuration configuration) { 
   
        super(configuration);
        this.jobId = configuration.getInt(
                CoreConstant.DATAX_CORE_CONTAINER_JOB_ID);
        super.setCollector(new ProcessInnerCollector(this.jobId));
        this.taskGroupId = configuration.getInt(
                CoreConstant.DATAX_CORE_CONTAINER_TASKGROUP_ID);
    }

    @Override
    public void registerCommunication(List<Configuration> configurationList) { 
   
        super.getCollector().registerTaskCommunication(configurationList);
    }

    @Override
    public final Communication collect() { 
   
        return this.getCollector().collectFromTask();
    }

    @Override
    public final State collectState() { 
   
        Communication communication = new Communication();
        communication.setState(State.SUCCEEDED);

        for (Communication taskCommunication :
                super.getCollector().getTaskCommunicationMap().values()) { 
   
            communication.mergeStateFrom(taskCommunication);
        }

        return communication.getState();
    }

    @Override
    public final Communication getCommunication(Integer taskId) { 
   
        Validate.isTrue(taskId >= 0, "注册的taskId不能小于0");

        return super.getCollector().getTaskCommunication(taskId);
    }

    @Override
    public final Map<Integer, Communication> getCommunicationMap() { 
   
        return super.getCollector().getTaskCommunicationMap();
    }

}

从类的继承实现看最终实现类是StandaloneTGContainerCommunicator, 该类主要处理TaskGroupContainer和Task之间的信息,处理逻辑和StandAloneJobContainerCommunicator差不多

/** * 独立模式的taskGroup 的通讯类 主要处理TaskGroupContainer和Task之间的信息,处理逻辑和StandAloneJobContainerCommunicator差不多 */
public class StandaloneTGContainerCommunicator extends AbstractTGContainerCommunicator { 
   

  /** * 单机版的容器沟通者(独立模式的taskGroup 的通讯类) * * @param configuration */
  public StandaloneTGContainerCommunicator(Configuration configuration) { 
   
    super(configuration);
    super.setReporter(new ProcessInnerReporter());
  }

  @Override
  public void report(Communication communication) { 
   
    super.getReporter().reportTGCommunication(super.taskGroupId, communication);
  }

}

注:

  1. 对源码进行略微改动,主要修改为 1 阿里代码规约扫描出来的,2 clean code;

  2. 所有代码都已经上传到github(master分支和dev),可以免费白嫖

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/145375.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • msfconsole爆破ftp_check point防火墙

    msfconsole爆破ftp_check point防火墙Msfconsole爆破ssh一:什么是sshSsh为linux系统下的远程登录命令。好比windows下的远程桌面。登录ssh后可以对服务器进行相关的操作!二:端口扫描目标发现1:用namp或者netcat扫描网络,发现我的路由器是开启了ssh服务2:用netcap扫描三:msfconsole激活成功教程密码1:msfconsole运行…

  • JLINK V9项目启动【jlink接口定义】【开启VCOM(虚拟串口)功能】「建议收藏」

    JLINK V9项目启动【jlink接口定义】【开启VCOM(虚拟串口)功能】「建议收藏」jlink接口定义摘录于:https://blog.csdn.net/u014124220/article/details/50829713仿真器端口 连接目标板 备注 1.VCC MCU电源VCC VCC 2.VCC MCU电源VCC VCC 3.TRST TRST Tes…

  • Java新手、小白入门。多敲练习代码!!!

    Java新手、小白入门。多敲练习代码!!!如果你喜欢Java,但是想学不会!我建议你没事的时候敲敲这些代码,希望对你有用!publicclassDemo{ publicstaticvoidmain(String[]args){ System.out.print(“你好\n世界”); System.out.println(“你好\tJava”); System.out.println(“1.电脑要求相对干…

  • JDBC从入门到精通

    JDBC从入门到精通JDBC概念我们学习了数据库,数据库实现了数据的持久化,但我们最终要在程序里处理数据啊,那java代码中怎么去访问数据库读写数据呢?这就要用到sun公司设定的一套数据库标准了,这套标准就是JDBC(JavaDatabaseConnectivity)。但它只是规范,不做具体实现。于是数据库厂商又根据JDBC标准,实现自家的驱动Driver。如:mysql驱动com.mysql.cj.jdbc.Driver,Oracle的驱动oracle.jdbc.OracleDriver。有了这套解决方案,java

  • 逻辑运算符Python_逻辑运算符的优先级

    逻辑运算符Python_逻辑运算符的优先级#练习1:定义一个整数变量age,编写代码判断年龄是否正确age=180#要求人的年龄在0~120之间#if0<=age<=120:#新语法ifage>=0andage<=120:print(“年龄正确”)else:print(“年龄不正确”)…

  • Java50个关键字总结「建议收藏」

    Java50个关键字总结「建议收藏」作业一:Java基础知识复习一、关键字总结1.abstract修饰类:abstract修饰类,这个类就是抽象类,抽象类中可以有非抽象变量和成员变量,也可以有普通方法、构造方法。但是不能实例化,只能被子类继承。如果子类不是抽象类,则必须重写父类的抽象方法。publicabstractclassAbstractList&lt;E&gt;e…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号