/**
 * @program: 
 * @description: redis延時(shí)隊(duì)列生產(chǎn)者類注解，標(biāo)記生產(chǎn)者類，用來掃描生產(chǎn)者類中的生產(chǎn)者方法，將生產(chǎn)者方法注冊到redis延時(shí)隊(duì)列中
 * @author: jiangchengxuan
 * @created: 2023/12/09 10:32
 */
@Component
@Documented
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface RedisMessageQueue {}

/**
 * @program: 
 * @description: 
 * 帶有此注解的方法，方法的入?yún)⑹紫葧晦D(zhuǎn)換為json字符串，然后存入redis的zset中，score為當(dāng)前時(shí)間+延時(shí)時(shí)間，value為json字符串
 * 當(dāng)延時(shí)時(shí)間到達(dá)后，會從redis的zset中取出value，然后將value轉(zhuǎn)換為入?yún)㈩愋?，調(diào)用此方法，執(zhí)行業(yè)務(wù)邏輯
 * 此注解只能標(biāo)記在方法上，且方法必須為public，且只能有一個(gè)參數(shù)
 * 此注解標(biāo)記的方法，必須在redis延時(shí)隊(duì)列生產(chǎn)者類中，否則不會生效
 * @author: jiangchengxuan
 * @created:  2023/12/09 10:37
 */
@Documented
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RedisMessageQueueMethod {
    String threadName() default "redis消息隊(duì)列默認(rèn)線程";
    String queueKey();   // 隊(duì)列key值
    int threadNum() default 1;      //默認(rèn)線程數(shù)量
    int threadSleepTime() default 500;  //默認(rèn)線程休眠時(shí)間默認(rèn)500ms
}

/**
 * @program: 
 * @description:  生產(chǎn)者類具體實(shí)現(xiàn)
 * @author: jiangchengxuan
 * @created: 2023/12/09 10:44
 */
@Slf4j
@Component
public class DelayQueueWorkerConfig implements InitializingBean {
    private volatile boolean monitorStarted = false;

    private volatile boolean monitorShutDowned = false;

    private ExecutorService executorService;

    // 需要監(jiān)控的延時(shí)隊(duì)列
    @Autowired
    protected IDelayQueue<String> monitorQueue;

    @Autowired
    private ApplicationContext applicationContext;


    @Override
    public void afterPropertiesSet(){
        //spring工具類，可以獲取指定注解的類
        Map<String, Object> allNeedClass = applicationContext.getBeansWithAnnotation(RedisMessageQueue.class);
        for (Map.Entry<String, Object> entry : allNeedClass.entrySet()) {
            Object bean = entry.getValue();
            Method[] methods = bean.getClass().getMethods();
            for (Method method : methods) {
                Annotation[] annotations = method.getDeclaredAnnotations();
                for (Annotation annotation : annotations) {
                    if (annotation instanceof RedisMessageQueueMethod) {
                        RedisMessageQueueMethod queueMethod = (RedisMessageQueueMethod) annotation;
                        //找的需要使用消息隊(duì)列的方法后，
                        initExecuteQueue(queueMethod, method, bean);
                    }
                    }
                }
            }
        }


    /**
     * 初始化執(zhí)行造作
     * @param queueAnnotations 注解
     * @param method 方法
     * @param bean 對象
     */
    void initExecuteQueue(RedisMessageQueueMethod queueAnnotations ,Method method,Object bean) {
        String threadName = queueAnnotations.threadName();
        int threadNum = queueAnnotations.threadNum();
        int threadSheepTime = queueAnnotations.threadSleepTime();
        String queueKey = queueAnnotations.queueKey();
        //獲取所有消息隊(duì)列名稱
        executorService = Executors.newFixedThreadPool(threadNum);
        for (int i = 0; i < threadNum; i++) {
            final int num = i;
            executorService.execute(() -> {
                Thread.currentThread().setName(threadName + "[" + num + "]");
                //如果沒有設(shè)置隊(duì)列queuekey或者已經(jīng)暫停則不執(zhí)行
                while (!monitorShutDowned) {
                    String value = null;
                    try {
                        value = monitorQueue.get(queueKey);
                        // 獲取數(shù)據(jù)時(shí)進(jìn)行刪除操作，刪除成功，則進(jìn)行處理，業(yè)務(wù)邏輯處理失敗則繼續(xù)添加回隊(duì)列但是時(shí)間設(shè)置最大以達(dá)到保存現(xiàn)場的目的，防止并發(fā)獲取重復(fù)數(shù)據(jù)
                        if (StringUtils.isNotEmpty(value)) {
                            if (log.isDebugEnabled()) {
                                log.debug("Monitor Thread[" + Thread.currentThread().getName() + "], get from queue,value = {}", value);
                            }
                            boolean success = (Boolean) method.invoke(bean, value);
                            // 失敗重試
                            if (!success) {
                                success =  (Boolean) method.invoke(bean, value);;
                                if (!success) {
                                    log.warn("Monitor Thread[" + Thread.currentThread().getName() + "] execute Failed,value = {}", value);
                                    monitorQueue.add(TimeUnit.DAYS,365, value, queueKey);
                                }
                            } else {
                                if (log.isDebugEnabled()) {
                                    log.debug("Monitor Thread[" + Thread.currentThread().getName() + "]：execute successfully!values = {}", value);
                                }
                            }
                        } else {
                            if (log.isDebugEnabled()) {
                                log.debug("Monitor Thread[" + Thread.currentThread().getName() + "]：monitorThreadRunning = {}", monitorStarted);
                            }
                            Thread.sleep(threadSheepTime);
                        }
                    } catch (Exception e) {
                        log.error("Monitor Thread[" + Thread.currentThread().getName() + "] execute Failed,value = " + value, e);
                    }
                }
                log.info("Monitor Thread[" + Thread.currentThread().getName() + "] Completed...");
            });
        }
        log.info("thread pool is started...");
    }

}

