import lombok.Data;
import lombok.experimental.Accessors;

import javax.validation.constraints.NotNull;
import java.io.Serializable;

/**
 * Redis 消息隊(duì)列中的消息體
 * @author shikanatsu
 */
@Data
@Accessors(chain = true)
public class RedisMessage implements Serializable {

    /** 消息隊(duì)列組 **/
    private String group;

    /**
     * 消息id
     */
    private String id;

    /**
     * 消息延遲/ 秒
     */
    @NotNull(message = "消息延時(shí)時(shí)間不能為空")
    private long delay;

    /**
     * 消息存活時(shí)間 單位：秒
     */
    @NotNull(message = "消息存活時(shí)間不能為空")
    private int ttl;
    /**
     * 消息體，對(duì)應(yīng)業(yè)務(wù)內(nèi)容
     */
    private Object body;
    /**
     * 創(chuàng)建時(shí)間，如果只有優(yōu)先級(jí)沒(méi)有延遲，可以設(shè)置創(chuàng)建時(shí)間為0
     * 用來(lái)消除時(shí)間的影響
     */
    private long createTime;
}

package com.shixun.base.redisMq;

import com.shixun.base.jedis.service.RedisService;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * Redis消息隊(duì)列
 *
 * @author shikanatsu
 */
@Component
public class RedisMq {

    /**
     * 消息池前綴，以此前綴加上傳遞的消息id作為key，以消息{@link MSG_POOL}
     * 的消息體body作為值存儲(chǔ)
     */
    public static final String MSG_POOL = "Message:Pool:";

    /**
     * zset隊(duì)列 名稱(chēng) queue
     */
    public static final String QUEUE_NAME = "Message:Queue:";

//    private static final int SEMIH = 30 * 60;


    @Resource
    private RedisService redisService;

    /**
     * 存入消息池
     *
     * @param message
     * @return
     */
    public boolean addMsgPool(RedisMessage message) {
        if (null != message) {
            redisService.set(MSG_POOL + message.getGroup() + message.getId(), message, message.getTtl());
            return true;
        }
        return false;
    }

    /**
     * 從消息池中刪除消息
     *
     * @param id
     * @return
     */
    public void deMsgPool(String group, String id) {
        redisService.remove(MSG_POOL + group + id);
    }

    /**
     * 向隊(duì)列中添加消息
     *
     * @param key
     * @param score 優(yōu)先級(jí)
     * @param val
     * @return 返回消息id
     */
    public void enMessage(String key, long score, String val) {
        redisService.zsset(key, val, score);
    }

    /**
     * 從隊(duì)列刪除消息
     *
     * @param id
     * @return
     */
    public boolean deMessage(String key, String id) {
        return redisService.zdel(key, id);
    }
}

import cn.hutool.core.convert.Convert;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.IdUtil;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.validation.annotation.Validated;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * 消息生產(chǎn)者
 *
 * @author shikanatsu
 */
@Component
public class MessageProvider {

    static Logger logger = LoggerFactory.getLogger(MessageProvider.class);

    @Resource
    private RedisMq redisMq;

    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");


    public boolean sendMessage(@Validated RedisMessage message) {
        Assert.notNull(message);
        //The priority is if there is no creation time
//        message.setCreateTime(System.currentTimeMillis());
        message.setId(IdUtil.fastUUID());
        Long delayTime = message.getCreateTime() + Convert.convertTime(message.getDelay(), TimeUnit.SECONDS, TimeUnit.MILLISECONDS);
        try {
            redisMq.addMsgPool(message);
            redisMq.enMessage(RedisMq.QUEUE_NAME+message.getGroup(), delayTime, message.getId());
            logger.info("RedisMq發(fā)送消費(fèi)信息{}，當(dāng)前時(shí)間:{},消費(fèi)時(shí)間預(yù)計(jì){}",message.toString(),new Date(),sdf.format(delayTime));
        }catch (Exception e){
            e.printStackTrace();
            logger.error("RedisMq 消息發(fā)送失敗，當(dāng)前時(shí)間:{}",new Date());
            return false;
        }
        return true;
    }
}

