版权申明

原创文章:本博所有原创文章,欢迎转载,转载请注明出处,并联系本人取得授权。 版权邮箱地址:banquan@mrdwy.com

使用场景

因为公司业务的原因,需要实现一个可以在指定时间执行一些任务的功能,比如订单发货通知,过期未付款订单删除,或者流程到期剩余24小时提醒等场景,需要支持由客户端发送一个任务,在指定才执行任务,并且允许客户端回收任务。

刚开始想到可以使用JDK的Timer、ScheduledExecutorService、或者调度框架Quartz等,使用定时器来执行,但是这就存在一个任务执行的实时性不够高的问题,不符合业务需求,另外由于业务量比较大,采用定时调度需要耗费巨大的资源来执行调度任务,比如定时器15分钟执行一次,那么15分钟内可能产生需要执行的任务太多了,调度服务执行不完,导致下次定时轮询到来时上一次轮询还没有结束。或者在不繁忙的时间,定时任务执行扫描任务库发现没有任务可以执行,这样会造成很多无意义的操作,无形中增加了数据库的压力。而且随着业务量的增长,这些情况会越来越明显,显然这个方案是行不通的。

那么就想到只有使用延时消息队列了,这个看上去比前面一个方法就要靠谱多了。

延时消息

需求点: 1、允许发送延时消息,可以支持延时多少时间发送,也可以指定具体的时间发送;
2、客户端可以通过消息的主键删除尚未发送成功消息;
3、需要支持消息消费者分布式部署,支持多个消费者同时消费消息;
4、支持消息的大量堆积,业务繁忙时允许消息发送适量延迟,但是必须保障不能丢消息;

然后这里我进行了一些成熟产品的选型,发现都无法完全满足上面的需求: 直接使用JDK自带的DelayQueue类,显然这个只支持单机运行,并不满足分布式消费,并且不能大量堆积消息,所有的消息都保存在计算机内存中,因此该方案否决。 使用市面上的成熟消息队列产品,主要有ActivitMq、RabbitMq、RocketMq、Kafka等产品,这些产品都能很好的满足延时消费和分布式的需求,但是都不支持回收消息功能,因此最后决定自行开发一个适合公司业务场景使用的延时消息队列。

实现方式

由于公司大量使用了Redis作为缓存数据库,因此相对来说使用起来比较方便,所以就想到了使用Redis的有序集合来实现延时消息队列,主要思路是将消费时间转换成时间戳,然后作为排序分值保存到有序队列中,每个队列代表一种业务场景,消费者循环从有序队列中获取最上一条数据,然后将分值与当前时间进行比较,如果大于当前时间,则执行消费动作,否则等待一段时间。 对于消息回收功能,则只需要将消息ID作为Redis Value值与并且业务ID关联保存起来,然后要回收消息的时候通过Redis API直接删除相应的Value就行了。

Redis有序集合数据结构

查找方式:首先通过业务类型,查到Redis的Key,然后通过Msgid找到具体哪条消息,最后删除消息。

主要实现代码

/**
客户端使用接口代码
**/
import java.util.Date;  
import java.util.List;

/**
 * @author tcrow.luo
 * @date 2019/4/22.
 * 延时消息服务类
 */
public interface SysDelayQueueService {

    /**
     * 注册服务,会自动启动一个QueueWorker
     *
     * @param consumer
     */
    void register(Consumer consumer);

    /**
     * 注销服务(暂不支持注销)
     *
     * @param consumer
     */
    void unregister(Consumer consumer);

    /**
     * 暂停程序,关闭程序时调用关闭功能安全关闭
     */
    void shutdown();

    /**
     * 系统初始化时将队列初始化到redis队列中
     */
    void init();

    /**
     * 发送消息
     *
     * @param tag
     * @param keyword  关键词,可以用作查询消息,必须唯一,例如可以使用订单编号作为关键词
     * @param reqParam
     * @param execTime 执行事件
     */
    void send(String tag, String keyword, String reqParam, Date execTime);

    /**
     * 回收消息
     *
     * @param msgId
     */
    void recover(Integer msgId);

    /**
     * 回收消息
     *
     * @param tag
     * @param keyword
     */
    void recover(String tag, String keyword);

    /**
     * 通过关键词查找消息
     *
     * @param tag
     * @param keyword
     * @return
     */
    SysDelayQueue findByKeyword(String tag, String keyword);

}
/**
 * @author tcrow.luo
 * @date 2019/4/22.
 * 定义消息消费者的模型
 */
public interface Consumer {

    /**
     * 消费消息
     *
     * @param reqParam
     */
    void consume(String reqParam);

    /**
     * 获取订阅TAG消息,用于系统启动时自动将消费者注册到注册中心订阅对应TAG的消息
     *
     * @return
     */
    String getTag();

}
import lombok.extern.slf4j.Slf4j;  
import org.springframework.data.redis.core.ZSetOperations;  
import com.alibaba.fastjson.JSONObject;  
import java.util.Set;  
//..........省略自定义类

