`
dalan_123
  • 浏览: 87050 次
  • 性别: Icon_minigender_1
  • 来自: 郑州
社区版块
存档分类
最新评论

jstorm源码之RotatingMap

 
阅读更多
一、作用
      基于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;
    }
}
分享到:
评论

相关推荐

    jstorm源码解析之bolt异常处理方法.docx

    在JStorm的源码中,我们可以看到Bolt执行的核心逻辑位于`BasicBoltExecutor`类的`execute()`方法。该方法首先设置上下文,然后尝试执行Bolt的业务逻辑,最后确认处理成功。如果在执行过程中抛出`FailedException`,...

    Storm 源码分析

    - **Nimbus**:Nimbus是Storm集群的核心组件之一,负责整个集群的管理和协调工作,包括任务调度、故障恢复等。Nimbus通过Zookeeper来实现状态同步和集群协调。 - **Supervisor**:Supervisor运行在每个Worker节点上...

    jstorm源码解析之bolt异常处理方法

    在本篇文章中,我们将深入探讨JStorm源码中关于Bolt异常处理方法的实现细节。首先,我们了解到如果在Bolt代码中出现未被捕获的异常,所在的工作进程会直接退出。这一点与Storm的设计相似,旨在避免问题被掩盖而无法...

    JStorm:JStorm原始码学习-源码包

    JStorm原始码学习:主要包含Storm重新启动,Nimbus启动,Supervisor启动,Executor创建和启动 风暴编程模型 Nimbus:负责资源分配和任务调度。 主管:负责接受nimbus分配的任务,启动和停止属于自己管理的worker...

    jstorm storm入门demo

    1. **JStorm与Storm的关系**:讲解JStorm与Apache Storm的关系,包括JStorm的诞生背景、主要改进和优化之处。 2. **JStorm的核心概念**:介绍Spout、Bolt、拓扑等核心概念,解释它们在实时数据处理中的作用。 3. *...

    大数据技术分享 JStorm介绍 JStorm-分布式实时计算引擎 共40页.pptx

    **JStorm介绍** JStorm是由阿里巴巴开发的分布式实时计算引擎,它是基于Apache Storm的一个增强版本。JStorm的设计目标是提供比Storm更高的稳定性、更强的功能和更快的计算性能。这个技术分享主要涵盖了JStorm的...

    阿里中间件性能挑战赛初赛-RocketMQ+JStorm+Tair实时统计双11交易金额源码+学习说明.zip

    阿里中间件性能挑战赛初赛-RocketMQ+JStorm+Tair实时统计双11交易金额源码+学习说明.zip阿里中间件性能挑战赛初赛-RocketMQ+JStorm+Tair实时统计双11交易金额源码+学习说明.zip阿里中间件性能挑战赛初赛-RocketMQ+...

    jstorm集成kafka代码实例

    本示例将探讨如何将Apache JStorm与Kafka进行集成,以利用它们各自的优势。JStorm是一款分布式实时计算系统,它允许用户处理数据流,而Kafka则是一个高吞吐量的分布式消息系统,用于实时数据传递。接下来,我们将...

    jstorm集成kafka插件demo

    在IT行业中,分布式计算系统和实时数据处理是关键领域,其中JStorm和Kafka都是重要的组件。本示例探讨了如何将JStorm与Kafka进行集成,以实现高效的数据流处理。下面,我们将深入理解这两个工具,以及它们结合使用的...

    jstorm 阿里巴巴官方文档 pdf

    接着,文档会详细阐述JStorm的安装与配置过程,包括下载源码、编译、部署以及设置环境变量。对于初学者,这部分内容至关重要,因为正确配置JStorm环境是后续开发和运行的基础。 在JStorm的使用方法部分,你会了解到...

    JStorm 2.1.1 API

    **JStorm 2.1.1 API 深度解析** JStorm是阿里巴巴开源的一款分布式实时计算系统,它基于Apache Storm并进行了大量的优化,提供了一种高效、稳定且易用的流处理框架。JStorm 2.1.1是其一个重要的版本,此版本的API为...

    jstorm example

    **JStorm简介** JStorm是阿里巴巴开源的一款分布式实时计算系统,它是基于Apache Storm的一个高性能、高可用、热扩展的实时处理框架。JStorm的核心设计理念是简单、高效和稳定,能够处理大规模的数据流处理任务,...

    jstorm文档

    JStorm是中国淘宝团队开发的一款分布式实时计算系统,它是基于Apache Storm的设计理念,但在性能、稳定性和易用性上进行了大量的优化。本文档将详细介绍JStorm的基础知识,包括其核心概念、工作原理、安装配置以及...

    jstorm课程

    标题"jstorm课程"指的是关于JStorm的教育课程,JStorm是阿里巴巴开源的一个分布式实时计算框架,它是Apache Storm的Java版本,主要用于处理大规模数据流的实时计算。 描述中提到的"全套storm资料初学者必备 比较...

    jstorm-2.2.1

    《JStorm 2.2.1:分布式流处理框架详解》 JStorm是阿里巴巴开源的一款高性能、高可靠的分布式实时计算系统,它基于Apache Storm并针对大规模数据处理进行了优化。JStorm 2.2.1是该框架的一个稳定版本,提供了许多...

    storm-core源码

    storm-core-1.0.3-sources.jar 源码文件,1.0.3版本

    Jstorm 到 Apache Flink 的迁移实践.pdf

    1. **内存隔离与资源管理**:Flink作业运行在YARN之上,提供了内存隔离,确保了作业间的资源独立,同时YARN的队列机制便于进行资源分配和预算审核。 2. **统一资源管理**:通过YARN管理资源,减轻了运维负担,实现了...

    Jstorm介绍

    jstorm框架介绍,包含架构图、jstorm安装部署以及配置、如何在jstorm框架里写业务代码。

    alibaba-jstorm

    阿里巴巴JStorm是一款由阿里巴巴开源的分布式实时计算系统,它基于Apache Storm,但在性能、稳定性、易用性等多个方面进行了优化,是大数据处理领域的重要工具之一。JStorm的设计理念是“简单、稳定、低延迟”,旨在...

    JStorm 2.2.1下载 (百度网盘提取链接和密码)

    JStorm 2.2.1 百度网盘下载,Alibaba JStorm 是一个强大的企业级流式计算引擎

Global site tag (gtag.js) - Google Analytics