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

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

前面看了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），可以免费白嫖