使用BitSet实现日期连续签到

  @GetMapping("/user/sign/{id}")
  public Result<String> userSign(@PathVariable("id") Long id,
                                @RequestParam(value = "date", required = false) 
                                String completeDated) throws IOException {
   
    LocalDateTime now = LocalDateTime.now();
    String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
    String key = "M2B:USER:SIGN" + ":" + id + ":" + keySuffix;
    RBitSet bitSet = redissonClient.getBitSet(key);
    bitSet.expire(50, TimeUnit.MINUTES);
    if (id == 1) {
      // 签到,获取今天是本月的第几天,bitset下标从0开始
      int dayOfMonth = now.getDayOfMonth();
      if (bitSet.get(dayOfMonth - 1)) {
        return Result.accept("您已签到，请勿重复签到");
      } else {
        bitSet.set(dayOfMonth - 1, true);
        return Result.accept("签到成功");
      }
    } else if (id == 2) {
      // 补签
      Date date = DateUtil.parseDate(completeDate);
      Instant instant = date.toInstant();
      ZoneId zoneId = ZoneId.systemDefault();
      LocalDateTime localDateTime = instant.atZone(zoneId).toLocalDateTime();
      //4.获取今天是本月的第几天
      int dayOfMonth = localDateTime.getDayOfMonth();
      //5.写入redis SETBIT key offset 1
      bitSet.set(dayOfMonth - 1, true);
    } else if (id == 3) {
      // 签到天数（标识为1的）
      long cardinality = bitSet.cardinality();
      return Result.accept("总共签到了：" + cardinality + "天");
    } else if (id == 4) {
      // 连续签到，以及签到的天数
      int dayOfMonth = now.getDayOfMonth();
      BitSet bitSet1 = bitSet.asBitSet();
      List<Integer> signList = Lists.newArrayList();
      bitSet1.stream().forEach(signList::add);
      signList.removeIf(b -> b > dayOfMonth);
      boolean continuousSign = signList.stream().anyMatch(c -> c.equals(dayOfMonth));
      if (continuousSign) {
        System.out.println("连续签到：" + getLongDay(signList) + "天");
      }
      List<DateTime> dateTimes = DateUtil.rangeToList(DateUtil.beginOfMonth(new Date()),
          DateUtil.endOfMonth(new Date()), DateField.DAY_OF_MONTH);
      System.out.println("连续签到的日期");
      for (int signIndex : signList) {
        DateTime dateTime = dateTimes.get(signIndex);
        System.out.println(dateTime.toDateStr());
      }
      System.out.println(1);
    }
    return Result.accept(null);
  }
/**
   * 获取连续签到的天数
   *
   * @param list 传入日期的集合 此处以 List<Integer>举例
   * @return
   */
  private static int getLongDay(List<Integer> list) {
    List<Integer> list2 = new ArrayList<Integer>();// 存放中断元素的位置
    List<Integer> list3 = new ArrayList<Integer>();
    //先获取连续中断的位置，放在list2中
    for (int i = 0; i < list.size(); i++) {
      if (i == list.size() - 1) {
        break;
      } else if (list.get(i + 1) - list.get(i) != 1) {
        list2.add(i);
      }
    }
    //通过判断获取连续的个数，在list3中取最大值即可。
    if (0 == list2.size()) {
      // 没有中断 返回原集合长度
      return list.size();
    } else {
      for (int i = 0; i < list2.size(); i++) {
        if (1 == list2.size()) {
          list3.add(list2.get(0) + 1);// 中断前的天数
          list3.add(list.size() - 1 - list2.get(i));// 剩余的天数
        } else {
          if (i == 0) {
            list3.add(list2.get(0) + 1);
            list3.add(list2.get(i + 1) - list2.get(i));
          } else if (i == list2.size() - 1) {
            list3.add(list.size() - 1 - list2.get(i));
          } else {
            list3.add(list2.get(i + 1) - list2.get(i));
          }
        }
      }
      return Collections.max(list3);
    }
  }

避坑
bitset占用的内存是用最大的offset来决定的，根本不会管你实际要存多少有效数据，计算公式为

● 占用内存bit = 最大offset              
● 占用内存B = 最大offset / 8     
● 占用内存KB = 最大offset / 8 / 1024
● 占用内存MB = 最大offset / 8 / 1024 / 1024

• 比如现在我们要存用户对一篇文章的点赞数， 以及判断用户对文章有没有点赞。

• 如果使用bitset， 那么确实一篇文章只要一个key就行了，也没有hash或set的结构复杂，比如文章id为1， 简单点key的名字就为article⭐1。

