Redis怎麼實作延遲隊列

Redis實作延遲佇列

Redis延遲佇列

Redis 是透過有序集合（ZSet）的方式來實現延遲訊息佇列的，ZSet 有一個Score 屬性可以用來儲存延遲執行的時間。

但需要無限循環檢查任務，會消耗系統資源

class RedisDelayQueue(object):
    """Simple Queue with Redis Backend
    dq = RedisDelayQueue('delay:commtrans')
    dq.put( 5 ,{'info':'测试 5555','time': timestamp_to_datetime_str(t + 5)})
    print(dq.get())
    """
    def __init__(self, name, namespace='queue'):
        """The default connection parameters are: host='localhost', port=6379, db=0"""
        self.__db = get_redis_engine(database_name='spdb')
        self.key = '%s:%s' % (namespace, name)
    def qsize(self):
        """Return the approximate size of the queue."""
        return self.__db.zcard(self.key)
    def empty(self):
        """Return True if the queue is empty, False otherwise."""
        return self.qsize() == 0
    def rem(self, value):
        return self.__db.zrem(self.key, value)
    def get(self):
        # 获取任务，以0和当前时间为区间，返回一条在当前区间的记录
        items = self.__db.zrangebyscore(self.key, 0, int(time.time()), 0, 1)
        if items:
            item = items[0]
            if self.rem(item):  # 解决并发问题  如能删就让谁取走
                return json.loads(item)
        return None
    def put(self, interval, item):
        """:param interval 延时秒数"""
        # 以时间作为score，对任务队列按时间戳从小到大排序
        """Put item into the queue."""
        d = json.dumps(item)
        return self.__db.zadd(self.key, {d: int(time.time()) + int(interval)})

Redis實作延時佇列的最佳化方案

延時佇列的應用

近期在開發部門的新項目，其中有個關鍵功能就是智慧推送，即根據用戶行為在特定的時間點向用戶推送相應的提醒訊息，例如以下業務場景：

• 在使用者點擊儲值項目後，半小時內未儲值，並向使用者推播儲值未完成提醒。

• 在用戶最近一次閱讀行為2小時後，向用戶推送繼續閱讀提醒。

• 在用戶新註冊或退出應用程式N分鐘後，向用戶推送合適的推薦訊息。

• …

上述場景的共同特徵就是在某事件觸發後延遲一定時間後再執行特定任務，若事件觸發時間點可知，則上述邏輯也可等價於在指定時間點（事件觸發時間點延遲時間長度）執行特定任務。

實作這類需求一般採用延時佇列，其中建立的延時訊息中需要包含任務延遲時間或任務執行時間點等信息，當任務滿足時間條件需要執行時，該訊息就會被消費，也就是說可以指定隊列中的消息在哪個時間點被消費。

延時佇列的實作

在單機環境中，JDK已經自帶了許多能夠實現延時佇列功能的元件，例如DelayQueue, Timer, ScheduledExecutorService等元件，都可以較為簡單地建立延時任務，但上述元件使用一般需要把任務儲存在記憶體中，服務重啟存在任務遺失風險，且任務規模體量受記憶體限制，同時也造成長時間記憶體佔用，並不靈活，通常適用於單一進程客服端程式或對任務要求不高的項目。

在分散式環境下，僅使用JDK自帶元件並不能可靠且有效率地實現延時佇列，通常需要引入第三方中間件或框架。

例如常見的經典任務排程框架Quartz或基於此框架的xxl-job等其它框架，這些框架的主要功能是實現定時任務或週期性任務，在Redis、RabbitMQ還未廣泛應用時，譬如常見的超時未支付取消訂單等功能都是由定時任務實現的，透過定時輪詢來判斷是否已到達觸發執行的時間點。

但由於定時任務需要一定的週期性，週期掃描的間隔時間不好控制，太短會造成很多無意義的掃描，且增大系統壓力，太長又會造成執行時間誤差太大，且可能造成單次掃描所處理的堆積記錄數量過大。

此外，利用MQ做延時佇列也是一種常見的方式，例如透過RabbitMQ的TTL和死信佇列實作訊息的延遲投遞，考慮到投遞出去的MQ訊息無法方便地實現刪除或修改,即無法實現任務的取消或任務執行時間點的更改，同時也不能方便地對訊息進行去重，因此在專案中並未選擇使用MQ實作延時佇列。