/**
 * @program: 
 * @description: 
 * 延時(shí)隊(duì)列接口實(shí)現(xiàn)類，
 * 使用redis的zset實(shí)現(xiàn)延時(shí)隊(duì)列，
 * @author: jiangchengxuan
 * @created:  2023/12/09 23:34
 */
public interface IDelayQueue <E> {
    /**
     * 向延時(shí)隊(duì)列中添加數(shù)據(jù)
     *
     * @param score 分?jǐn)?shù)
     * @param data  數(shù)據(jù)
     * @return true 成功 false 失敗
     */
    boolean add(long score, E data,String queueKey);


    /**
     * 向延時(shí)隊(duì)列中添加數(shù)據(jù)
     *
     * @param timeUnit 時(shí)間單位
     * @param time     延后時(shí)間
     * @param data     數(shù)據(jù)
     * @param queueKey
     * @return true 成功 false 失敗
     */
    boolean add(TimeUnit timeUnit, long time, E data, String queueKey);

    /**
     * 從延時(shí)隊(duì)列中獲取數(shù)據(jù)
     * @param queueKey 隊(duì)列key
     * @return 數(shù)據(jù)
     */
    String get(String queueKey);

    /**
     * 刪除數(shù)據(jù)
     *
     * @param key
     * @param data 數(shù)據(jù)
     * @return
     */
    public<T> boolean rem(String key, T data) ;
}

/**
 * @program: 
 * @description:  redis操作類，封裝了redis的操作方法，使用時(shí)直接注入即可使用，不需要關(guān)心redis的操作細(xì)節(jié)，使用時(shí)只需要關(guān)心業(yè)務(wù)邏輯即可
 * @author: jiangchengxuan
 * @created: 2023/12/09 23:35
 */
@Service
public class RedisDelayQueue implements IDelayQueue<String> {

    @Autowired
    private RedisService redisService;


    @Override
    public boolean add(long score, String data,String queueKey) {
        return redisService.opsForZSet(Constant.DEFAULT_REDIS_QUEUE_KEY_PREFIX+queueKey, data, score);
    }

    @Override
    public boolean add(TimeUnit timeUnit, long time, String data, String queueKey) {
        switch (timeUnit) {
            case SECONDS:
                return add(LocalDateTime.now().plusSeconds(time).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), data, queueKey);
            case MINUTES:
                return add(LocalDateTime.now().plusMinutes(time).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), data,queueKey);
            case HOURS:
                return add(LocalDateTime.now().plusHours(time).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), data,queueKey);
            case DAYS:
                return add(LocalDateTime.now().plusDays(time).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(), data,queueKey);
            default:
                return false;
        }
    }


    @Override
    public String get(String queueKey) {
        long now = System.currentTimeMillis();
        long min = Long.MIN_VALUE;
        Set<String> res = redisService.rangeByScoreZSet(Constant.DEFAULT_REDIS_QUEUE_KEY_PREFIX+queueKey, min, now, 0, 10);
        if (!CollectionUtils.isEmpty(res)) {
            for (String data : res){
                // 刪除成功，則進(jìn)行處理，防止并發(fā)獲取重復(fù)數(shù)據(jù)
                if (rem(queueKey, data)){
                    return data;
                }
            }
        }
        return null;
    }


    @Override
    public<T> boolean rem(String key, T data) {
        return redisService.remZSet(Constant.DEFAULT_REDIS_QUEUE_KEY_PREFIX+key, data);
    }
}

@RedisMessageQueue
public class SomethingClass
{
    @Autowired
    private IDelayQueue<String> messageQueue;

    /**
     * 生產(chǎn)者，向隊(duì)列中添加數(shù)據(jù),30秒后消費(fèi)者進(jìn)行消費(fèi)
     */
    public void test(){
        messageQueue.add(TimeUnit.SECONDS,30L,"這是參數(shù)數(shù)據(jù)","new_queue");
    }
    
    /**
     * 消費(fèi)者,如果按此配置的話，會啟動一個(gè)線程，線程名稱為：測試線程名稱，線程數(shù)量為1，線程休眠時(shí)間為10毫秒
     * 注意：queueKey需要與生產(chǎn)者中的queueKey保持一致才能進(jìn)行消費(fèi)
     * @param data 
     */
    @Override
    @RedisMessageQueueMethod(threadName = "測試線程名稱",queueKey = "new_queue",threadNum = 1,threadSleepTime = 10)
    public void testMethod(String data) {
        //do something
    }

}