背景
項目中的流程監控，有幾種節點，需要監控每一個節點是否超時。按傳統的做法，肯定是通過定時任務，去掃描然后判斷，但是定時任務有缺點：1，數據量大會慢；2，時間不好控制，太短，怕一次處理不完，太長狀態就會有延遲。所以就想到用延遲隊列的方式去實現。
一，redis的過期key監控
1，開啟過期key監聽
在redis的配置里把這個注釋去掉
notify-keyspace-events Ex
然后重啟redis
2，使用redis過期監聽實現延遲隊列
繼承KeyExpirationEventMessageListener類，實現父類的方法，就可以監聽key過期時間了。當有key過期，就會執行這里。這里就把需要的key過濾出來，然后發送給kafka隊列。
@Component
@Slf4j
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
@Autowired
private KafkaProducerService kafkaProducerService;
public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
super(listenerContainer);
}
/**
* 針對 redis 數據失效事件，進行數據處理
* @param message
* @param pattern
*/
@Override
public void onMessage(Message message, byte[] pattern){
if(message == null || StringUtils.isEmpty(message.toString())){
return;
}
String content = message.toString();
//key的格式為 flag:時效類型:運單號 示例如下
try {
if(content.startsWith(AbnConstant.EMS)){
kafkaProducerService.sendMessageSync(TopicConstant.EMS_WAYBILL_ABN_QUEUE,content);
}else if(content.startsWith(AbnConstant.YUNDA)){
kafkaProducerService.sendMessageSync(TopicConstant.YUNDA_WAYBILL_ABN_QUEUE,content);
}
} catch (Exception e) {
log.error("監控過期key，發送kafka異常，",e);
}
}
}
可以看的出來，這種方式其實是很簡單的，但是有幾個問題需要注意，一是，這個盡量單機運行，因為多臺機器都會執行，浪費cpu，增加數據庫負擔。二是，機器頻繁部署的時候，如果有時間間隔，會出現數據的漏處理。
二，redis的zset實現延遲隊列
1，生產者實現
可以看到生產者很簡單，其實就是利用zset的特性，給一個zset添加元素而已，而時間就是它的score。
public void produce(Integer taskId, long exeTime) {
System.out.println("加入任務， taskId: " + taskId + ", exeTime: " + exeTime + ", 當前時間：" + LocalDateTime.now());
RedisOps.getJedis().zadd(RedisOps.key, exeTime, String.valueOf(taskId));
}
2,消費者實現
消費者的代碼也不難，就是把已經過期的zset中的元素給刪除掉，然后處理數據。
public void consumer() {
Executors.newSingleThreadExecutor().submit(new Runnable() {
@Override
public void run() {
while (true) {
SetString> taskIdSet = RedisOps.getJedis().zrangeByScore(RedisOps.key, 0, System.currentTimeMillis(), 0, 1);
if (taskIdSet == null || taskIdSet.isEmpty()) {
System.out.println("沒有任務");
} else {
taskIdSet.forEach(id -> {
long result = RedisOps.getJedis().zrem(RedisOps.key, id);
if (result == 1L) {
System.out.println("從延時隊列中獲取到任務，taskId:" + id + " , 當前時間：" + LocalDateTime.now());
}
});
}
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
}
可以看到這種方式其實是比上個方式要好的。因為，他的那兩個缺點都被克服掉了。多臺機器也沒事兒，也不用再擔心部署時間間隔長的問題。
總結
兩個方式都是不錯的，都能解決問題。碰到問題，多思考，多總結。
到此這篇關于redis實現延時隊列的兩種方式(小結)的文章就介紹到這了,更多相關redis 延時隊列內容請搜索腳本之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:- golang實現redis的延時消息隊列功能示例
- 使用Redis實現延時任務的解決方案
- 利用Redis實現延時處理的方法實例