Redis的資料結構zset，同樣可以實現延遲佇列的效果，且更加靈活，可以實現MQ無法做到的一些特性，因此專案最終採用Redis實現延時佇列，並對其進行最佳化與封裝。

實作原理是利用zset的score屬性，redis會將zset集合中的元素依照score進行從小到大排序，透過zadd指令向zset#加入元素，如下述指令所示，其中value值為延時任務訊息，可依業務定義訊息格式，score值為任務執行的時間點，例如13位元毫秒時間戳記。

zadd delayqueue 1614608094000 taskinfo

任務加入後，取得任務的邏輯只要從zset中篩選score值小於目前時間戳記的元素，所得結果便是目前時間節點下需要執行的任務，透過zrangebyscore指令來獲取，如下述指令所示，其中timestamp為當前時間戳，可用limit限制每次拉取的記錄數，防止單次獲取記錄數過大。

zrangebyscore delayqueue 0 timestamp limit 0 1000

在实际实现过程中，从zset中获取到当前需要执行的任务后，需要先确保将任务对应的元素从zset中删除，删除成功后才允许执行任务逻辑，这样是为了在分布式环境下，当存在多个线程获取到同一任务后，利用redis删除操作的原子性，确保只有一个线程能够删除成功并执行任务，防止重复执行。

实际任务的执行通常会再将其发送至MQ异步处理，将“获取任务”与“执行任务”两者分离解耦，更加灵活，“获取任务”只负责拿到当前时间需要执行的任务，并不真正运行任务业务逻辑，因此只需相对少量的执行线程即可，而实际的任务执行逻辑则由MQ消费者承担，方便调控负载能力。

整体过程如下图所示。

采用zset做延时队列的另一个好处是可以实现任务的取消和任务执行时间点的更改，只需要将任务信息从zset中删除，便可取消任务，同时由于zset拥有集合去重的特性，只需再次写入同一个任务信息，但是value值设置为不同的执行时间点，便可更改任务执行时间，实现单个任务执行时间的动态调整。

了解实现原理后，再进行具体编程实现。创建延时任务较为简便，准备好任务消息和执行时间点，写入zset即可。获取延时任务最简单的方案是通过定时任务，周期性地执行上述逻辑，如下代码所示。

@XxlScheduled(cron = "0/5 * * * * ?", name = "scan business1 delayqueue")
public void scanBusiness1() {
	// 某业务逻辑的zset延迟队列对应的key
	String zsetKey = "delayqueue:business1";
	while (true) {
		// 筛选score值小于当前时间戳的元素，一次最多拉取1000条
		Set<String> tasks = stringRedisTemplate.opsForZSet().rangeByScore(zsetKey, 0, System.currentTimeMillis(), 0, 1000);
		if (CollectionUtils.isEmpty(tasks)) {
			// 当前时间下已没有需要执行的任务，结束本次扫描
			return;
		}
		for (String task : tasks) {
			// 先删除，再执行，确保多线程环境下执行的唯一性
			Boolean delete = stringRedisTemplate.delete(task);
			if (delete) {
				// 删除成功后，将其再发送到指定MQ异步处理，将“获取任务”与“执行任务”分离解耦
				rabbitTemplate.convertAndSend("exchange_business1", "routekey_business1", task);
			}
		}
	}
}

上述方案使用xxl-job做分布式定时任务，间隔5秒执行一次，代码借助spring提供的api来完成redis和MQ的操作。

由于是分布式定时任务，每次执行只有一个线程在获取任务，机器利用率低，当数据规模较大时，单靠一个线程无法满足吞吐量要求，因此这种方案只适用于小规模数据量级别。

此处间隔时间也可适当调整，例如缩短为1秒，调整所需考虑原则在上文已提到：间隔太短会造成很多无意义的扫描，且增大系统压力，太长又会造成执行时间误差太大。

为了提升整体吞吐量，考虑不使用分布式定时任务，对集群内每台机器（或实例）均设置独立的定时任务，同时采用多个zset队列，以数字后缀区分。

