Java TimeCacheMap类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Java›Java编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Java中backtype.storm.utils.TimeCacheMap类的典型用法代码示例。如果您正苦于以下问题：Java TimeCacheMap类的具体用法？Java TimeCacheMap怎么用？Java TimeCacheMap使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

TimeCacheMap类属于backtype.storm.utils包，在下文中一共展示了TimeCacheMap类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: finishFileUpload

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public void finishFileUpload(String location) throws TException {

	TimeCacheMap<Object, Object> uploaders = data.getUploaders();
	Object obj = uploaders.get(location);
	if (obj == null) {
		throw new TException(
				"File for that location does not exist (or timed out)");
	}
	try {
		if (obj instanceof WritableByteChannel) {
			WritableByteChannel channel = (WritableByteChannel) obj;
			channel.close();
			uploaders.remove(location);
			LOG.info("Finished uploading file from client: " + location);
		} else {
			throw new TException("Object isn't WritableByteChannel for "
					+ location);
		}
	} catch (IOException e) {
		LOG.error(" WritableByteChannel close failed when finishFileUpload "
				+ location);
	}

}

开发者ID:zhangjunfang，项目名称:jstorm-0.9.6.3-，代码行数:26，代码来源:ServiceHandler.java

示例2: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public void prepare(Map conf, TopologyContext context,
		OutputCollector collector) {

	taskId = String.valueOf(context.getThisTaskId());
	taskName = context.getThisComponentId() + "_" + context.getThisTaskId();

	this.basicCollector = new BasicOutputCollector(collector);
	this.collector = collector;

	if (delegate instanceof ICommitter) {
		isCommiter = true;
		commited = new TimeCacheMap<Object, Object>(
				context.maxTopologyMessageTimeout());
		mkCommitDir(conf);
	}

	delegate.prepare(conf, context);

}

开发者ID:zhangjunfang，项目名称:jstorm-0.9.6.3-，代码行数:21，代码来源:CoordinatedBolt.java

示例3: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
public void prepare(Map conf, TopologyContext context,
		OutputCollector collector) {

	taskId = String.valueOf(context.getThisTaskId());
	taskName = context.getThisComponentId() + "_" + context.getThisTaskId();

	this.basicCollector = new BasicOutputCollector(collector);
	this.collector = collector;

	if (delegate instanceof ICommitter) {
		isCommiter = true;
		commited = new TimeCacheMap<Object, Object>(
				context.maxTopologyMessageTimeout());
		mkCommitDir(conf);
	}

	delegate.prepare(conf, context);

}

开发者ID:songtk，项目名称:learn_jstorm，代码行数:20，代码来源:CoordinatedBolt.java

示例4: createFileHandler

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
public void createFileHandler() {
    TimeCacheMap.ExpiredCallback<Object, Object> expiredCallback = new TimeCacheMap.ExpiredCallback<Object, Object>() {
        @Override
        public void expire(Object key, Object val) {
            try {
                LOG.info("Close file " + String.valueOf(key));
                if (val != null) {
                    if (val instanceof Channel) {
                        Channel channel = (Channel) val;
                        channel.close();
                    } else if (val instanceof BufferFileInputStream) {
                        BufferFileInputStream is = (BufferFileInputStream) val;
                        is.close();
                    }
                }
            } catch (IOException e) {
                LOG.error(e.getMessage(), e);
            }

        }
    };

    int file_copy_expiration_secs = JStormUtils.parseInt(conf.get(Config.NIMBUS_FILE_COPY_EXPIRATION_SECS), 30);
    uploaders = new TimeCacheMap<Object, Object>(file_copy_expiration_secs, expiredCallback);
    downloaders = new TimeCacheMap<Object, Object>(file_copy_expiration_secs, expiredCallback);
}

开发者ID:kkllwww007，项目名称:jstrom，代码行数:27，代码来源:NimbusData.java

示例5: uploadChunk

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
/**
 * uploading topology jar data
 */
@Override
public void uploadChunk(String location, ByteBuffer chunk) throws TException {
    TimeCacheMap<Object, Object> uploaders = data.getUploaders();
    Object obj = uploaders.get(location);
    if (obj == null) {
        throw new TException("File for that location does not exist (or timed out) " + location);
    }
    try {
        if (obj instanceof WritableByteChannel) {
            WritableByteChannel channel = (WritableByteChannel) obj;
            channel.write(chunk);
            uploaders.put(location, channel);
        } else {
            throw new TException("Object isn't WritableByteChannel for " + location);
        }
    } catch (IOException e) {
        String errMsg = " WritableByteChannel write filed when uploadChunk " + location;
        LOG.error(errMsg);
        throw new TException(e);
    }

}

