@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);
    }
@ 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 ) ; }
组件业务执行方法

@param msg 具体的消息