• 那么点赞确实很简单，如果用户id=100，看下占用内存忽略不计

• 但如果这个时候又来了一个用户点赞，用户id=2560000呢，使用方式没有变化，也完成了点赞的统计。但是占用内存呢？
  这是什么概念？记录一篇小小的文章点赞数，即使只有两个人点赞， 就占用了312.5KB，来个几百篇文章或者动态之类的，再加上其他功能也这么使用用，这要浪费吃掉多少内存。因此用bitset的时候除了考虑系统人数规模（主要是决定bitset offset值的条件），还要考虑实际会有多少人用到这个功能。即使人数达到了很大的量级，但某个功能是个很偏僻的功能，还是要少用。

总结如下

1. 使用之前永远要作为第一要务考虑的就是offset的基数大小问题
2. 如果uv很小，而且offset基数又很大，不要使用，offset基数很小，可以用
3. 如果uv非常高，offset基数即使大一点，也可以使用，但这个要综合考虑去计算offset基数到底多大，uv又有没有高到离谱的程度。否则仍然还是set或者hash之类的更适用。

HyperLogLog

场景

在移动互联网的业务场景中，数据量很大，我们需要保存这样的信息：一个 key 关联了一个数据集合，同时对这个数据集合做统计。

• 统计一个 APP 的日活、月活数；

• 统计一个页面的每天被多少个不同账户访问量（Unique Visitor，UV））；

• 统计用户每天搜索不同词条的个数；

• 统计注册 IP 数。

通常情况下，我们面临的用户数量以及访问量都是巨大的，比如百万、千万级别的用户数量，或者千万级别、甚至亿级别的访问信息。
你可以通过 set 集合、bitmap 这类常用工具，但有个最大的缺点是，如果数据量巨大，比如 1 亿，甚至 10 亿将耗费巨大内存消耗。

原理

Redis HyperLogLog基于一种称为HyperLogLog算法的概率性算法来估计基数。 HyperLogLog使用一个长度为m的位数组和一些hash函数来估计集合中的唯一元素数。

在 HyperLogLog 算法中，对每个元素进行哈希处理，把哈希值转换为二进制后，根据二进制串前缀中 1 的个数来给每个元素打分。例如，一个元素的哈希值为01110100011，那么前缀中1的个数是3，因此在 HyperLogLog 算法中，这个元素的分数为3。

当所有元素的分数统计完之后，取每一个分数的倒数（1 / 2^n），然后将这些倒数相加后取倒数，就得到一个基数估计值，这个值就是HyperLogLog算法的估计结果。
HyperLogLog算法通过对位数组的长度m的大小进行取舍，折衷数据结构占用的内存与估计值的精准度（即估计误差），得到了在数据占用空间与错误较小程度之间完美的平衡。
简而言之，HyperLogLog算法的核心思想是基于哈希函数和位运算，通过将哈希值转换成比特流并统计前导0的个数，从而快速估算大型数据集中唯一值的数量。通过 hyperloglog 算法我们可以在非常大的数据集中进行极速的网页浏览器去重。

算法简介

HyperLogLog 算法的基本思想来自伯努利过程。
伯努利过程就是一个抛硬币实验的过程。抛一枚正常硬币，落地可能是正面，也可能是反面，二者的概率都是 1/2 。伯努利过程就是一直抛硬币，直到落地时出现正面位置，并记录下抛掷次数k。比如说，抛一次硬币就出现正面了，此时 k 为 1; 第一次抛硬币是反面，则继续抛，直到第三次才出现正面，此时 k 为 3。
那么如何通过伯努利过程来估算抛了多少次硬币呢？还是假设 1 代表抛出正面，0 代表反面。连续出现两次 0 的序列应该为“001”，那么它出现的概率应该是三个二分之一相乘，即八分之一。那么可以估计大概抛了 8 次硬币。
HyperLogLog 原理思路是通过给定 n 个的元素集合，记录集合中数字的比特串第一个1出现位置的最大值k，也可以理解为统计二进制低位连续为零（前导零）的最大个数。通过k值可以估算集合中不重复元素的数量m，m近似等于 2^k。

延迟队列

主要处理非立即生效的业务场景，比如

• 订单下单30分钟内，超时未支付；

• 优惠券、活动等需要在指定时间内生效的，

• 会议开始前10分钟消息通知

源码实现

首先创建DelayedQueueManager 用于延迟队列的创建、获取、销毁