开发者ID:kkllwww007，项目名称:jstrom，代码行数:26，代码来源:ServiceHandler.java

示例6: finishFileUpload

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public void finishFileUpload(String location) throws TException {
    TimeCacheMap<Object, Object> uploaders = data.getUploaders();
    Object obj = uploaders.get(location);
    if (obj == null) {
        throw new TException("File for that location does not exist (or timed out)");
    }
    try {
        if (obj instanceof WritableByteChannel) {
            WritableByteChannel channel = (WritableByteChannel) obj;
            channel.close();
            uploaders.remove(location);
            LOG.info("Finished uploading file from client: " + location);
        } else {
            throw new TException("Object isn't WritableByteChannel for " + location);
        }
    } catch (IOException e) {
        LOG.error(" WritableByteChannel close failed when finishFileUpload " + location);
    }

}

开发者ID:kkllwww007，项目名称:jstrom，代码行数:22，代码来源:ServiceHandler.java

示例7: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
public void prepare(Map conf, TopologyContext context, OutputCollector collector) {

        taskId = String.valueOf(context.getThisTaskId());
        taskName = context.getThisComponentId() + "_" + context.getThisTaskId();

        this.basicCollector = new BasicOutputCollector(collector);
        this.collector = collector;

        if (delegate instanceof ICommitter) {
            isCommiter = true;
            commited = new TimeCacheMap<Object, Object>(context.maxTopologyMessageTimeout());
            mkCommitDir(conf);
        }

        delegate.prepare(conf, context);

    }

开发者ID:kkllwww007，项目名称:jstrom，代码行数:18，代码来源:CoordinatedBolt.java

示例8: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
    TimeCacheMap.ExpiredCallback<Object, TrackingInfo> callback = null;
    if (_delegate instanceof TimeoutCallback) {
        callback = new TimeoutItems();
    }
    _tracked = new TimeCacheMap<Object, TrackingInfo>(context.maxTopologyMessageTimeout(), callback);
    _collector = collector;
    _delegate.prepare(config, context, new OutputCollector(new CoordinatedOutputCollector(collector)));
    for (String component : Utils.get(context.getThisTargets(), Constants.COORDINATED_STREAM_ID, new HashMap<String, Grouping>()).keySet()) {
        for (Integer task : context.getComponentTasks(component)) {
            _countOutTasks.add(task);
        }
    }
    if (!_sourceArgs.isEmpty()) {
        _numSourceReports = 0;
        for (Entry<String, SourceArgs> entry : _sourceArgs.entrySet()) {
            if (entry.getValue().singleCount) {
                _numSourceReports += 1;
            } else {
                _numSourceReports += context.getComponentTasks(entry.getKey()).size();
            }
        }
    }
}

开发者ID:kkllwww007，项目名称:jstrom，代码行数:25，代码来源:CoordinatedBolt.java

示例9: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
public void prepare(Map config, TopologyContext context, OutputCollector collector) {
    TimeCacheMap.ExpiredCallback<Object, TrackingInfo> callback = null;
    if (_delegate instanceof TimeoutCallback) {
        callback = new TimeoutItems();
    }
    _tracked = new TimeCacheMap<>(context.maxTopologyMessageTimeout(), callback);
    _collector = collector;
    _delegate.prepare(config, context, new OutputCollector(new CoordinatedOutputCollector(collector)));
    for (String component : Utils.get(context.getThisTargets(), Constants.COORDINATED_STREAM_ID, new HashMap<String, Grouping>()).keySet()) {
        for (Integer task : context.getComponentTasks(component)) {
            _countOutTasks.add(task);
        }
    }
    if (!_sourceArgs.isEmpty()) {
        _numSourceReports = 0;
        for (Entry<String, SourceArgs> entry : _sourceArgs.entrySet()) {
            if (entry.getValue().singleCount) {
                _numSourceReports += 1;
            } else {
                _numSourceReports += context.getComponentTasks(entry.getKey()).size();
            }
        }
    }
}

开发者ID:alibaba，项目名称:jstorm，代码行数:25，代码来源:CoordinatedBolt.java