假设有M个zset队列，创建延时消息时选取消息的某个ID字段，计算hash值再对M取余，根据余数决定发送到对应数字后缀的zset队列中（分散消息，此处ID字段选取需要考虑做到均匀分布，不要造成数据倾斜）。

队列数量M的选取需要考虑机器数量N，理想情况下有多少台机器就定义多少个队列，保持M与N基本相等即可。

因为队列太少，会造成机器对队列的竞争访问处理，队列太多又会导致任务得不到及时的处理。

为了灵活应对集群机器数量的变化，建议将队列数量配置为动态可调的，例如采用分布式配置中心的方案。

每台机器在触发定时任务时，需要通过适当的负载均衡来决定从哪个队列拉取消息，负载均衡的好坏也会影响整个集群的效率，如果负载分布不均可能会导致多台机器竞争处理同一队列，降低效率。

一个简单实用的做法是利用redis的自增操作再对队列数量取余即可，只要保持队列数量和机器数量基本相等，这种做法在很大程度上就可以保证不会有多台机器竞争同一队列。

至于每台机器从对应zset中的任务获取逻辑，仍然和前面代码一致。以上方式简化实现代码如下所示。

@Scheduled(cron = "0/5 * * * * ?")
public void scanBusiness1() {
	// 队列数量M，考虑动态配置，保持和机器数量基本一致
	int M = 10;
	// redis自增key，用于负载均衡
	String incrKey = "incrkey:delayqueue:business1";
	// 每台机器执行时，从不同的zset中拉取消息，尽量确保不同机器访问不同zset
	String zsetKey = "delayqueue:business1:" + (stringRedisTemplate.opsForValue().increment(incrKey) % M);
	while (true) {
		// 此处逻辑和前面代码一致，省略。。。
	}
}

上述方案和第一种方案的主要的不同点在于zsetKey的获取上，这里是根据负载均衡算法算出来的，确保每台机器访问不同zset并拉取消息，同时定时任务采用spring提供的进程内注解@Scheduled，集群内每台机器都会间隔5秒执行，因此相比之前的方案，能够较为明显地提升整个集群的吞吐量。

但是这种方案的步骤相对更为复杂，需要动态配置队列数量，同时在创建延时任务时需要选择合适的消息ID字段来决定发送的目标zset队列，此处还要考虑均匀分布，整体实现要考虑的因素较多。

上面一种方案已经能够较好地满足整体吞吐量要求，但其缺点是步骤相对复杂，因此项目中没有采用这种方案，而是采用下面一种也能满足吞吐量要求，步骤相对简单，又方便通用化的方案。

该方案不使用定时任务，而是单独启动后台线程，在线程中执行永久循环，每次循环逻辑为：从目标zset中获取score值小于当前时间戳的元素集合中的score最小的那个元素，相当于获取当前时间点需要执行且执行时间点最早的那个任务，如果获取不到，表示当前时间点下暂无需要执行的任务，则线程休眠100ms（可视情况调整），否则，对获取到的元素进行处理，在分布式多线程环境下，仍然需要先删除成功才能进行处理。

此外，考虑到每个线程获取元素后都需要再次访问redis尝试删除操作，为了避免多线程争抢浪费资源，降低效率，这里采用lua脚本将获取和删除操作原子化。lua脚本逻辑代码如下所示。

local zsetKey = 'delayqueue'
local timestamp = 1614608094000
local items = redis.call('zrangebyscore',zsetKey,0,timestamp,'limit',0,1)
if #items == 0 then
    return ''
else
    redis.call('zremrangebyrank',zsetKey,0,0)
    return items[1]
end

其中timestamp为当前时间戳，通过在zrangebyscore命令中指定limit为1来获取score最小的元素，若获取不到，即结果集长度为0，则返回空字符串，否则，通过zremrangebyrank命令删除头部元素，即score最小的元素，也就是之前获取到的那个元素，由于redis内部保证lua脚本的原子性，上述获取并删除的操作能够运行无误。具体JAVA实现中还对其进行了多线程操作的封装和通用化的抽象，使不同业务都能够使用该组件实现延时队列。具体实现代码如下所示。

/**
 * 基于ZSET实现消息延迟处理，score存储执行时间点，到达时间点即会向指定队列发送该消息；
 * 定义一个继承本类的bean即可；
 */
