docs: public DbLoadContext load(RowBatch rowBatch, WeightController controller) {
          Assert.notNull(rowBatch);
          Identity identity = rowBatch.getIdentity();
          DbLoadContext context = buildContext(identity);
  
          try {
              List<EventData> datas = rowBatch.getDatas();
              context.setPrepareDatas(datas);
              // 执行重复录入数据过滤
              datas = context.getPrepareDatas();
              if (datas == null || datas.size() == 0) {
                  logger.info("##no eventdata for load, return");
                  return context;
              }
  
              // 因为所有的数据在DbBatchLoader已按照DateMediaSource进行归好类，不同数据源介质会有不同的DbLoadAction进行处理
              // 设置media source时，只需要取第一节点的source即可
              context.setDataMediaSource(ConfigHelper.findDataMedia(context.getPipeline(), datas.get(0).getTableId())
                  .getSource());
              interceptor.prepare(context);
              // 执行重复录入数据过滤
              datas = context.getPrepareDatas();
              // 处理下ddl语句，ddl/dml语句不可能是在同一个batch中，由canal进行控制
              // 主要考虑ddl的幂等性问题，尽可能一个ddl一个batch，失败或者回滚都只针对这条sql
              if (isDdlDatas(datas)) {
                  doDdl(context, datas);
              } else {
                  WeightBuckets<EventData> buckets = buildWeightBuckets(context, datas);
                  List<Long> weights = buckets.weights();
                  controller.start(weights);// weights可能为空，也得调用start方法
                  if (CollectionUtils.isEmpty(datas)) {
                      logger.info("##no eventdata for load");
                  }
                  adjustPoolSize(context); // 根据manager配置调整线程池
                  adjustConfig(context); // 调整一下运行参数
                  // 按权重构建数据对象
                  // 处理数据
                  for (int i = 0; i < weights.size(); i++) {
                      Long weight = weights.get(i);
                      controller.await(weight.intValue());
                      // 处理同一个weight下的数据
                      List<EventData> items = buckets.getItems(weight);
                      logger.debug("##start load for weight:" + weight);
                      // 预处理下数据
  
                      // 进行一次数据合并，合并相同pk的多次I/U/D操作
                      items = DbLoadMerger.merge(items);
                      // 按I/U/D进行归并处理
                      DbLoadData loadData = new DbLoadData();
                      doBefore(items, context, loadData);
                      // 执行load操作
                      doLoad(context, loadData);
                      controller.single(weight.intValue());
                      logger.debug("##end load for weight:" + weight);
                  }
              }
              interceptor.commit(context);
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
              interceptor.error(context);
          } catch (Exception e) {
              interceptor.error(context);
              throw new LoadException(e);
          }
  
          return context;// 返回处理成功的记录
      }