示例10: createNimbusData

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
private NimbusData createNimbusData(Map conf, INimbus inimbus)
		throws Exception {

	TimeCacheMap.ExpiredCallback<Object, Object> expiredCallback = new TimeCacheMap.ExpiredCallback<Object, Object>() {
		@Override
		public void expire(Object key, Object val) {
			try {
				LOG.info("Close file " + String.valueOf(key));
				if (val != null) {
					if (val instanceof Channel) {
						Channel channel = (Channel) val;
						channel.close();
					} else if (val instanceof BufferFileInputStream) {
						BufferFileInputStream is = (BufferFileInputStream) val;
						is.close();
					}
				}
			} catch (IOException e) {
				LOG.error(e.getMessage(), e);
			}

		}
	};

	int file_copy_expiration_secs = JStormUtils.parseInt(
			conf.get(Config.NIMBUS_FILE_COPY_EXPIRATION_SECS), 30);
	TimeCacheMap<Object, Object> uploaders = new TimeCacheMap<Object, Object>(
			file_copy_expiration_secs, expiredCallback);
	TimeCacheMap<Object, Object> downloaders = new TimeCacheMap<Object, Object>(
			file_copy_expiration_secs, expiredCallback);

	// Callback callback=new TimerCallBack();
	// StormTimer timer=Timer.mkTimerTimer(callback);
	NimbusData data = new NimbusData(conf, downloaders, uploaders, inimbus);

	return data;

}

开发者ID:zhangjunfang，项目名称:jstorm-0.9.6.3-，代码行数:39，代码来源:NimbusServer.java

示例11: uploadChunk

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
/**
 * uploading topology jar data
 */
@Override
public void uploadChunk(String location, ByteBuffer chunk)
		throws TException {
	TimeCacheMap<Object, Object> uploaders = data.getUploaders();
	Object obj = uploaders.get(location);
	if (obj == null) {
		throw new TException(
				"File for that location does not exist (or timed out) "
						+ location);
	}
	try {
		if (obj instanceof WritableByteChannel) {
			WritableByteChannel channel = (WritableByteChannel) obj;
			channel.write(chunk);
			uploaders.put(location, channel);
		} else {
			throw new TException("Object isn't WritableByteChannel for "
					+ location);
		}
	} catch (IOException e) {
		String errMsg = " WritableByteChannel write filed when uploadChunk "
				+ location;
		LOG.error(errMsg);
		throw new TException(e);
	}

}

开发者ID:zhangjunfang，项目名称:jstorm-0.9.6.3-，代码行数:31，代码来源:ServiceHandler.java

示例12: downloadChunk

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public ByteBuffer downloadChunk(String id) throws TException {
	TimeCacheMap<Object, Object> downloaders = data.getDownloaders();
	Object obj = downloaders.get(id);
	if (obj == null) {
		throw new TException("Could not find input stream for that id");
	}

	try {
		if (obj instanceof BufferFileInputStream) {

			BufferFileInputStream is = (BufferFileInputStream) obj;
			byte[] ret = is.read();
			if (ret != null) {
				downloaders.put(id, is);
				return ByteBuffer.wrap(ret);
			}
		} else {
			throw new TException("Object isn't BufferFileInputStream for "
					+ id);
		}
	} catch (IOException e) {
		LOG.error("BufferFileInputStream read failed when downloadChunk ",
				e);
		throw new TException(e);
	}
	byte[] empty = {};
	return ByteBuffer.wrap(empty);
}

开发者ID:zhangjunfang，项目名称:jstorm-0.9.6.3-，代码行数:30，代码来源:ServiceHandler.java

示例13: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public void prepare(Map conf, TopologyContext context,
		OutputCollector collector) {
	_fieldLocations = new HashMap<String, GlobalStreamId>();
	_collector = collector;
	int timeout = ((Number) conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS))
			.intValue();
	_pending = new TimeCacheMap<List<Object>, Map<GlobalStreamId, Tuple>>(
			timeout, new ExpireCallback());
	_numSources = context.getThisSources().size();
	Set<String> idFields = null;
	for (GlobalStreamId source : context.getThisSources().keySet()) {
		Fields fields = context.getComponentOutputFields(
				source.get_componentId(), source.get_streamId());
		Set<String> setFields = new HashSet<String>(fields.toList());
		if (idFields == null)
			idFields = setFields;
		else
			idFields.retainAll(setFields);

		for (String outfield : _outFields) {
			for (String sourcefield : fields) {
				if (outfield.equals(sourcefield)) {
					_fieldLocations.put(outfield, source);
				}
			}
		}
	}
	_idFields = new Fields(new ArrayList<String>(idFields));

	if (_fieldLocations.size() != _outFields.size()) {
		throw new RuntimeException(
				"Cannot find all outfields among sources");
	}
}

