import storm.trident.operation.MultiReducer; //导入依赖的package包/类
public Stream multiReduce(List<Fields> inputFields, List<Stream> streams, MultiReducer function, Fields outputFields) {
    List<String> names = new ArrayList<String>();
    for(Stream s: streams) {
        if(s._name!=null) {
            names.add(s._name);
        }
    }
    Node n = new ProcessorNode(getUniqueStreamId(), Utils.join(names, "-"), outputFields, outputFields, new MultiReducerProcessor(inputFields, function));
    return addSourcedNode(streams, n);
}
 

import storm.trident.operation.MultiReducer; //导入依赖的package包/类
public Stream multiReduce(List<Fields> inputFields, List<Stream> streams, MultiReducer function, Fields outputFields) {
    List<String> names = new ArrayList<>();
    for(Stream s: streams) {
        if(s._name!=null) {
            names.add(s._name);
        }
    }
    Node n = new ProcessorNode(getUniqueStreamId(), Utils.join(names, "-"), outputFields, outputFields, new MultiReducerProcessor(inputFields, function));
    return addSourcedNode(streams, n);
}
 

import storm.trident.operation.MultiReducer; //导入依赖的package包/类
public MultiReducerProcessor(List<Fields> inputFields, MultiReducer reducer) {
    _reducer = reducer;
    _projectFields = inputFields;
}