public abstract class AbstractDelayedMsgScanTrigger implements Runnable, DisposableBean {
	private static final RedisScript<String> TRY_GET_AND_DEL_SCRIPT;
	static {
		// 获取并删除的lua脚本，使用spring提供的api
		String sb = "local items = redis.call(&#39;zrangebyscore&#39;,KEYS[1],0,ARGV[1],&#39;limit&#39;,0,1)\n" +
				"if #items == 0 then\n" +
				"\treturn &#39;&#39;\n" +
				"else\n" +
				"\tredis.call(&#39;zremrangebyrank&#39;,KEYS[1],0,0)\n" +
				"\treturn items[1]\n" +
				"end";
		// 自有工具类，只要能创建出spring包下的 RedisScript 的实现类对象均可
		TRY_GET_AND_DEL_SCRIPT = RedisScriptHelper.createScript(sb, String.class);
	}
	private final ThreadPoolExecutor EXECUTOR = new ThreadPoolExecutor(getThreadNum(), getThreadNum(),
			0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(getThreadNamePrefix()));
	private volatile boolean quit = false;
	@Autowired
	private StringRedisTemplate stringRedisTemplate;
	@Autowired
	private RabbitTemplate rabbitTemplate;
	@PostConstruct
	public void startScan() {
		// bean构建完成后，启动若干执行线程
		int threadNum = getThreadNum();
		for (int i = 0; i < threadNum; i++) {
			EXECUTOR.execute(this);
		}
	}
	@Override
	public void run() {
		while (!quit) {
			try {
				// 循环，采用lua获取当前需要执行的任务并将其从redis中删除
				String msg = stringRedisTemplate.execute(TRY_GET_AND_DEL_SCRIPT,
						Lists.newArrayList(getDelayedMsgSourceKey()), String.valueOf(System.currentTimeMillis()));
				if (StringUtils.isNotBlank(msg)) {
					// 消息不为空，表示获取任务成功，将其再发送到指定MQ异步处理，将“获取任务”与“执行任务”分离解耦
					rabbitTemplate.convertAndSend(getSendExchange(), getSendRoutingKey(), msg);
				} else {
					// 获取不到任务，表示当前时间点下暂无需要执行的任务，则线程休眠1S（可视情况调整）
					SleepUtils.sleepSeconds(1);
				}
			} catch (Exception e) {
				Logs.MSG.error("delayed msg scan error, sourceKey:{}", getDelayedMsgSourceKey(), e);
			}
		}
	}
	@Override
	public void destroy() throws Exception {
		quit = true;
	}
	public void setQuit(boolean quit) {
		this.quit = quit;
	}
	/**
	 * 获取消息的工作线程数量
	 */
	protected abstract int getThreadNum();
	/**
	 * 线程名称前缀，方便问题定位
	 */
	protected abstract String getThreadNamePrefix();
	/**
	 * 存放延迟消息的ZSET队列名
	 */
	protected abstract String getDelayedMsgSourceKey();
	/**
	 * 消息到达执行时间点时将其通过指定 exchange 发送到实时消费队列中
	 */
	protected abstract String getSendExchange();
	/**
	 * 消息到达执行时间点时将其通过指定 routingKey 发送到实时消费队列中
	 */
	protected abstract String getSendRoutingKey();
}

在具体业务应用中，只需定义一个继承上述类的bean即可，需要实现的方法主要是提供一些配置，比如该业务对应的zset延时队列名称，同时工作拉取消息的线程数量，由于采用rabbitMq，因此这里需要提供exchange和routingKey。

实际使用中只需向该zset队列中添加消息，并将score设为该任务需要执行的时间点（此处为13位毫秒时间戳），则到该时间点后，上述组件便会将该消息从zset中取出并删除，再将其通过指定的路由发送到实时MQ消费队列中，由消费者负责执行任务业务逻辑。目前该组件在项目中正常平稳运行。

注意：

本文结合项目中的实际需求介绍了延时队列的应用场景，分析了延时队列的多种实现，重点讲述了利用redis实现延时队列的原理，对其实现方案进行比较与优化，并将最终方案实际运用于项目需求中。

以上是Redis怎麼實作延遲隊列的詳細內容。更多資訊請關注PHP中文網其他相關文章！