datax(10): 源码解读Communication(Datax通讯类)「建议收藏」

datax(10): 源码解读Communication(Datax通讯类)「建议收藏」前面看了datax的通讯机制,继续看源码—具体的通讯类Communication。根据datax的运行模式的区别,数据的收集会有些区别,这篇文章都是讲的在standalone模式下。一、communication概述DataX所有的统计信息都会保存到Communication类里面。Communication支持下列数据的统计计数器,比如读取的字节速度,写入成功的数据条数/***所有的数值key-value对**/privateMap<String.

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

前面看了datax的通讯机制,继续看源码—具体的通讯类 Communication。根据datax的运行模式的区别, 数据的收集会有些区别,这篇文章都是讲的在standalone模式下。


一、communication概述

DataX所有的统计信息都会保存到Communication类里面。

Communication支持下列数据的统计

  1. 计数器,比如读取的字节速度,写入成功的数据条数
  2. 统计的时间点 字符串类型的消息
  3. 执行时的异常
  4. 执行的状态, 比如成功或失败

  /** * 所有的数值key-value对 * */
  private Map<String, Number> counter;

  /** * 运行状态 * */
  private State state;

  /** * 异常记录 * */
  private Throwable throwable;

  /** * 记录的timestamp * */
  private long timestamp;

  /** * task给job的信息 * */
  Map<String, List<String>> message;
  

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

如果需要汇总多个Communication的数据,Communication提供了mergeFrom方法。根据不同的数据类型,对应着不同的操作计数器类型,相同的key的数值累加

  • 合并异常,当自身的异常为null,才合并别的异常

  • 合并状态,如果有任意一个的状态失败了,那么返回失败的状态。如果有任意一个的状态正在运行,那么返回正在运行的状态

  • 合并消息, 相同的key的消息添加到同一个列表


二、communication主要方法

在这里插入图片描述


三、Communication的管理类

对于每个task组都有一个单独的Communication,用来存储这个组的统计数据。对于这些Communication,在LocalTGCommunicationManager类实现了集中管理。接下来看看LocalTGCommunicationManager的原理。

LocalTGCommunicationManager有个重要的属性taskGroupCommunicationMap,它是一个Map,保存了每个task组的统计数据。


public final class LocalTGCommunicationManager { 
   

  private static Map<Integer, Communication> taskGroupCommunicationMap = new ConcurrentHashMap<>();

  /** * 根据tgId注册comm * 当task组在初始化的时候,都会向LocalTGCommunicationManager这里注册。// 这里只是简单保存到taskGroupCommunicationMap变量里 * @param taskGroupId * @param communication */
  public static void registerTaskGroupCommunication(int taskGroupId, Communication communication) { 
   
    taskGroupCommunicationMap.put(taskGroupId, communication);
  }

  /** * 获取(合并)tg里面所有的comm * * @return Communication */
  public static Communication getJobCommunication() { 
   
    Communication communication = new Communication();
    communication.setState(State.SUCCEEDED);

    for (Communication taskGroupCommunication : taskGroupCommunicationMap.values()) { 
   
      communication.mergeFrom(taskGroupCommunication);
    }
    return communication;
  }

  /** * 采用获取taskGroupId后再获取对应communication的方式, * 防止map遍历时修改,同时也防止对map key-value对的修改 * * @return */
  public static Set<Integer> getTaskGroupIdSet() { 
   
    return taskGroupCommunicationMap.keySet();
  }

  public static Communication getTaskGroupCommunication(int taskGroupId) { 
   
    Validate.isTrue(taskGroupId >= 0, "taskGroupId不能小于0");
    return taskGroupCommunicationMap.get(taskGroupId);
  }


  /** * 根据tgId 将taskGroupCommunicationMap中没有的comm 插入 * @param taskGroupId * @param comm */
  public static void updateTaskGroupCommunication(final int taskGroupId, final Communication comm) { 
   
    Validate.isTrue(taskGroupCommunicationMap.containsKey(
        taskGroupId), String.format("taskGroupCommunicationMap中没有注册taskGroupId[%d]的Communication," +
        "无法更新该taskGroup的信息", taskGroupId));
    taskGroupCommunicationMap.put(taskGroupId, comm);
  }

  public static void clear() { 
   
    taskGroupCommunicationMap.clear();
  }

  public static Map<Integer, Communication> getTaskGroupCommunicationMap() { 
   
    return taskGroupCommunicationMap;
  }
}

四、谁会注册Communication

AbstractScheduler会根据切分后的任务,为每个task组注册一个Communication。registerCommunication接收task配置列表,里面每个配置都包含了task group id。

