Redis实现延时队列

缓存队列延时向接口报工，并支持多实例部署。

引入依赖

<dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-data</artifactId><version>3.17.4</version>
</dependency>

注入RedisClient

import org.springframework.util.StringUtils;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RedissonConfig {@Value("${spring.redis.host}")private String host;@Value("${spring.redis.port}")private String port;@Value("${spring.redis.password}")private String password;@Value("${spring.redis.database}")private Integer database;@Bean()public RedissonClient redissonClient() {Config config = new Config();SingleServerConfig serversConfig = config.useSingleServer();serversConfig.setAddress("redis://" + host + ":" + port);if (!StringUtils.isEmpty(password)){serversConfig.setPassword(password);}serversConfig.setDatabase(database);return Redisson.create(config);}
}

注入延时队列Bean

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** @author* redisson延迟队列*/
@Configuration
public class RedissonQueueConfig {@Beanpublic RBlockingQueue<String> rBlockingQueue(@Qualifier("redissonClient") RedissonClient redissonClient) {String queueName = "queue";return redissonClient.getBlockingQueue(queueName);}@Bean(name = "rDelayedQueue")public RDelayedQueue<String> rDelayedQueue(@Qualifier("redissonClient") RedissonClient redissonClient,@Qualifier("rBlockingQueue") RBlockingQueue<String> blockQueue) {return redissonClient.getDelayedQueue(blockQueue);}
}

编写方法

import java.util.concurrent.TimeUnit;/*** @author*/
public interface DelayQueue {/*** 发布** @param object* @return*/Boolean offer(Object object);/*** 带延迟功能的队列** @param object* @param time* @param timeUnit*/void offer(Object object, Long time, TimeUnit timeUnit);void offerAsync(Object object, Long time, TimeUnit timeUnit);Boolean offerAsync(Object object);
}import org.redisson.api.RDelayedQueue;
import org.redisson.api.RFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;/*** @author*/
@Component
public class RedissonDelayQueue implements DelayQueue {private static Logger log = LoggerFactory.getLogger(RedissonDelayQueue.class);@Resource(name = "rDelayedQueue")private RDelayedQueue<Object> rDelayedQueue;@Overridepublic Boolean offer(Object object) {return rDelayedQueue.offer(object);}@Overridepublic void offer(Object object, Long time, TimeUnit timeUnit) {rDelayedQueue.offer(object, time, timeUnit);}@Overridepublic void offerAsync(Object object, Long time, TimeUnit timeUnit) {rDelayedQueue.offerAsync(object, time, timeUnit);}@Overridepublic Boolean offerAsync(Object object) {boolean flag = false;RFuture<Boolean> rFuture = rDelayedQueue.offerAsync(object);try {flag = rFuture.get();} catch (InterruptedException | ExecutionException e) {log.info("offerAsync exception:{}", e.getMessage());e.printStackTrace();}return flag;}
}

延时任务

队列生产

@Resource(name = "rDelayedQueue")
private RDelayedQueue<Object> rDelayedQueue;public void delayedQueue(){//rDelayedQueue.size()队列中元素数量//添加到延时队列中rDelayedQueue.offerAsync("延时30秒执行任务",30, TimeUnit.SECONDS);
}

队列消费

import com.coctrl.mom.common.service.WebService;
import com.coctrl.mom.process.entity.vo.LESPassStationDetailQueueVO;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import javax.annotation.Resource;/*** @author*/
@Component
@Slf4j
public class RedissonTask {@Resource(name = "rBlockingQueue")private RBlockingQueue<Object> rBlockingQueue;@PostConstructpublic void take() {new Thread(() -> {while (true) {try {log.info("延时报工信息===============" +rBlockingQueue.take());//业务代码} catch (InterruptedException e) {log.error("延时报工失败===============" + e.getMessage());e.printStackTrace();}}}).start();}
}