Redisson延迟队列遇到的坑

做个Redisson延迟任务踩了好多坑。1、项目启动时报Unexpected exception while processing command2、经常性Redis server response timeout (3000 ms)3、项目启动那一刻消费任务队列中积压的任务，项目运行过程中不消费4、使用rbucket做幂等，set延时导致旧值覆盖新值，幂等失败。

真名骏

6676人浏览 · 2022-09-26 22:06:26

真名骏 · 2022-09-26 22:06:26 发布

一、问题描述

做个Redisson延迟任务踩了好多坑。

1、项目启动时报Unexpected exception while processing command

2、经常性Redis server response timeout (3000 ms)

3、项目启动那一刻消费任务队列中积压的任务，项目运行过程中不消费

4、使用rbucket做幂等，set延时导致旧值覆盖新值，幂等失败

二、解决方法（可能只适用我个人）

1、项目启动时报Unexpected exception while processing command

（1）一开始是延迟任务不起效，按照网上的解决方案加了：

redissonClient.getDelayedQueue(blockingFairQueue)

public <T> void runByType(RedisDelayedQueueTaskListener taskEventListener, String queueName) {
        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
        //redissonClient.getDelayedQueue(blockingFairQueue);//项目启动时报错
        ((Runnable) () -> {
            while (true) {
                try {
                    log.error("我来了！");
                    T t = blockingFairQueue.take();//如果没有元素，则阻塞队列
                    if (t != null) {
                        log.error("我进来了！");
                        taskEventListener.invoke(t);
                    }
                    log.error("我还在！");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).run();
    }

然而任务还是不起效。并且项目启动时报如标题的错。

根据源码，这段代码的在这里的作用是创建底层的时间轮定时任务：如果有旧的定时任务，那么使用旧的定时任务。duBug结果是每次都能拿到还在跑的旧定时任务。所以把这个语句去掉了。

（2）去掉上面这段代码后，Unexpected exception报错还在。

改为用Spring容器初始化队列后解决。

    /**
     * 添加机器人
     */
    @Bean(name = "delayedQueueRobotCreate")
    public RDelayedQueue<XXX> delayedQueueRobotCreate(@Qualifier("blockingQueueRobotCreate") RBlockingQueue blockingQueue) {
        RDelayedQueue<XXX> carIdRDelayedQueue = redissonClient.getDelayedQueue(blockingQueue);
        return carIdRDelayedQueue;
    }

    /**
     * 添加机器人
     */
    @Bean(name = "blockingQueueRobotCreate")
    public RBlockingQueue<XXX> blockingQueueRobotCreate() {
        RBlockingQueue<XXX> carIdRBlockingQueue = redissonClient.getBlockingQueue(TaskQueueType.ROBOT_CREATE.getName());
        return carIdRBlockingQueue;
    }

2、经常性Redis server response timeout (3000 ms)

（1）添加定时任务的时候需要用redisson做幂等，在set的时候三次有一次timeOut

使用下面的方法大致解决：

A.首先配置

/**resdisson配置心跳，设置空闲超时时间（原设为60，后发现单位为毫秒）*/
    @Bean(destroyMethod = "shutdown")
    public RedissonClient redisson() throws IOException {
        Config config = new Config();
        config.setExecutor(Executors.cacheExecutor)
                .useSingleServer()
                .setAddress("redis://localhost:6379")
                .setPingConnectionInterval(1000)//1秒发一次心跳，默认是30秒
                .setConnectionPoolSize(8)
                .setConnectionMinimumIdleSize(8)
                .setIdleConnectionTimeout(60000)//60秒，默认是10秒
                .setClientName("xxxx")
                .setKeepAlive(true)
                .setTcpNoDelay(true);
        return Redisson.create(config);
    }

B.然后异步

CompletableFuture fs = CompletableFuture.runAsync(() -> {
            try {
                log.warn("用户[{}]XXXX[{}]，触发“XXXX”定时任务", userId, XXXId);
                cacheUtils.setAsync(now, RedisCacheKey.FORMATION_SUCCESS, carId);//幂等处理：定时任务中跟XXXX.update_at比较
                redisDelayedQueueUtils.addQueueFormationSuccessionFuture(xxxx, TaskQueueType.FORMATION_SUCCESS.getDuration(), TaskQueueType.FORMATION_SUCCESS.getTimeUnit());//5分钟
            } catch (Exception e) {
                e.printStackTrace();
                log.error("用户[{}]XXXX[{}]，触发“XXXX”定时任务，触发失败！ERR", userId, XXXId);
            }
        }, Executors.formationSucccessExecutor);

3、项目启动那一刻消费任务队列中积压的任务，项目运行过程中不消费

最后一个坑在队列的take()方法：如果没有元素，则阻塞队列直到队列中有任务后返回

经过各种调试，在项目刚启动时take会获取并消费任务，阻塞和返回过程也正常，但是消费几次就不消费了，我不理解。

当两个定时任务使用同一个线程池时则项目启动时也不消费，我不理解。

最后改为poll方法：如果没有元素，则阻塞一定时长后返回。

public void runRobotCreate(RedisDelayedQueueTaskListener taskEventListener) {
        ((Runnable) () -> {
            while (true) {
                try {
                    //10秒后返回
                    XXX xxx = blockingQueueRobotCreate.poll(10, TimeUnit.SECONDS);
                    if (xxx != null) {
                        taskEventListener.invoke(xxx);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    log.error("我要被中断了！RobotCreate");
                }
            }
        }).run();
    }