code: @Override
      public void execute(UnitRequest msg, Handler<UnitResponse> handler) {
          Single<UnitResponse> result;
          boolean flush = msg.get(FLUSH, Boolean.class, false);//是否马上提交
          msg.getArgMap().remove(FLUSH);
          //判断是否需要批量操作
          if (getBatchSize() >= MIN_BATCH_SIZE) {
              recordCacheList.add(msg.getArgMap());
              //判断是否达到批量执行的数量阈值，或者业务需要立即执行
              if (recordCacheList.size() >= getBatchSize() || flush) {
                  Map<String, Object> params = new HashMap<>();
                  synchronized (recordCacheList) {
                      params.put(VALUES, new ArrayList<Map>(recordCacheList));
                      recordCacheList.clear();
                  }
                  preBatchExecute(msg);
                  result = SingleRxXian.call(getBatchGroupName(), getBatchUnitName(), params);
              } else {
                  result = doCache(msg);
              }
          } else {//没有批量需求，立即执行
              result = SingleRxXian.call(getBatchGroupName(), getBatchUnitName(), msg.getArgMap());
          }
          result.subscribe(handler::handle);
      }
docs: 组件业务执行方法
  
  @param msg 具体的消息