Redis implementiert verzögerte Nachrichtenwarteschlangen über geordnete Sätze (ZSet) verfügt über ein Score-Attribut, das zum Speichern der verzögerten Ausführungszeit verwendet werden kann.
Aber es ist eine Endlosschleife erforderlich, um die Aufgabe zu überprüfen, was Systemressourcen verbraucht.
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)})Nachdem der Benutzer auf das Aufladeelement geklickt hat und es nicht innerhalb einer halben Stunde aufgeladen wird Nach Ablauf einer Stunde senden Sie dem Benutzer eine Erinnerung an eine nicht abgeschlossene Aufladung.
Senden Sie 2 Stunden nach dem letzten Leseverhalten des Benutzers eine Erinnerung zum Weiterlesen an den Benutzer.
N Minuten nachdem sich der Benutzer neu registriert oder die App verlässt, werden entsprechende Empfehlungsnachrichten an den Benutzer gesendet.
…
Das gemeinsame Merkmal der oben genannten Szenarien besteht darin, eine bestimmte Zeitspanne nach dem Auslösen eines Ereignisses zu verzögern, bevor eine bestimmte Aufgabe ausgeführt wird. Wenn die Ereignisauslösezeit bekannt ist, kann die obige Logik ebenfalls gelten entspricht einem bestimmten Zeitpunkt (Zeitpunkt des Ereignisauslösers + Länge der Verzögerungszeit), um bestimmte Aufgaben auszuführen.
Implementierung der Verzögerungswarteschlange
In einer eigenständigen Umgebung enthält JDK bereits viele Komponenten, die Verzögerungswarteschlangenfunktionen implementieren können, wie z. B. DelayQueue, Timer , ScheduledExecutorService und andere Komponenten können leicht verzögerte Aufgaben erstellen. Bei der Verwendung der oben genannten Komponenten besteht jedoch im Allgemeinen die Gefahr eines Aufgabenverlusts Es wird neu gestartet und die Aufgabenskala ist durch den Speicher begrenzt. Außerdem ist es nicht flexibel und eignet sich normalerweise für Einzelprozess-Clientprogramme oder Projekte mit geringen Aufgabenanforderungen.
JDK nicht zuverlässig und effizient implementiert werden. Normalerweise ist die Einführung von Middleware oder Frameworks von Drittanbietern erforderlich. Zum Beispiel das gängige klassische Task-Scheduling-Framework Quartz oder andere auf diesem Framework basierende Frameworks xxl-job. Die Hauptfunktion dieser Frameworks besteht darin, geplante oder periodische Aufgaben zu implementieren Als Redis und RabbitMQ nicht weit verbreitet waren, wurden gängige Funktionen wie Timeout und unbezahlte Auftragsstornierung durch geplante Aufgaben implementiert und geplante Abfragen wurden verwendet, um festzustellen, ob die Bestellung ausgeführt wurde verarbeitet wurde. Der Zeitpunkt, der die Ausführung auslöst, ist erreicht. 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
MQ als Verzögerungswarteschlange zu verwenden, z. B. die Verwendung von TTL und Dead Letter von RabbitMQ Warteschlange zum Implementieren der Nachricht. Verzögerte Zustellung, da die zugestellte MQ-Nachricht nicht einfach gelöscht oder geändert werden kann, d. h. die Aufgabe kann nicht abgebrochen oder der Zeitpunkt der Aufgabenausführung geändert werden, und die Nachricht kann nicht einfach gelöscht oder geändert werden leicht deduplizierbar. Daher haben wir uns nicht dafür entschieden, MQ zu verwenden, um die Verzögerungswarteschlange im Projekt zu implementieren. 🎜🎜Die Datenstruktur von Redis zset kann auch den Effekt einer verzögerten Warteschlange erzielen und ist flexibler. Sie kann einige Funktionen von MQ realisieren kann nicht erreicht werden, daher verwendet das Projekt schließlich Redis, um die Verzögerungswarteschlange zu implementieren, und optimiert und kapselt sie. 🎜🎜Das Implementierungsprinzip besteht darin, das score-Attribut von zset zu verwenden, um die Elemente im zset festzulegen Die Sammlung wird entsprechend dem >score von klein nach groß sortiert und Elemente werden über den Befehl zadd zu zset hinzugefügt, wie im Folgenden gezeigt Befehl, wobei value der Wert eine verzögerte Aufgabennachricht ist und das Nachrichtenformat entsprechend dem Unternehmen definiert werden kann. Der score-Wert ist der Zeitpunkt der Aufgabenausführung, z als 13-stelliger Millisekunden-Zeitstempel. 🎜zrangebyscore delayqueue 0 timestamp limit 0 1000
score-Wert kleiner ist als der aktuelle Zeitstempel von zset, und das Ergebnis ist das, was am aktuellen Zeitknoten ausgeführt werden muss. Die Aufgabe wird über den Befehl zrangebyscore abgerufen, wie im folgenden Befehl gezeigt, wobei timestamp der aktuelle Zeitstempel ist und < code>limit kann verwendet werden, um jeden Pull zu begrenzen. Die Anzahl der Datensätze, um zu verhindern, dass die Anzahl der gleichzeitig abgerufenen Datensätze zu groß wird. 🎜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('zrangebyscore',KEYS[1],0,ARGV[1],'limit',0,1)\n" +
"if #items == 0 then\n" +
"\treturn ''\n" +
"else\n" +
"\tredis.call('zremrangebyrank',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实现延时队列的原理，对其实现方案进行比较与优化，并将最终方案实际运用于项目需求中。
Das obige ist der detaillierte Inhalt vonSo implementieren Sie eine Verzögerungswarteschlange in Redis. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
Häufig verwendete Datenbanksoftware
Was sind In-Memory-Datenbanken?
Welches hat eine schnellere Lesegeschwindigkeit, Mongodb oder Redis?
So verwenden Sie Redis als Cache-Server
Wie Redis die Datenkonsistenz löst
Wie stellen MySQL und Redis die Konsistenz beim doppelten Schreiben sicher?
Welche Daten speichert der Redis-Cache im Allgemeinen?
Was sind die 8 Datentypen von Redis?
![[Web-Frontend] Node.js-Schnellstart](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
