基于Disruptor时间滑窗统计redis热度（简约篇）

Bertha 。 2022-11-28 00:20 147阅读 0赞

### 文章目录 ###

*  redis热点key实现方案
 *  时间滑窗统计热度
 *   *  代码
     *  讲解
     *  打印结果
     *  优化点
 *  参考文章

# redis热点key实现方案 #

*  [有赞透明多级缓存解决方案（TMC）][TMC]

大家可以看下里面的思路：

1.  就是get的时候如果是热点key走本地缓存，如果不是走redis集群。
2.  热点key发现就是get的时候进行上报统计，以及利用时间滑窗进行统计
3.  set，expire,del会进行相应的保存起来，如果是热点key就进行监听，进行相应的处理。
4.  本地缓存跟redis缓存一致性问题，就是如果是主动删除的话最好两边都删除。如果是过期又是热点key就不要让它去访问redis集群了，一旦后台有更新，顺便把本地缓存更新即可。

# 时间滑窗统计热度 #

这个我还是蛮感兴趣的，尝试了一下如何去实现这个功能。

首先是时间轮的数据结构吧，让我想起Disruptor ，美团技术篇也有介绍。

*  [高性能队列——Disruptor][Disruptor]

## 代码 ##

import com.lmax.disruptor.*;
    import com.lmax.disruptor.dsl.Disruptor;
    import com.lmax.disruptor.dsl.ProducerType;
    
    import java.time.Duration;
    import java.time.LocalDateTime;
    import java.util.Map;
    import java.util.Random;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ThreadFactory;
    
    public class DisruptorMain { 
        public static void main(String[] args) throws Exception { 
            // 队列中的元素
            class Element { 
    
                private Map<String, Integer> map = new ConcurrentHashMap<>(100);
    
                public Map<String, Integer> get() { 
                    return map;
                }
    
                public void set(String key) { 
                    if (map.containsKey(key)) { 
                        map.put(key, map.get(key) + 1);
                    } else { 
                        map.put(key, 1);
                    }
                }
    
            }
    
            // 生产者的线程工厂
            ThreadFactory threadFactory = new ThreadFactory() { 
                @Override
                public Thread newThread(Runnable r) { 
                    return new Thread(r, "simpleThread");
                }
            };
    
            // RingBuffer生产工厂,初始化RingBuffer的时候使用
            EventFactory<Element> factory = new EventFactory<Element>() { 
                @Override
                public Element newInstance() { 
                    return new Element();
                }
            };
    
            // 处理Event的handler
            EventHandler<Element> handler = new EventHandler<Element>() { 
                @Override
                public void onEvent(Element element, long sequence, boolean endOfBatch) throws InterruptedException { 
                    //Thread.sleep(1000);
                    System.out.println(LocalDateTime.now() + "----" + "Element: " + element.get());
                }
            };
    
            // 阻塞策略
            BlockingWaitStrategy strategy = new BlockingWaitStrategy();
    
            // 指定RingBuffer的大小
            int bufferSize = 16;
    
            // 创建disruptor，采用单生产者模式
            Disruptor<Element> disruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);
    
            // 设置EventHandler
            disruptor.handleEventsWith(handler);
    
            // 启动disruptor的线程
            disruptor.start();
    
            RingBuffer<Element> ringBuffer = disruptor.getRingBuffer();
    
            LocalDateTime time = LocalDateTime.now();
            // 获取下一个可用位置的下标
            long sequence = ringBuffer.next();
            for (int i = 0; i < 1000; i++) { 
                try { 
                    LocalDateTime time1 = LocalDateTime.now();
                    Duration duration = Duration.between(time, time1);
                    if (duration.getSeconds() >= 3) { 
                        ringBuffer.publish(sequence);
                        sequence = ringBuffer.next();
                        time = LocalDateTime.now();
                    }
                    // 返回可用位置的元素
                    Element event = ringBuffer.get(sequence);
                    // 设置该位置元素的值
                    event.set(String.valueOf(new Random().nextInt(20)));
                } finally { 
                    Thread.sleep(100);
                }
            }
    
        }
    
    }

这里主要基于有赞TMC设计的3秒为维度进行统计的需求。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70_pic_center]

## 讲解 ##

LocalDateTime time = LocalDateTime.now(); 1
            // 获取下一个可用位置的下标
            long sequence = ringBuffer.next(); 2
            for (int i = 0; i < 1000; i++) { 
                try { 
                    LocalDateTime time1 = LocalDateTime.now(); 3
                    Duration duration = Duration.between(time, time1); 4
                    if (duration.getSeconds() >= 3) { 
                        ringBuffer.publish(sequence); 5
                        sequence = ringBuffer.next(); 
                        time = LocalDateTime.now(); 
                    }
                    // 返回可用位置的元素
                    Element event = ringBuffer.get(sequence);
                    // 设置该位置元素的值
                    event.set(String.valueOf(new Random().nextInt(20))); 6
                } finally { 
                    Thread.sleep(100);
                }
            }

第一步：获取 当前时间点  
第二步：获取下一个节点  
第三，四步：获取插入是时候的时间点，判断之间的时间相差秒数  
第五步：如果相差超过3秒，则需要把之前的publish，然后获取下一个节点，还有时间也换成当前时间点。  
第6步：模拟一下redis key查询，set到这个时间滑窗里头

## 打印结果 ##

2020-08-16T14:16:21.343----Element: { 11=2, 12=2, 13=1, 14=4, 15=1, 16=3, 17=2, 18=1, 19=1, 0=2, 1=1, 2=1, 3=1, 5=2, 6=2, 7=1, 8=1, 9=2}
    2020-08-16T14:16:24.359----Element: { 10=2, 13=1, 15=2, 16=3, 17=3, 18=2, 19=2, 0=3, 2=2, 3=1, 4=1, 5=4, 6=1, 7=2, 8=1}
    2020-08-16T14:16:27.375----Element: { 10=1, 12=3, 14=2, 16=4, 17=1, 18=1, 19=1, 0=2, 2=2, 3=1, 5=1, 6=2, 7=2, 8=4, 9=3}
    2020-08-16T14:16:30.396----Element: { 10=2, 12=1, 13=3, 14=1, 15=2, 16=1, 18=1, 19=1, 0=3, 1=2, 2=1, 3=1, 4=1, 5=2, 6=2, 7=3, 8=2, 9=1}
    2020-08-16T14:16:33.411----Element: { 10=3, 11=1, 13=3, 14=1, 15=3, 16=1, 19=1, 0=1, 1=4, 2=1, 3=4, 4=1, 5=2, 6=2, 7=1, 9=1}
    2020-08-16T14:16:36.426----Element: { 10=3, 11=1, 12=4, 13=2, 14=1, 15=2, 19=2, 0=1, 1=2, 2=3, 4=2, 5=2, 6=2, 7=1, 8=2}
    2020-08-16T14:16:39.440----Element: { 10=2, 11=1, 12=4, 13=1, 14=1, 15=3, 16=1, 17=1, 19=1, 0=1, 1=1, 2=1, 4=3, 6=4, 7=1, 8=2, 9=2}

就是每3秒打印下，然后消费者进行统计。

## 优化点 ##

就是在消费者端，可以使用kafka或者mq等等消费能力高的，进行后续的统计记录。

# 参考文章 #

*  [有赞透明多级缓存解决方案（TMC）][TMC]
 *  [高性能队列——Disruptor][Disruptor]

[TMC]: https://segmentfault.com/a/1190000017142556
[Disruptor]: https://tech.meituan.com/2016/11/18/disruptor.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/8245d9634d184859abe852b3a8bee299.png