开发者ID:zhangjunfang，项目名称:jstorm-0.9.6.3-，代码行数:36，代码来源:SingleJoinBolt.java

示例14: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public void prepare(Map config, TopologyContext context,
		OutputCollector collector) {
	TimeCacheMap.ExpiredCallback<Object, TrackingInfo> callback = null;
	if (_delegate instanceof TimeoutCallback) {
		callback = new TimeoutItems();
	}
	_tracked = new TimeCacheMap<Object, TrackingInfo>(
			context.maxTopologyMessageTimeout(), callback);
	_collector = collector;
	_delegate.prepare(config, context, new OutputCollector(
			new CoordinatedOutputCollector(collector)));
	for (String component : Utils.get(context.getThisTargets(),
			Constants.COORDINATED_STREAM_ID,
			new HashMap<String, Grouping>()).keySet()) {
		for (Integer task : context.getComponentTasks(component)) {
			_countOutTasks.add(task);
		}
	}
	if (!_sourceArgs.isEmpty()) {
		_numSourceReports = 0;
		for (Entry<String, SourceArgs> entry : _sourceArgs.entrySet()) {
			if (entry.getValue().singleCount) {
				_numSourceReports += 1;
			} else {
				_numSourceReports += context.getComponentTasks(
						entry.getKey()).size();
			}
		}
	}
}

开发者ID:zhangjunfang，项目名称:jstorm-0.9.6.3-，代码行数:32，代码来源:CoordinatedBolt.java

示例15: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public void prepare(Map stormConf, TopologyContext context) {
	this.conf = stormConf;

	int timeoutSeconds = JStormUtils.parseInt(
			conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS), 30);
	counters = new TimeCacheMap<BatchId, AtomicLong>(timeoutSeconds);
	
	LOG.info("Successfully do prepare");

}

开发者ID:zhangjunfang，项目名称:jstorm-0.9.6.3-，代码行数:12，代码来源:CountBolt.java

示例16: createNimbusData

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
/**
 * 1.此处定义一个TimcacheMap缓存失效时候的callback函数，关闭channel和stream
 * 建立上传和下载两个缓存，缓存流
 * 新建一个Nimbus对象
 */
@SuppressWarnings("rawtypes")
private NimbusData createNimbusData(Map conf, INimbus inimbus) throws Exception {

    TimeCacheMap.ExpiredCallback<Object, Object> expiredCallback = new TimeCacheMap.ExpiredCallback<Object, Object>() {
        @Override
        public void expire(Object key, Object val) {
            try {
                LOG.info("Close file " + String.valueOf(key));
                if (val != null) {
                    if (val instanceof Channel) {
                        Channel channel = (Channel) val;
                        channel.close();
                    } else if (val instanceof BufferFileInputStream) {
                        BufferFileInputStream is = (BufferFileInputStream) val;
                        is.close();
                    }
                }
            } catch (IOException e) {
                LOG.error(e.getMessage(), e);
            }

        }
    };

    int file_copy_expiration_secs = JStormUtils.parseInt(conf.get(Config.NIMBUS_FILE_COPY_EXPIRATION_SECS), 30);
    TimeCacheMap<Object, Object> uploaders = new TimeCacheMap<Object, Object>(file_copy_expiration_secs,
            expiredCallback);
    TimeCacheMap<Object, Object> downloaders = new TimeCacheMap<Object, Object>(file_copy_expiration_secs,
            expiredCallback);

    // Callback callback=new TimerCallBack();
    // StormTimer timer=Timer.mkTimerTimer(callback);
    NimbusData data = new NimbusData(conf, downloaders, uploaders, inimbus);

    return data;

}

开发者ID:songtk，项目名称:learn_jstorm，代码行数:43，代码来源:NimbusServer.java

