import backtype.storm.utils.WindowedTimeThrottler; //导入依赖的package包/类
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
    _throttler = new WindowedTimeThrottler((Number)conf.get(Config.TOPOLOGY_TRIDENT_BATCH_EMIT_INTERVAL_MILLIS), 1);
    for(String spoutId: _managedSpoutIds) {
        _states.add(TransactionalState.newCoordinatorState(conf, spoutId));
    }
    _currTransaction = getStoredCurrTransaction();

    _collector = collector;
    Number active = (Number) conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING);
    if(active==null) {
        _maxTransactionActive = 1;
    } else {
        _maxTransactionActive = active.intValue();
    }
    _attemptIds = getStoredCurrAttempts(_currTransaction, _maxTransactionActive);

    
    for(int i=0; i<_spouts.size(); i++) {
        String txId = _managedSpoutIds.get(i);
        _coordinators.add(_spouts.get(i).getCoordinator(txId, conf, context));
    }
}
 

import backtype.storm.utils.WindowedTimeThrottler; //导入依赖的package包/类
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
    _throttler = new WindowedTimeThrottler((Number) conf.get(Config.TOPOLOGY_TRIDENT_BATCH_EMIT_INTERVAL_MILLIS), 1);
    for (String spoutId : _managedSpoutIds) {
        _states.add(TransactionalState.newCoordinatorState(conf, spoutId));
    }
    _currTransaction = getStoredCurrTransaction();

    _collector = collector;
    Number active = (Number) conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING);
    if (active == null) {
        _maxTransactionActive = 1;
    } else {
        _maxTransactionActive = active.intValue();
    }
    _attemptIds = getStoredCurrAttempts(_currTransaction, _maxTransactionActive);


    for (int i = 0; i < _spouts.size(); i++) {
        String txId = _managedSpoutIds.get(i);
        _coordinators.add(_spouts.get(i).getCoordinator(txId, conf, context));
    }
}
 

import backtype.storm.utils.WindowedTimeThrottler; //导入依赖的package包/类
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
    _throttler = new WindowedTimeThrottler((Number) conf.get(Config.TOPOLOGY_TRIDENT_BATCH_EMIT_INTERVAL_MILLIS),
            Integer.MAX_VALUE);
    for (String spoutId : _managedSpoutIds) {
        _states.add(TransactionalState.newCoordinatorState(conf, spoutId));
    }
    _currTransaction = getStoredCurrTransaction();

    _collector = collector;
    Number active = (Number) conf.get(Config.TOPOLOGY_MAX_SPOUT_PENDING);
    if (active == null) {
        _maxTransactionActive = 1;
    } else {
        _maxTransactionActive = active.intValue();
    }
    _attemptIds = getStoredCurrAttempts(_currTransaction, _maxTransactionActive);
    avgWaittingTime = context.registerMetric("commit-wait-ms", new ReducedMetric(new MeanReducer()),
            Utils.getInt(conf.get(Config.TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS)));
    commitWaittingCount = context.registerMetric("commit-wait-cnt", new CountMetric(),
            Utils.getInt(conf.get(Config.TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS)));
    activeTxCount = context.registerMetric("active-tx-cnt", new ReducedMetric(new MeanReducer()),
            Utils.getInt(conf.get(Config.TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS)));
    for (int i = 0; i < _spouts.size(); i++) {
        String txId = _managedSpoutIds.get(i);
        _coordinators.add(_spouts.get(i).getCoordinator(txId, conf, context));
    }
}