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