import storm.trident.state.CombinerValueUpdater; //导入依赖的package包/类
@Override
public void updateState(MapState map, List<TridentTuple> tuples, TridentCollector collector) {
    List<List<Object>> groups = new ArrayList<List<Object>>(tuples.size());
    List<ValueUpdater> updaters = new ArrayList<ValueUpdater>(tuples.size());
            
    for(TridentTuple t: tuples) {
        groups.add(_groupFactory.create(t));
        updaters.add(new CombinerValueUpdater(_agg,_inputFactory.create(t).getValue(0)));
    }
    List<Object> newVals = map.multiUpdate(groups, updaters);
   
    for(int i=0; i<tuples.size(); i++) {
        List<Object> key = groups.get(i);
        Object result = newVals.get(i);            
        collector.emit(_factory.create(new List[] {key, new Values(result) }));
    }
}
 

import storm.trident.state.CombinerValueUpdater; //导入依赖的package包/类
@Override
public void updateState(MapState map, List<TridentTuple> tuples, TridentCollector collector) {
    List<List<Object>> groups = new ArrayList<List<Object>>(tuples.size());
    List<ValueUpdater> updaters = new ArrayList<ValueUpdater>(tuples.size());

    for (TridentTuple t : tuples) {
        groups.add(_groupFactory.create(t));
        updaters.add(new CombinerValueUpdater(_agg, _inputFactory.create(t).getValue(0)));
    }
    List<Object> newVals = map.multiUpdate(groups, updaters);

    for (int i = 0; i < tuples.size(); i++) {
        List<Object> key = groups.get(i);
        Object result = newVals.get(i);
        collector.emit(_factory.create(new List[] { key, new Values(result) }));
    }
}
 

import storm.trident.state.CombinerValueUpdater; //导入依赖的package包/类
@Override
public void updateState(Snapshottable state, List<TridentTuple> tuples, TridentCollector collector) {
    if(tuples.size()!=1) {
        throw new IllegalArgumentException("Combiner state updater should receive a single tuple. Received: " + tuples.toString());
    }
    Object newVal = state.update(new CombinerValueUpdater(_agg, tuples.get(0).getValue(0)));
    collector.emit(new Values(newVal));
}
 

import storm.trident.state.CombinerValueUpdater; //导入依赖的package包/类
@Override
public void updateState(Snapshottable state, List<TridentTuple> tuples, TridentCollector collector) {
    if (tuples.size() != 1) {
        throw new IllegalArgumentException("Combiner state updater should receive a single tuple. Received: " + tuples.toString());
    }
    Object newVal = state.update(new CombinerValueUpdater(_agg, tuples.get(0).getValue(0)));
    collector.emit(new Values(newVal));
}