异步结果通知实现——基于Redis实现，我这操作很可以

notify-keyspace-events Ex

@Component
public class OrderExpireEventListener extends MessageListenerAdapter {
    @Override
    public void onMessage(Message message, byte[] pattern) {
          byte[] body = message.getBody();
          String msg = redisWrapper.getRedisTemplate().getStringSerializer().deserialize(body);
          // do something...

          // 假如通知失败，需要重新计算下次通知时间，设置 Redis
          // 至于数据类型，String 即可
    }
}

@Bean
public RedisMessageListenerContainer container(RedisConnectionFactory factory, 
OrderExpireEventListener adapter) {
    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    container.setConnectionFactory(factory);
    container.addMessageListener(adapter, new PatternTopic("__keyevent@0__:expired"));
    return container;
}

app1:trans:notice:1615283234

@Slf4j
@Component
public class ConsumerTask {
    @Autowired
    RedisTemplate<String, Object> redisTemplate;
    // Sorted Set 队列键 
    private static String KEY = "TEST:ZSET";

    @Scheduled(cron = "0/1 * * * * ?")
    public void run() {
        try {
            this.doRun();
        } catch (Exception e) {
            log.error("消费异常", e);
        }
    }

    private void doRun() {
        // zrange 分数从小到大 zrevrange 分数从大到小
        // 拿出最新的待处理消息
        Set<ZSetOperations.TypedTuple<Object>> tuples = 
        redisTemplate.opsForZSet().rangeWithScores(KEY, 0, 0);
        if (CollectionUtils.isEmpty(tuples)) {
            log.info("队列无数据");
            return ;
        }

        ZSetOperations.TypedTuple<Object> typedTuple = tuples.iterator().next();
        if (typedTuple == null) {
            log.info("队列无数据");
            return ;
        }
        Double score = typedTuple.getScore();
        Object value = typedTuple.getValue();

        if (System.currentTimeMillis() < score) {
            log.info("未到执行时间...");
            return ;
        }

        Long zrem = redisTemplate.opsForZSet().remove(KEY, value);
        if (Long.compare(1L, zrem) == 0) {
            log.info("删除数据成功，开始处理，数据：{}", value.toString());

            // do someting...

            // 假如通知失败，需要重新计算通知时间（score 值）并在 Redis 设置（ZADD）该消息
        }
        else {
            log.info("被其他的消费端抢占，不处理...");
        }
    }
}

org.springframework.data.redis.core.ZSetOperations#add(K, V, double)
org.springframework.data.redis.core.ZSetOperations#rangeByScore(K, double, double)

System.out.println(Long.MAX_VALUE); // 2 的 64 次方-1，19 个数位
System.out.println(Long.MIN_VALUE); // 负的 2 的 64 次方 
System.out.println(Double.MAX_VALUE); // 2 的 1024 次方 -1,308 个数位
System.out.println(Double.MIN_VALUE); // 2 的 -1074次方 

// 1、从 Sorted Set 中拿出 score 值在 前五秒 到 目前（包含现在）的所有元素
Date now = new Date();
Date fiveSecondBefore = DateUtils.addSeconds(now, -5);
Set<Object> objects = redisTemplate.opsForZSet()
.rangeByScore("Sorted Set:Key", fiveSecondBefore.getTime(), now.getTime());
if (CollectionUtils.isEmpty(objects)) {
    return ;
}
// 2、将这些元素从 Sorted Set 中删除
Long removeResult = redisTemplate.opsForZSet().remove("Sorted Set:Key", objects);
if (Long.compare(removeResult, objects.size()) != 0) {
    return ;
}
// 3、将这些元素放进 List
Long result = redisTemplate.opsForList().leftPushAll("List:Key", objects);

local list_key = KEYS[1];
local sorted_set_key = KEYS[2];
local now = ARGV[1];
local sorted_set_size = redis.call('ZCARD', sorted_set_key)

if (tonumber(sorted_set_size) <= 0) then
    return
end

local members = redis.call('ZRANGEBYSCORE', sorted_set_key, 0, tonumber(now));
if (next(members) == nil) then
    return
end

for key,value in ipairs(members)
do
    local zscore = redis.call('ZSCORE',sorted_set_key,value);

    if (tonumber(now) < tonumber(zscore)) then
        return zscore;
    end

    redis.call('ZREM', sorted_set_key, value);
    redis.call('RPUSH', list_key, value);
end

local topmember = redis.call('ZRANGE', sorted_set_key, 0, 0);
local nextvalue = next(topmember);
if (nextvalue == nil) then
    return
end

for k,v in ipairs(topmember)
do
    return redis.call('ZSCORE', sorted_set_key, v);
end

@Scheduled(cron = "0/1 * * * * ?")
public void run4() {
    ClassPathResource resource = new ClassPathResource("sorted_set_to_list.lua");
    String luaScript = FileUtils.readFileToString(resource.getFile());
    DefaultRedisScript<String> redisScript = new DefaultRedisScript<>(luaScript, String.class);
    //
    List<String> keys = Lists.newArrayList("TEST:LIST", "TEST:ZSET");
    String now = String.valueOf(System.currentTimeMillis());
    // 注意这里的序列化器，需要换成 StringSerializer
    // 替换的默认的 Jackson2JsonRedisSerializer
    String executeResult = redisTemplate.execute(redisScript, redisTemplate.getStringSerializer(),
            redisTemplate.getStringSerializer(), keys, now);
    log.info("lua 脚本执行结果：{}", executeResult);
}

@Component
@Slf4j
public class ListConsumer implements ApplicationListener<ContextRefreshedEvent> {
    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        Executors.newSingleThreadExecutor().submit(new PopEventRunner());
    }
    private static class PopEventRunner implements Runnable {
        @Override
        public void run() {
            RedisTemplate<String, Object> redisTemplate  = (RedisTemplate<String, Object>) SpringUtil.getBean3("redisTemplate");
            while (true) {
                try {
                    Object leftPop = redisTemplate.opsForList().leftPop("TEST:LIST", Integer.MAX_VALUE, TimeUnit.SECONDS);
                    if (leftPop == null) {
                        continue ;
                    }
                    // do something...

                    // 当通知失败时，重新计算通知时间并设置（ZADD）Redis
                } catch (Exception e) {
                    log.error("监听异常", e);
                    sleep(5); // 发生异常睡五秒
                }
            }
        }
    }
}