利用Redis实现延时处理的方法实例

ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(10, new NamedThreadFactory("scheduleThreadPool"), new 
ThreadPoolExecutor.AbortPolicy());
//从消息中取出延迟时间及相关信息的代码略
int delayTime = 0;
executorService.scheduleWithFixedDelay(new Runnable() {
  @Override
  public void run() {
   //具体操作逻辑
  }},0,delayTime, TimeUnit.SECONDS);

public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory) {
 super(corePoolSize, Integer.MAX_VALUE, 0, 
 TimeUnit.NANOSECONDS,new DelayedWorkQueue(), threadFactory);}

//添加元素
ZADD key score member [[score member] [score member] …]
//根据分值及限制数量查询
ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count]
//从zset中删除指定成员
ZREM key member [member …]

public void onMessage(String topic, String message) {
  String orderId;
		int delayTime = 0;
  try {
   Map<String, String> msgMap = gson.fromJson(message, new TypeToken<Map<String, String>>() {
   }.getType());
   if (msgMap.isEmpty()) {
    return;
   }
   LOGGER.info("onMessage kafka content:{}", msgMap.toString());
	 orderId = msgMap.get("orderId");
   if(StringUtils.isNotEmpty(orderId)){
    delayTime = Integer.parseInt(msgMap.get("delayTime"));
    Calendar calendar = Calendar.getInstance();
    //计算出预计发送时间
    calendar.add(Calendar.MINUTE, delayTime);
    long sendTime = calendar.getTimeInMillis();
    RedisUtils.getInstance().zetAdd(Constant.DELAY, sendTime, orderId);
    LOGGER.info("orderId:{}---放入redis中等待发送---sendTime：{}", ---orderId:{}, sendTime);
   }
  } catch (Exception e) {
   LOGGER.info("onMessage 延时发送异常:{}", e);
  }
 }

public void run(){
  //获取批量大小
  int orderNum = Integer.parseInt(PropertyUtil.get(Constant.ORDER_NUM,"100"));
  try {
   //批量获取离发送时间最近的orderNum条数据
	 Calendar calendar = Calendar.getInstance();
	 long now = calendar.getTimeInMillis();
	 //获取无限早到现在的事件key(防止上次批量数量小于放入数量，存在历史数据未消费情况)
	 Set<String> orderIds = RedisUtils.getInstance().zrangeByScore(Constant.DELAY, 0, now, 0, orderNum);
	 LOGGER.info("task.getOrderFromRedis---size:{}---orderIds:{}", orderIds.size(), gson.toJson(orderIds));
   if (CollectionUtils.isNotEmpty(orders)){
    //删除key 防止重复发送
    for (String orderId : orderIds) {
     RedisUtils.getInstance().zrem(Constant.DELAY, orderId);
    }
	  //接下来执行发送等业务逻辑     
   }
  } catch (Exception e) {
   LOGGER.warn("task.run exception:{}", e);
  }
 }