/**
 * @author tcrow.luo
 * @date 2019/4/22.
 * 消费者工作类,系统启动时会自动启动对应消费者的工作线程
 */
@Slf4j
public class QueueWorker implements Runnable {

    private Consumer consumer;
    private RedisClient redis;
    private SysDelayQueueMapper sysDelayQueueMapper;
    private volatile boolean shutdown;

    public final static String QUEUE_WORKER = "QUEUE_WORKER";

    public QueueWorker(Consumer consumer) {
        this.consumer = consumer;
        this.redis = SpringContext.getApplicationContext().getBean(RedisClient.class);
        this.sysDelayQueueMapper = SpringContext.getApplicationContext().getBean(SysDelayQueueMapper.class);
        log.info("init [{}] queue worker success ....", consumer.getTag());
        shutdown = false;
    }

    public void shutdown() {
        this.shutdown = true;
    }

    @Override
    public void run() {
        String uuid;
        boolean lock;
        Set<ZSetOperations.TypedTuple<Object>> tuples;
        ZSetOperations.TypedTuple tuple;
        long now;
        String msgId;
        log.info("start [{}] queue loop ...", consumer.getTag());
        while (true) {
            //try{}catch{}防止线程因为意外错误而终止
            if (shutdown) {
                break;
            }
            try {
                now = System.currentTimeMillis() / 1000;

                tuples = redis.zrangeWithScores(consumer.getTag(), 0, 0);
                if (tuples == null || tuples.size() == 0) {
                    Threads.sleep(3000);
                    continue;
                }
                tuple = (ZSetOperations.TypedTuple) tuples.toArray()[0];
                uuid = UUIDUtil.getKey();
                if (now < tuple.getScore().longValue()) {
                    Threads.sleep(500);
                    continue;
                }
                msgId = (String) tuple.getValue();
                //只对消息本身加锁,允许多个线程订阅
                lock = redis.lock(QUEUE_WORKER + msgId, uuid, 3);
                if (!lock) {
                    Threads.sleep(500);
                    continue;
                }
                try {
                    SysDelayQueue sysDelayQueue = sysDelayQueueMapper.selectById(Integer.valueOf(msgId));
                    if (sysDelayQueue == null) {
                        log.error("数据异常,找不到对应的延迟消息,可能数据被异常删除,消息ID:[{}],消息类型[{}]", msgId, consumer.getTag());
                        redis.zrem(consumer.getTag(), tuple.getValue());
                        continue;
                    }
                    try {
                        consumer.consume(sysDelayQueue.getReqParam());
                    } catch (Exception e) {
                        log.error("完成延迟消息的消费,但是发生错误,消息体:[" + JSONObject.toJSONString(sysDelayQueue) + "]", e);
                    } finally {
                        //无论是否消费成功,都需要将消息设置为已消费,否则会造成消费者停止的问题
                        redis.zrem(consumer.getTag(), tuple.getValue());
                        sysDelayQueue.setMsgStatus(Const.Y);
                        sysDelayQueueMapper.updateById(sysDelayQueue);
                    }
                    log.info("完成延迟消息的消费,消息体:[{}]", JSONObject.toJSONString(sysDelayQueue));
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                } finally {
                    redis.unlock(consumer.getTag(), uuid);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                Threads.sleep(5000);
            }

        }
    }
}
/**
 * @author tcrow.luo
 * @date 2019/4/22.
 * 消息消费者初始化类,通过Spring的getBeansOfType找到所有实现Consumer接口的Bean,然后将bean通过延时队列的Service方法注册成消费者
 */
@Slf4j
@Component
public class SysDelayQueueInit implements CommandLineRunner {

    @Autowired
    private SysDelayQueueService sysDelayQueueService;

    @Override
    public void run(String... args) {

        sysDelayQueueService.init();

        Map<String, Consumer> beansOfType = SpringContext.getApplicationContext().getBeansOfType(Consumer.class);

        Set<Map.Entry<String, Consumer>> entries = beansOfType.entrySet();
        for (Map.Entry<String, Consumer> entry : entries) {
            Consumer consumer = entry.getValue();
            sysDelayQueueService.register(consumer);
        }

    }

}

这里因为业务原因没有给出SysDelayQueueService接口的实现,自己实现也很简单,基本上send方法就是把消息ID保存到redis有序队列中,而recover则是从有序队列中删除对应的数据,需要注意的是,我这边把消息的请求参数保存在了其它关系型数据库中,没有保存到Redis里面,根据业务场景也可以直接把请求参数另外保存到Redis中,作为字符串保存,Key则直接设置成msgid就行了,这样都使用Redis效率更加高。

使用方式

1、首先实现Consumer 接口,一类业务场景实现一个Consumer接口,则在系统启动时会被自动注册成为消费者;
2、消费场景直接使用SysDelayQueueService.send方法发送消息