- 浏览: 87050 次
- 性别:
- 来自: 郑州
文章分类
- 全部博客 (69)
- java (28)
- linux (6)
- redis (4)
- C# (3)
- 架构 (10)
- java ee (1)
- web (1)
- 操作系统 (7)
- sqlserver (1)
- android (2)
- Hadoop (12)
- 大数据 (21)
- 分布式 事务 消息 (10)
- linux mysql (1)
- 数据库 (3)
- 关于hadoop之bootshell使用 (2)
- 关于hbase---HTableInterfaceFactory (1)
- Spring (3)
- Hbase (5)
- jstorm (10)
- nginx (1)
- 分布式 (1)
- 区块链 (3)
- dubbo (1)
- nacos (1)
- 阿里 (1)
- go (3)
- 缓存 (1)
- memcached (1)
- ssdb (1)
- 源码 (1)
最新评论
-
想个可以用的名字:
楼主,能不能给发一份源代码,1300246542@qqq.co ...
spring+websocket的使用 -
wahahachuang5:
web实时推送技术使用越来越广泛,但是自己开发又太麻烦了,我觉 ...
websocket -
dalan_123:
前提是你用的是spring mvc 才需要加的1、在web.x ...
spring+websocket的使用 -
string2020:
CharacterEncodingFilter这个filter ...
spring+websocket的使用
一、作用
基于LinkedList + HashMap<K,V>实现一个循环Map
二、源码
// 通过结合LinkedList 和 HashMap 构成一个循环Map数据结构
public class RotatingMap<K, V> {
// this default ensures things expire at most 50% past the expiration time
// 默认bucket数目为3
private static final int DEFAULT_NUM_BUCKETS = 3;
// 定义内部接口 key 到期的回调操作
public static interface ExpiredCallback<K, V> {
public void expire(K key, V val);
}
// 通过链表维护 一个 buckets 结构
private LinkedList<HashMap<K, V>> _buckets;
private ExpiredCallback _callback;
// 构造RotationgMap的 bucket数目不能少于2
public RotatingMap(int numBuckets, ExpiredCallback<K, V> callback) {
if (numBuckets < 2) {
throw new IllegalArgumentException("numBuckets must be >= 2");
}
_buckets = new LinkedList<HashMap<K, V>>();
for (int i = 0; i < numBuckets; i++) {
_buckets.add(new HashMap<K, V>());
}
_callback = callback;
}
public RotatingMap(ExpiredCallback<K, V> callback) {
this(DEFAULT_NUM_BUCKETS, callback);
}
public RotatingMap(int numBuckets) {
this(numBuckets, null);
}
// 执行rotate
// 首先获取LinkedList 末尾元素
// 其次在first位置 添加占位新元素
// 结合expire的callback
// 最末尾的batch多半是时间最久 极可能是超出有效期的
public Map<K, V> rotate() {
Map<K, V> dead = _buckets.removeLast();
_buckets.addFirst(new HashMap<K, V>());
if (_callback != null) {
for (Entry<K, V> entry : dead.entrySet()) {
_callback.expire(entry.getKey(), entry.getValue());
}
}
return dead;
}
// 是否存在key
public boolean containsKey(K key) {
for (HashMap<K, V> bucket : _buckets) {
if (bucket.containsKey(key)) {
return true;
}
}
return false;
}
// 获取value
public V get(K key) {
for (HashMap<K, V> bucket : _buckets) {
if (bucket.containsKey(key)) {
return bucket.get(key);
}
}
return null;
}
// 添加item
public void put(K key, V value) {
Iterator<HashMap<K, V>> it = _buckets.iterator();
HashMap<K, V> bucket = it.next();
bucket.put(key, value);
while (it.hasNext()) {
bucket = it.next();
bucket.remove(key);
}
}
// 移除item
public Object remove(K key) {
for (HashMap<K, V> bucket : _buckets) {
if (bucket.containsKey(key)) {
return bucket.remove(key);
}
}
return null;
}
// 总容量
public int size() {
int size = 0;
for (HashMap<K, V> bucket : _buckets) {
size += bucket.size();
}
return size;
}
}
基于LinkedList + HashMap<K,V>实现一个循环Map
二、源码
// 通过结合LinkedList 和 HashMap 构成一个循环Map数据结构
public class RotatingMap<K, V> {
// this default ensures things expire at most 50% past the expiration time
// 默认bucket数目为3
private static final int DEFAULT_NUM_BUCKETS = 3;
// 定义内部接口 key 到期的回调操作
public static interface ExpiredCallback<K, V> {
public void expire(K key, V val);
}
// 通过链表维护 一个 buckets 结构
private LinkedList<HashMap<K, V>> _buckets;
private ExpiredCallback _callback;
// 构造RotationgMap的 bucket数目不能少于2
public RotatingMap(int numBuckets, ExpiredCallback<K, V> callback) {
if (numBuckets < 2) {
throw new IllegalArgumentException("numBuckets must be >= 2");
}
_buckets = new LinkedList<HashMap<K, V>>();
for (int i = 0; i < numBuckets; i++) {
_buckets.add(new HashMap<K, V>());
}
_callback = callback;
}
public RotatingMap(ExpiredCallback<K, V> callback) {
this(DEFAULT_NUM_BUCKETS, callback);
}
public RotatingMap(int numBuckets) {
this(numBuckets, null);
}
// 执行rotate
// 首先获取LinkedList 末尾元素
// 其次在first位置 添加占位新元素
// 结合expire的callback
// 最末尾的batch多半是时间最久 极可能是超出有效期的
public Map<K, V> rotate() {
Map<K, V> dead = _buckets.removeLast();
_buckets.addFirst(new HashMap<K, V>());
if (_callback != null) {
for (Entry<K, V> entry : dead.entrySet()) {
_callback.expire(entry.getKey(), entry.getValue());
}
}
return dead;
}
// 是否存在key
public boolean containsKey(K key) {
for (HashMap<K, V> bucket : _buckets) {
if (bucket.containsKey(key)) {
return true;
}
}
return false;
}
// 获取value
public V get(K key) {
for (HashMap<K, V> bucket : _buckets) {
if (bucket.containsKey(key)) {
return bucket.get(key);
}
}
return null;
}
// 添加item
public void put(K key, V value) {
Iterator<HashMap<K, V>> it = _buckets.iterator();
HashMap<K, V> bucket = it.next();
bucket.put(key, value);
while (it.hasNext()) {
bucket = it.next();
bucket.remove(key);
}
}
// 移除item
public Object remove(K key) {
for (HashMap<K, V> bucket : _buckets) {
if (bucket.containsKey(key)) {
return bucket.remove(key);
}
}
return null;
}
// 总容量
public int size() {
int size = 0;
for (HashMap<K, V> bucket : _buckets) {
size += bucket.size();
}
return size;
}
}
发表评论
-
nacos单机源码调试
2018-12-17 11:35 1219首先从github上获取对应的源码Nacos源码git cl ... -
jstorm源码之TransactionalState
2016-03-21 19:31 888一、作用 主要是通过结合zookeeper,在zookee ... -
jstorm源码之RotatingTransactionalState
2016-03-21 19:29 581一、作用 构建一个Rotationg transacti ... -
jstorm源码之PartitionedTridentSpoutExecutor
2016-03-21 19:28 887一、作用 Partition Spout对应的exec ... -
jstorm源码之 RichSpoutBatchExecutor
2016-03-21 19:28 0一、作用 RichSpoutBatchExecutor是IRi ... -
jstorm源码之 RichSpoutBatchExecutor
2016-03-21 19:24 615一、作用 RichSpoutBatchExecutor是IRi ... -
jstorm源码之TridentTopology
2016-03-16 18:12 2363在jstorm中对应TridentTopology的源码如下, ... -
jstorm操作命令
2016-03-15 18:04 2736启动ZOOPKEEPER zkServer.sh start ... -
JStorm之Supervisor简介
2016-03-15 18:02 1248一、简介Supervisor是JStorm中的工作节点,类似 ... -
JStorm介绍
2016-03-15 17:56 918一、简介Storm是开源的 ... -
mycat的使用---sqlserver和mysql
2016-01-11 14:33 8617数据库中间件mycat的使 ... -
jstorm安装
2015-12-03 19:43 1747关于jstorm单机安装可以 ... -
HBase系列一
2015-11-30 16:17 711关于hbase 一、客户端类 HTable 和 HTabl ... -
spring hadoop系列(六)---HbaseSystemException
2015-11-30 09:13 497一、源码 /** * HBase Data Access e ... -
spring hadoop系列(五)---spring hadoop hbase之HbaseSynchronizationManager
2015-11-27 18:16 873一、源码如下 /** * Synchronization m ... -
spring hadoop 系列(二)
2015-11-27 15:26 597一、源码分析 /** * * HbaseAccesso ... -
spring hadoop之batch处理(二)
2015-11-24 18:10 1522一、测试 public class MrBatchApp { ... -
spring hadoop之mapreduce batch
2015-11-24 15:51 635一、测试 // 定义hadoop configuration ... -
centos6.7 64位 伪分布 安装 cdh5.4.8 + jdk 8
2015-11-09 00:37 2320一、安装JAVA # 创建JAVA的目录 mkdir -p / ... -
spring hadoop系列二(MapReduce and Distributed cache)
2015-11-06 15:22 986关于MapReduce and Distributed Ca ...
相关推荐
在JStorm的源码中,我们可以看到Bolt执行的核心逻辑位于`BasicBoltExecutor`类的`execute()`方法。该方法首先设置上下文,然后尝试执行Bolt的业务逻辑,最后确认处理成功。如果在执行过程中抛出`FailedException`,...
- **Nimbus**:Nimbus是Storm集群的核心组件之一,负责整个集群的管理和协调工作,包括任务调度、故障恢复等。Nimbus通过Zookeeper来实现状态同步和集群协调。 - **Supervisor**:Supervisor运行在每个Worker节点上...
在本篇文章中,我们将深入探讨JStorm源码中关于Bolt异常处理方法的实现细节。首先,我们了解到如果在Bolt代码中出现未被捕获的异常,所在的工作进程会直接退出。这一点与Storm的设计相似,旨在避免问题被掩盖而无法...
JStorm原始码学习:主要包含Storm重新启动,Nimbus启动,Supervisor启动,Executor创建和启动 风暴编程模型 Nimbus:负责资源分配和任务调度。 主管:负责接受nimbus分配的任务,启动和停止属于自己管理的worker...
1. **JStorm与Storm的关系**:讲解JStorm与Apache Storm的关系,包括JStorm的诞生背景、主要改进和优化之处。 2. **JStorm的核心概念**:介绍Spout、Bolt、拓扑等核心概念,解释它们在实时数据处理中的作用。 3. *...
**JStorm介绍** JStorm是由阿里巴巴开发的分布式实时计算引擎,它是基于Apache Storm的一个增强版本。JStorm的设计目标是提供比Storm更高的稳定性、更强的功能和更快的计算性能。这个技术分享主要涵盖了JStorm的...
阿里中间件性能挑战赛初赛-RocketMQ+JStorm+Tair实时统计双11交易金额源码+学习说明.zip阿里中间件性能挑战赛初赛-RocketMQ+JStorm+Tair实时统计双11交易金额源码+学习说明.zip阿里中间件性能挑战赛初赛-RocketMQ+...
本示例将探讨如何将Apache JStorm与Kafka进行集成,以利用它们各自的优势。JStorm是一款分布式实时计算系统,它允许用户处理数据流,而Kafka则是一个高吞吐量的分布式消息系统,用于实时数据传递。接下来,我们将...
在IT行业中,分布式计算系统和实时数据处理是关键领域,其中JStorm和Kafka都是重要的组件。本示例探讨了如何将JStorm与Kafka进行集成,以实现高效的数据流处理。下面,我们将深入理解这两个工具,以及它们结合使用的...
接着,文档会详细阐述JStorm的安装与配置过程,包括下载源码、编译、部署以及设置环境变量。对于初学者,这部分内容至关重要,因为正确配置JStorm环境是后续开发和运行的基础。 在JStorm的使用方法部分,你会了解到...
**JStorm 2.1.1 API 深度解析** JStorm是阿里巴巴开源的一款分布式实时计算系统,它基于Apache Storm并进行了大量的优化,提供了一种高效、稳定且易用的流处理框架。JStorm 2.1.1是其一个重要的版本,此版本的API为...
**JStorm简介** JStorm是阿里巴巴开源的一款分布式实时计算系统,它是基于Apache Storm的一个高性能、高可用、热扩展的实时处理框架。JStorm的核心设计理念是简单、高效和稳定,能够处理大规模的数据流处理任务,...
JStorm是中国淘宝团队开发的一款分布式实时计算系统,它是基于Apache Storm的设计理念,但在性能、稳定性和易用性上进行了大量的优化。本文档将详细介绍JStorm的基础知识,包括其核心概念、工作原理、安装配置以及...
标题"jstorm课程"指的是关于JStorm的教育课程,JStorm是阿里巴巴开源的一个分布式实时计算框架,它是Apache Storm的Java版本,主要用于处理大规模数据流的实时计算。 描述中提到的"全套storm资料初学者必备 比较...
《JStorm 2.2.1:分布式流处理框架详解》 JStorm是阿里巴巴开源的一款高性能、高可靠的分布式实时计算系统,它基于Apache Storm并针对大规模数据处理进行了优化。JStorm 2.2.1是该框架的一个稳定版本,提供了许多...
storm-core-1.0.3-sources.jar 源码文件,1.0.3版本
1. **内存隔离与资源管理**:Flink作业运行在YARN之上,提供了内存隔离,确保了作业间的资源独立,同时YARN的队列机制便于进行资源分配和预算审核。 2. **统一资源管理**:通过YARN管理资源,减轻了运维负担,实现了...
jstorm框架介绍,包含架构图、jstorm安装部署以及配置、如何在jstorm框架里写业务代码。
阿里巴巴JStorm是一款由阿里巴巴开源的分布式实时计算系统,它基于Apache Storm,但在性能、稳定性、易用性等多个方面进行了优化,是大数据处理领域的重要工具之一。JStorm的设计理念是“简单、稳定、低延迟”,旨在...
JStorm 2.2.1 百度网盘下载,Alibaba JStorm 是一个强大的企业级流式计算引擎