进行注册communication的类

  • AbstractScheduler的schedule方法里 registerCommunication
  • TaskGroupContainer的start方法里 registerCommunication
  • AbstractTGContainerCommunicator的registerCommunication方法
  • AbstractContainerCommunicator的registerCommunication方法
  • StandAloneJobContainerCommunicator的registerCommunication方法

在这里插入图片描述


五、更新communication统计数据

主要更新communication的类
在这里插入图片描述

每个任务执行都会对应着Channel,Channel当每处理一条数据时,都会更新对应Communication的统计信息。
例如下面的pull方法是Writer从Channel拉取数据,每次pull的时候,都会调用statPull函数,会更新写入数据条数和字节数的信息。


public abstract class Channel{ 
   

    private Communication currentCommunication;

    public Record pull() { 
   
        Record record = this.doPull();
        this.statPull(1L, record.getByteSize());
        return record;
    }
    
    /** * statPull方法,并没有限速。因为数据的整个流程是Reader -》 Channle -》 Writer, Reader的push速度限制了, * Writer的pull速度也就没必要限速 * * @param recordSize * @param byteSize */
    private void statPull(long recordSize, long byteSize) { 
   
        currentCommunication.increaseCounter(CommunicationTool.WRITE_RECEIVED_RECORDS, recordSize);
        currentCommunication.increaseCounter(CommunicationTool.WRITE_RECEIVED_BYTES, byteSize);
    }
    

六、收集communication统计数据

  1. AbstractScheduler想统计汇总后的数据,需要调用AbstractContainerCommunicator的collect方法

  2. StandAloneJobContainerCommunicator继承AbstractContainerCommunicator,实现了collect方法,它会调用AbstractCollector的collectFromTaskGroup方法获取数据

  3. ProcessInnerCollector实现了AbstractCollector的collectFromTaskGroup方法,它会调用LocalTGCommunicationManager的getJobCommunication方法, getJobCommunication方法会统计所有task的数据,然后返回。

在这里插入图片描述


注:

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

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

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

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

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

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

(0)


相关推荐

  • WIN10查看CUDA版本「建议收藏」

    WIN10查看CUDA版本「建议收藏」WIN10下查看CUDA版本1.打开控制面板将红色方框里的类别更改为小图标选择红色方框里的NVIDIA控制面板,选择帮助选择系统信息选择组件可以看到CUDA的版本是10.1以上です。…

  • LoadImage()的使用

    LoadImage()的使用

  • A记录、CNAME和URL转发区别[通俗易懂]

    A记录、CNAME和URL转发区别[通俗易懂]我们在做域名解析时,尤其是很多虚拟主机,大都会使用到CNAME解析,独立主机、VPS则用A记录较多,而URL转发则会在更换域名时用到,从设置效果来看,都是“解析”到一个“其它”URL地址,而实际上它们之间还是有些区别的,尤其是URL转发和其它两个之间区别很大的,首先A记录和CNAME属于标准的DNS记录,而URL转发则实际上只是个简单的重定向。另外,我们还常遇到别名ALIAS这个词,ALIAS对解

  • 大数据开发工程师面试题以及答案整理(二)

    大数据开发工程师面试题以及答案整理(二)Redis性能优化,单机增加CPU核数是否会提高性能1、根据业务需要选择合适的数据类型,并为不同的应用场景设置相应的紧凑存储参数。2、当业务场景不需要数据持久化时,关闭所有的持久化方式可以获得最佳的性能以及最大的内存使用量。3、如果需要使用持久化,根据是否可以容忍重启丢失部分数据在快照方式与语句追加方式之间选择其一,不要使用虚拟内存以及diskstore方式。4、不要让你的Redis所在机…

  • MT4行情交易API接口开发手记

    MT4行情交易API接口开发手记之前开发的外汇量化交易系统,行情和交易接口都是通过在MT4平台下编写EA来实现,具体方法是:1、用C++编写一个动态库文件,在里面实现行情和交易数据调用接口,将报价数据和K线数据写入数据库中,并从数据库中获取外汇量化系统发出的交易指令。2、在MT4中编写EA文件,在MT4上不间断运行,从MT4平台实时获取报价和K线数据,并调用动态库写入数据库中,于此同时,不断从数据库中获取交易指令,再调用MT4的交易指令完成交易。采用此种方法的好处就是兼容性强,只要打开MT4软件运行EA,就可以完成行情和交易接..

  • 高可用的Redis主从复制集群,从理论到实践

    点击上方“全栈程序员社区”,星标公众号 重磅干货,第一时间送达 作者:Sicimike blog.csdn.net/Baisitao_/article/details/105545…

发表回复

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

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