@Slf4j
@RequiredArgsConstructor
public class DelayedQueueManager {

   private final RedissonClient redissonClient;
   private final ThreadPoolTaskExecutor threadPoolExecutor;

   private final ConcurrentHashMap<String, DelayedQueue<?>> DELAYED_QUEUE_MAP = new ConcurrentHashMap<>();

    /**
    * 创建延时对象实例
    *
    * @param queueId       队列标识
    * @param <T>           泛型类型
    * @return              延时队列实例
    */
    public <T> DelayedQueue<T> create(String queueId) {
        // 先尝试从容器中获取延迟队列
        DelayedQueue<T> delayedQueue = (DelayedQueue<T>) DELAYED_QUEUE_MAP.get(queueId);
        // 如果未获取到，则实例化一个延迟队列对象
        if (Objects.isNull(delayedQueue)) {
            delayedQueue = new DelayedQueue<>(redissonClient, queueId, threadPoolExecutor);
            DELAYED_QUEUE_MAP.putIfAbsent(queueId, delayedQueue);
        }
        return delayedQueue;
    }

        /**
        * 获取延迟队列实例
        *
        * @param queueId               队列标识id
        * @param createIfNotExists     对象不存在时是否直接创建
        * @param <T>                   元素泛型类型
        * @return                      延时队列实例
        */
        public <T> DelayedQueue<T> get(String queueId, boolean createIfNotExists) {
            // 先尝试从容器中获取延迟队列
            final DelayedQueue<T> delayedQueue = (DelayedQueue<T>) DELAYED_QUEUE_MAP.get(queueId);
            if (Objects.isNull(delayedQueue)) {
                if (createIfNotExists) {
                    return create(queueId);
                }
                throw new BizzException("找不到延时队列【{}】的实例对象！", queueId);
            }
            else {
                return delayedQueue;
            }
        }

        /**
        * 销毁指定延时队列实例
        *
        * @param queueId       延时队列id
        */
        public void destroy(String queueId) {
            // 获取延时队列实例并调用销毁方法
            Optional.ofNullable(DELAYED_QUEUE_MAP.get(queueId)).ifPresent(DelayedQueue::destroy);
        }
    }

增加延迟队列，从redisson的BlockingQueue获取DelayedQueue，并通过守护线程，获取延迟队列中的数据，执行响应的监听器

public class DelayedQueue<T> {
    /**
     * 队列id标识
     */
    private final String queueId;
    /**
     * 是否取消的标识
     */
    private final AtomicBoolean isCancelled = new AtomicBoolean(false);
    /**
     * 守护线程
     */
    private final Future<Void> daemonThread;
    // 阻塞队列
    private final RBlockingQueue<T> blockingQueue;
    // 延迟队列
    private final RDelayedQueue<T> delayedQueue;
    // 监听器列表
    private final ConcurrentHashSet<DelayedQueueListener<T>> delayedQueueListeners = new ConcurrentHashSet<>();

    /**
     * 构造器函数
     *
     * @param redissonClient            redisson客户端实例
     * @param queueId                   队列标识
     * @param threadPoolTaskExecutor    监听器通知线程池
     */
    public DelayedQueue(RedissonClient redissonClient, String queueId, ThreadPoolTaskExecutor threadPoolTaskExecutor) {
        this.blockingQueue = redissonClient.getBlockingQueue(queueId);
        this.delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
        this.queueId = queueId;
        // 添加守护线程监听阻塞队列变化，通知到每一个监听器
        this.daemonThread = CompletableFuture.runAsync(
                () -> {
                    while (!isCancelled.get()) {
                        try {
                            // 获取延时队列中的元素（会阻塞）
                            final T value = blockingQueue.take();

                            // 向每个注册监听器发送通知
                            delayedQueueListeners.forEach(
                                    // 异步通知防止阻塞
                                    listener -> CompletableFuture.runAsync(() ->  listener.invoke(value), threadPoolTaskExecutor)
                            );
                        } catch (InterruptedException e) {
                            log.error("获取阻塞队列元素异常！队列名称:{}", this.queueId, e);
                            ThreadUtil.safeSleep(GlobalConstants.Number.THOUSAND);
                        }
                    }
                }
        );
    }

    /**
     * 添加监听器
     *
     * @param delayedQueueListener      延迟队列事件监听器
     */
    public boolean addListener(DelayedQueueListener<T> delayedQueueListener) {
        // 注册监听器
        return delayedQueueListeners.add(delayedQueueListener);
    }