/**
 * Redis消息消費(fèi)者
 * @author shikanatsu
 */
@Component
public class RedisMqConsumer {

    private static final Logger log = LoggerFactory.getLogger(RedisMqConsumer.class);

    @Resource
    private RedisMq redisMq;

    @Resource
    private RedisService redisService;

    @Resource
    private MessageProvider provider;

    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");


    //@Scheduled(cron = "*/1 * * * * ? ")
    /**
     Instead of a thread loop, you can use Cron expressions to perform periodic tasks
     */
    public void baseMonitor(RedisMqExecute mqExecute){
        String queueName = RedisMq.QUEUE_NAME+mqExecute.getQueueName();
        //The query is currently expired
        Set<Object> set = redisService.rangeByScore(queueName, 0, System.currentTimeMillis());
        if (null != set) {
            long current = System.currentTimeMillis();
            for (Object id : set) {
                long  score = redisService.getScore(queueName, id.toString()).longValue();
                //Once again the guarantee has expired ， And then perform the consumption
                if (current >= score) {
                    String str = "";
                    RedisMessage message = null;
                    String msgPool = RedisMq.MSG_POOL+mqExecute.getQueueName();
                    try {
                        message = (RedisMessage)redisService.get(msgPool + id.toString());
                        log.debug("RedisMq:{},get RedisMessage success now Time:{}",str,sdf.format(System.currentTimeMillis()));
                        if(null==message){
                            return;
                        }
                        //Do something ； You can add a judgment here and if it fails you can add it to the queue again
                        mqExecute.execute(message);
                    } catch (Exception e) {
                        e.printStackTrace();
                        //If an exception occurs, it is put back into the queue
                        // todo：  If repeated, this can lead to repeated cycles
                        log.error("RedisMq: RedisMqMessage exception ,It message rollback , If repeated, this can lead to repeated cycles{}",new Date());
                        provider.sendMessage(message);
                    } finally {
                        redisMq.deMessage(queueName, id.toString());
                        redisMq.deMsgPool(message.getGroup(),id.toString());
                    }
                }
            }
        }
    }
}

/**
 * @author shikanatsu
 */


public interface RedisMqExecute {

    /**
     * 獲取隊(duì)列名稱(chēng)
     * @return
     */
    public String getQueueName();


    /**
     * 統(tǒng)一的通過(guò)執(zhí)行期執(zhí)行
     * @param message
     * @return
     */
    public boolean execute(RedisMessage message);


    /**
     * Perform thread polling
     */

    public void   threadPolling();

}

/**
 * 訂單執(zhí)行
 *
 * @author shikanatsu
 */
@Service
public class OrderMqExecuteImpl implements RedisMqExecute {


    private static Logger logger = LoggerFactory.getLogger(OrderMqExecuteImpl.class);

    public final static String name = "orderPoll:";

    @Resource
    private RedisMqConsumer redisMqConsumer;

    private RedisMqExecute mqExecute = this;

    @Resource
    private OrderService orderService;


    @Override
    public String getQueueName() {
        return name;
    }

    @Override
    /**
     * For the time being, only all orders will be processed. You can change to make orders
     */
    public boolean execute(RedisMessage message) {
        logger.info("Do orderMqPoll ; Time:{}",new Date());
  //Do 
        return true;
    }

    @Override
    /**  通過(guò)線程去執(zhí)行輪詢的過(guò)程，時(shí)間上可以自由控制 **/
    public void threadPolling() {
        ThreadUtil.execute(() -> {
            while (true) {
                redisMqConsumer.baseMonitor(mqExecute);
                ThreadUtil.sleep(5, TimeUnit.MICROSECONDS);
            }
        });
    }
}