示例17: downloadChunk

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public ByteBuffer downloadChunk(String id) throws TException {
	TimeCacheMap<Object, Object> downloaders = data.getDownloaders();
	Object obj = downloaders.get(id);
	if (obj == null) {
		throw new TException("Could not find input stream for that id");
	}

	try {
		if (obj instanceof BufferFileInputStream) {

			BufferFileInputStream is = (BufferFileInputStream) obj;
			byte[] ret = is.read();
			if (ret != null) {
				downloaders.put(id, (BufferFileInputStream) is);
				return ByteBuffer.wrap(ret);
			}
		} else {
			throw new TException("Object isn't BufferFileInputStream for "
					+ id);
		}
	} catch (IOException e) {
		LOG.error("BufferFileInputStream read failed when downloadChunk ",
				e);
		throw new TException(e);
	}
	byte[] empty = {};
	return ByteBuffer.wrap(empty);
}

开发者ID:songtk，项目名称:learn_jstorm，代码行数:30，代码来源:ServiceHandler.java

示例18: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
public void prepare(Map config, TopologyContext context,
		OutputCollector collector) {
	TimeCacheMap.ExpiredCallback<Object, TrackingInfo> callback = null;
	if (_delegate instanceof TimeoutCallback) {
		callback = new TimeoutItems();
	}
	_tracked = new TimeCacheMap<Object, TrackingInfo>(
			context.maxTopologyMessageTimeout(), callback);
	_collector = collector;
	_delegate.prepare(config, context, new OutputCollector(
			new CoordinatedOutputCollector(collector)));
	for (String component : Utils.get(context.getThisTargets(),
			Constants.COORDINATED_STREAM_ID,
			new HashMap<String, Grouping>()).keySet()) {
		for (Integer task : context.getComponentTasks(component)) {
			_countOutTasks.add(task);
		}
	}
	if (!_sourceArgs.isEmpty()) {
		_numSourceReports = 0;
		for (Entry<String, SourceArgs> entry : _sourceArgs.entrySet()) {
			if (entry.getValue().singleCount) {
				_numSourceReports += 1;
			} else {
				_numSourceReports += context.getComponentTasks(
						entry.getKey()).size();
			}
		}
	}
}

开发者ID:songtk，项目名称:learn_jstorm，代码行数:31，代码来源:CoordinatedBolt.java

示例19: prepare

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
  _fieldLocations = new HashMap<String, GlobalStreamId>();
  _collector = collector;
  int timeout = ((Number) conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS)).intValue();
  _pending = new TimeCacheMap<List<Object>, Map<GlobalStreamId, Tuple>>(timeout, new ExpireCallback());
  _numSources = context.getThisSources().size();
  Set<String> idFields = null;
  for (GlobalStreamId source : context.getThisSources().keySet()) {
    Fields fields = context.getComponentOutputFields(source.get_componentId(), source.get_streamId());
    Set<String> setFields = new HashSet<String>(fields.toList());
    if (idFields == null)
      idFields = setFields;
    else
      idFields.retainAll(setFields);

    for (String outfield : _outFields) {
      for (String sourcefield : fields) {
        if (outfield.equals(sourcefield)) {
          _fieldLocations.put(outfield, source);
        }
      }
    }
  }
  _idFields = new Fields(new ArrayList<String>(idFields));

  if (_fieldLocations.size() != _outFields.size()) {
    throw new RuntimeException("Cannot find all outfields among sources");
  }
}

开发者ID:luozhaoyu，项目名称:big-data-system，代码行数:31，代码来源:SingleJoinBolt.java

示例20: downloadChunk

import backtype.storm.utils.TimeCacheMap; //导入依赖的package包/类
@Override
public ByteBuffer downloadChunk(String id) throws TException {
    TimeCacheMap<Object, Object> downloaders = data.getDownloaders();
    Object obj = downloaders.get(id);
    if (obj == null) {
        throw new TException("Could not find input stream for that id");
    }

    try {
        if (obj instanceof BufferFileInputStream) {

            BufferFileInputStream is = (BufferFileInputStream) obj;
            byte[] ret = is.read();
            if (ret != null) {
                downloaders.put(id, is);
                return ByteBuffer.wrap(ret);
            }
        } else {
            throw new TException("Object isn't BufferFileInputStream for " + id);
        }
    } catch (IOException e) {
        LOG.error("BufferFileInputStream read failed when downloadChunk ", e);
        throw new TException(e);
    }
    byte[] empty = {};
    return ByteBuffer.wrap(empty);
}

开发者ID:kkllwww007，项目名称:jstrom，代码行数:28，代码来源:ServiceHandler.java

注：本文中的backtype.storm.utils.TimeCacheMap类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。