Java实现Redis延时消息队列

• • 2.redismq 消息队列实现类

什么是延时任务

延时任务，顾名思义，就是延迟一段时间后才执行的任务。举个例子，假设我们有个发布资讯的功能，运营需要在每天早上7点准时发布资讯，但是早上7点大家都还没上班，这个时候就可以使用延时任务来实现资讯的延时发布了。只要在前一天下班前指定第二天要发送资讯的时间，到了第二天指定的时间点资讯就能准时发出去了。如果大家有运营过公众号，就会知道公众号后台也有文章定时发送的功能。总而言之，延时任务的使用还是很广泛的。

延时任务的特点

• 时间有序性
• 时间具体性
• 任务中携带详细的信息 ，通常包括 任务id， 任务的类型 ，时间点。

实现思路：

将整个redis当做消息池，以kv形式存储消息，key为id，value为具体的消息body
使用zset做优先队列，按照score维持优先级（用当前时间+需要延时的时间作为score）
轮询zset，拿出score比当前时间戳大的数据（已过期的）
根据id拿到消息池的具体消息进行消费
消费成功，删除改队列和消息
消费失败，让该消息重新回到队列

代码实现

1.消息模型

import lombok.data;
import lombok.experimental.accessors;

import javax.validation.constraints.notnull;
import java.io.serializable;

/**
 * redis 消息队列中的消息体
 * @author shikanatsu
 */
@data
@accessors(chain = true)
public class redismessage implements serializable {

    /** 消息队列组 **/
    private string group;

    /**
     * 消息id
     */
    private string id;

    /**
     * 消息延迟/ 秒
     */
    @notnull(message = "消息延时时间不能为空")
    private long delay;

    /**
     * 消息存活时间 单位：秒
     */
    @notnull(message = "消息存活时间不能为空")
    private int ttl;
    /**
     * 消息体，对应业务内容
     */
    private object body;
    /**
     * 创建时间，如果只有优先级没有延迟，可以设置创建时间为0
     * 用来消除时间的影响
     */
    private long createtime;
}

2.redismq 消息队列实现类

package com.shixun.base.redismq;

import com.shixun.base.jedis.service.redisservice;
import org.springframework.stereotype.component;

import javax.annotation.resource;

/**
 * redis消息队列
 *
 * @author shikanatsu
 */
@component
public class redismq {

    /**
     * 消息池前缀，以此前缀加上传递的消息id作为key，以消息{@link msg_pool}
     * 的消息体body作为值存储
     */
    public static final string msg_pool = "message:pool:";

    /**
     * zset队列 名称 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);
    }

    /**
     * 向队列中添加消息
     *
     * @param key
     * @param score 优先级
     * @param val
     * @return 返回消息id
     */
    public void enmessage(string key, long score, string val) {
        redisservice.zsset(key, val, score);
    }

    /**
     * 从队列删除消息
     *
     * @param id
     * @return
     */
    public boolean demessage(string key, string id) {
        return redisservice.zdel(key, id);
    }
}

3.消息生产者

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;

/**
 * 消息生产者
 *
 * @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发送消费信息{}，当前时间:{},消费时间预计{}",message.tostring(),new date(),sdf.format(delaytime));
        }catch (exception e){
            e.printstacktrace();
            logger.error("redismq 消息发送失败，当前时间:{}",new date());
            return false;
        }
        return true;
    }
}

4.消息消费者

/**
 * redis消息消费者
 * @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());
                    }
                }
            }
        }
    }
}

5. 消息执接口

/**
 * @author shikanatsu
 */


public interface redismqexecute {

    /**
     * 获取队列名称
     * @return
     */
    public string getqueuename();


    /**
     * 统一的通过执行期执行
     * @param message
     * @return
     */
    public boolean execute(redismessage message);


    /**
     * perform thread polling
     */

    public void   threadpolling();

}

6. 任务类型的实现类：可以根据自己的情况去实现对应的队列需求 

/**
 * 订单执行
 *
 * @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
    /**  通过线程去执行轮询的过程，时间上可以*控制 **/
    public void threadpolling() {
        threadutil.execute(() -> {
            while (true) {
                redismqconsumer.basemonitor(mqexecute);
                threadutil.sleep(5, timeunit.microseconds);
            }
        });
    }
}

使用事例
 1. 实现redismqexecute 接口 创建对应的轮询或者采取定时器的方式执行 和实现具体的任务。
 2.  通过messageprovider 实现相对应的消息服务和绑定队列组，通过队列组的方式执行。
 3. 提示： 采取线程的方式需要在项目启动过程中执行，采取定时器或者调度的方式可以更加动态的调整。

到此这篇关于java实现redis延时消息队列的文章就介绍到这了,更多相关java redis延时消息队列内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！