    /**
     * 移除监听器
     *
     * @param delayedQueueListener      延迟队列事件监听器
     */
    public boolean removeListener(DelayedQueueListener<T> delayedQueueListener) {
        return delayedQueueListeners.remove(delayedQueueListener);
    }

    /**
     * 添加元素到延时队列
     *
     * @param element       队列元素
     * @param delay         延迟时间
     * @param timeUnit      时间单位
     */
    public void offer(T element, long delay, TimeUnit timeUnit) {
        this.delayedQueue.offer(element, delay, timeUnit);
    }

    /**
     * 销毁当前队列
     */
    public void destroy() {
        // 1. 取消守护线程任务
        if (isCancelled.compareAndSet(false, true)) {
            // 2. 取消当前正在运行的任务
            daemonThread.cancel(true);
            // 3. 清空阻塞队列和延迟队列中的元素
            this.delayedQueue.destroy();
            this.blockingQueue.clear();
        }
    }
}

延迟队列触发监听器

/**
 * 延迟队列触发监听器
 *
 */
@FunctionalInterface
public interface DelayedQueueListener<T> {
    /**
     * 触发延迟队列事件
     */
    void invoke(T t);
}

延迟队列Springboot配置，配置延迟队列线程池

@Configuration
@AutoConfigureAfter(RedissonAutoConfiguration.class)
public class DelayedQueueBeanConfiguration {
    /**
     * 延时队列异步线程池
     */
    @Bean
    @ConditionalOnMissingBean(name = "delayedQueueListenerAsyncPool")
    public ThreadPoolTaskExecutor delayedQueueListenerAsyncPool() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        //核心线程数=容器cpu核心的2倍+1
        int coreNum = TWO * Runtime.getRuntime().availableProcessors() + ONE;
        threadPoolTaskExecutor.setCorePoolSize(coreNum);
        //最大线程数是核心线程数的两倍
        threadPoolTaskExecutor.setMaxPoolSize(TWO * coreNum);
        //队列长度100
        threadPoolTaskExecutor.setQueueCapacity(HUNDRED);
        //拒绝策略是直接抛异常
        threadPoolTaskExecutor.setRejectedExecutionHandler(
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
        threadPoolTaskExecutor.setBeanName("delayedQueueListenerAsyncPool");
        threadPoolTaskExecutor.setThreadNamePrefix("delay-queue-notify-");
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }

    /**
     * 延时队列管理器注入
     *
     * @param redissonClient            redisson客户端实例
     * @param threadPoolTaskExecutor    线程池实例
     */
    @Bean
    @ConditionalOnBean(RedissonClient.class)
    public DelayedQueueManager delayedQueueManager(RedissonClient redissonClient, 
                  @Qualifier("delayedQueueListenerAsyncPool")  ThreadPoolTaskExecutor threadPoolTaskExecutor) {
        return new DelayedQueueManager(redissonClient, threadPoolTaskExecutor);
    }
}

方法调用，此处案例为：优化券创建后，根据配置生效的时间进行生效

String delayedQueueId = RedisKeyUtil.getCouponEffectDelayedQueueKey();
DelayedQueue<CouponVo> couponEffectDelayedQueue
    = delayedQueueManager.get(delayedQueueId, true);
couponEffectDelayedQueue.addListener(couponEffectDelayedQueueListener);
// couponVo 优惠券， between 优惠券开始的时间（延迟的时间）
couponEffectDelayedQueue.offer( couponVo , between, TimeUnit.SECONDS);

监听延迟队列生效，并执行生效方法

@Slf4j
@Component
@RequiredArgsConstructor
public class CouponEffectDelayedQueueListener implements DelayedQueueListener<CouponVo> {

    private final CouponHandler couponHandler;

    @Override
    public void invoke(CouponVo couponVo) {
        String effectTime = DateUtil.format(couponVo.getSendStartTime(), DatePattern.NORM_DATETIME_FORMAT);
        log.info("监听到优惠券编号{}生效延迟队列, 优惠券生效时间：{}",  couponVo.getCouponNo(), effectTime);
        try {
            couponHandler.changeCouponStatus(CouponDto.builder()
                .ids(Lists.newArrayList(couponVo.getId()))
                .status(SendCouponStatusEnum.IN_USING)
                .modifierId(-1L)
                .modifyTime(new Date()).build()
            );
            log.info("优惠券：{} 状态生效成功", couponVo.getCouponNo());
        } catch (Exception e) {
            log.warn("优惠券：{} 状态生效失败", couponVo.getCouponNo(), e);
        }
    }
}

Reids命令对应Redissson方法