一、添加POM依赖
<!--redisson-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.12.5</version>
</dependency>
二、增加redisson配置
package com.zensun.framework.config;
import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.codec.JsonJacksonCodec;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import java.io.IOException;
/**
* redis配置
*
* @author gmk
*/
@Configuration
@EnableCaching
public class RedisConfig extends CachingConfigurerSupport {
@Value("${spring.redis.host}")
private String host;
@Value("${spring.redis.port}")
private int port;
@Value("${spring.redis.password}")
private String password;
@Value("${spring.redis.database}")
private int database;
@Bean(destroyMethod = "shutdown")
public RedissonClient redisson() throws IOException {
Config config = new Config();
if (StrUtil.isNotBlank(password)) {
config.useSingleServer().setAddress("redis://" + host + ":" + port).setPassword(password).setDatabase(database);
} else {
config.useSingleServer().setAddress("redis://" + host + ":" + port).setDatabase(database);
}
config.setCodec(new JsonJacksonCodec());
return Redisson.create(config);
}
}
三、编写队列服务类
import java.util.concurrent.TimeUnit;
/**
* @Description 延迟队列服务接口 TODO
* @Author yuqianwei
* @Date 2020 2020/12/10 11:58
*/
public interface DelayQueueService<T> {
/**
* 推送数据
*
* @param data
* @param queueName
*/
void pushData(T data, String queueName);
/**
* 推送数据
*
* @param data
* @param time
* @param timeUnit
* @param queueName
*/
void pushData(T data, long time, TimeUnit timeUnit, String queueName);
/**
* 拉取数据
*
* @param queueName
* @return
*/
T pullData(String queueName);
}
import com.zensun.common.utils.spring.SpringUtils;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.TimeUnit;
/**
*
* 每一个RedisDelayQueueServiceImpl对象 是一个队列通道 多个队列通道就用多个RedisDelayQueueServiceImpl对象
* @Description redis延迟队列服务实现类 TODO
* @Author yuqianwei
* @Date 2020 2020/12/10 12:01
*/
public class RedisDelayQueueServiceImpl<T> implements DelayQueueService<T> {
private static RedissonClient redissonClient;
static {
redissonClient = SpringUtils.getBean(RedissonClient.class);
}
private final String queueName;
private Logger logger = LoggerFactory.getLogger(getClass());
private RBlockingQueue<T> blockingQueue;
private RDelayedQueue<T> delayedQueue;
public RedisDelayQueueServiceImpl(String queueName) {
this.queueName = queueName;
this.blockingQueue = getBlockingQueue();
this.delayedQueue = getDelayQueue(blockingQueue);
}
/**
* 获取阻塞队列
*
* @return
*/
private RBlockingQueue<T> getBlockingQueue() {
return redissonClient.getBlockingQueue(queueName);
}
/**
* 获取延迟队列
*
* @param blockingQueue
* @return
*/
private RDelayedQueue<T> getDelayQueue(RBlockingQueue<T> blockingQueue) {
return redissonClient.getDelayedQueue(blockingQueue);
}
/**
* 推送数据
*
* @param data
* @param queueName
*/
@Override
public void pushData(T data, String queueName) {
pushData(data, 0, TimeUnit.MILLISECONDS, queueName);
}
/**
* 推送数据
*
* @param data
* @param time
* @param timeUnit
* @param queueName
*/
@Override
public void pushData(T data, long time, TimeUnit timeUnit, String queueName) {
delayedQueue.offerAsync(data, time < 0 ? 0 : time, timeUnit);
}
/**
* 拉取数据
*
* @param queueName
* @return
*/
@Override
public T pullData(String queueName) {
T currentData = null;
try {
currentData = blockingQueue.take();
} catch (InterruptedException e) {
e.printStackTrace();
try {
TimeUnit.MILLISECONDS.sleep(500);
} catch (InterruptedException interruptedException) {
interruptedException.printStackTrace();
}
}
logger.info("获取到的数据 : {}", currentData);
return currentData;
}
}
四、DEMO
生产者：
long currentTimeMillis = System.currentTimeMillis();
long delayStartTime = advertInfo.getStartTime().getTime() - currentTimeMillis;
//开始时间在现在之后
try {
delayToStartQueueService.pushData(String.valueOf(advertInfo.getAdvertId()), delayStartTime, TimeUnit.MILLISECONDS, Constants.ADVERT_TOSTART_QUEUEKEY);
log.info("成功推送到队列 : {}, 广告信息id : {}, 延迟时间 : {} 秒", Constants.ADVERT_TOSTART_QUEUENAME, advertInfo.getAdvertId(), delayStartTime / 1000);
} catch (Exception e) {
e.printStackTrace();
log.error("失败推送到队列 : {}, 广告信息id : {}", Constants.ADVERT_TOSTART_QUEUENAME, advertInfo.getAdvertId());
}
消费者：
import cn.hutool.core.util.StrUtil;
import com.zensun.business.domain.AdvertInfo;
import com.zensun.business.domain.dto.PatchAdvertInfo;
import com.zensun.business.queue.DelayQueueService;
import com.zensun.business.queue.RedisDelayQueueServiceImpl;
import com.zensun.business.service.IAdvertInfoService;
import com.zensun.common.constant.Constants;
import com.zensun.common.enums.domain.AdvertInfoEnum;
import com.zensun.common.utils.DateUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Date;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
/**
* @Description 自动更改广告开始状态定时任务 TODO
* @Author gmk
* @Date 2020 2021/3/17 16:50
*/
@Slf4j
@Component
public class AutoNotStartToStartAdvertInfoStatusTask implements SmartInitializingSingleton {
@Autowired
private IAdvertInfoService advertInfoService;
/**
* 在所有单例bean实例化后 进行赋值, 否则有空指针错误, 后期优化
* 见此方法 afterSingletonsInstantiated
*/
private DelayQueueService<String> delayQueueService;
//定时任务是用来扫描数据内 没有被加入到队列的数据 用来兜底
/**
* 广告更改状态
//@Scheduled(fixedRate = SCHEDULE_RATE)
public void autoOffSaleGoodsSource() {
log.info("启动广告开始自动修改状态扫描--------------->");
try {
long currentTimeMillis = System.currentTimeMillis();
Date now = new Date(currentTimeMillis);
//扫描区间 扫描按照定时任务时间 只扫描离开始时间最近的一个区间
// （如 定时任务每10分钟执行一次 只扫描现在时间+10分钟大于等于开始时间的数据 ）
//long nextTime = SCHEDULE_RATE + currentTimeMillis;
long nextTime = currentTimeMillis;
Date futureTime = new Date(nextTime);
LambdaQueryWrapper<AdvertInfo> a = new LambdaQueryWrapper<>();
a.eq(AdvertInfo::getAdvertStatus, AdvertInfoEnum.Status.NOTSTART.getCode())
.apply("unix_timestamp(start_time) >= {0}", currentTimeMillis / 1000)
.apply("unix_timestamp(start_time) <= {0}", nextTime / 1000);
List<AdvertInfo> advertInfoList = advertInfoService.list(a);
log.info("扫描到开始时间在 : {}与 : {}之间 的广告信息 : {} 条", DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, now), DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, futureTime), advertInfoList.size());
if (CollectionUtils.isNotEmpty(advertInfoList)) {
advertInfoList.forEach(advertInfo -> {
long delayTime = advertInfo.getEndTime().getTime() - currentTimeMillis;
try {
// delayQueueService.pushData(String.valueOf(advertInfo.getAdvertId()), delayTime, TimeUnit.MILLISECONDS, queueName);
//log.info("推送到队列 : {}, 开始时间即将和现在时间一样的广告信息id : {}, 延迟时间 : {} 秒", queueName, advertInfo.getAdvertId(), delayTime / 1000);
} catch (Exception e) {
//log.error("推送到队列 : {}, 开始时间即将和现在时间一样的广告信息id : {}, 延迟时间 : {} 秒, 发生异常 : {}",
// queueName, advertInfo.getAdvertId(), delayTime / 1000, e.getMessage(), e);
}
});
}
} catch (Exception e) {
log.error("自动更改广告开始状态定时任务执行异常 : {}", e.getMessage(), e);
}
} */
/**
* 启动延迟队列监听
*/
public void listenerDelayQueueStart() {
Thread thread = new Thread(() -> {
log.info("启动监听{}:的数据...", Constants.ADVERT_TOSTART_QUEUENAME);
while (true) {
try {
String data = delayQueueService.pullData(Constants.ADVERT_TOSTART_QUEUEKEY);
log.info("取到 : {}, 的数据 : {}", Constants.ADVERT_TOSTART_QUEUENAME, data);
if (StrUtil.isNotBlank(data)) {
Date nowDate = DateUtils.getNowDate();
Long advertId = Long.valueOf(data);
AdvertInfo byId = Optional.ofNullable(advertInfoService.getById(advertId)).orElse(new AdvertInfo());
//消费端做幂等性 避免重复消费
if (byId.getAdvertStatus() != null && AdvertInfoEnum.Status.NOTSTART.getCode().equals(byId.getAdvertStatus()) && DateUtils.compareTo(nowDate, byId.getStartTime()) == 0) {
//修改广告的状态
PatchAdvertInfo patchAdvertInfo = new PatchAdvertInfo();
patchAdvertInfo.setAdvertId(advertId);
patchAdvertInfo.setAdvertStatus(AdvertInfoEnum.Status.START.getCode());
advertInfoService.updateAdvertInfoStatusById(patchAdvertInfo);
log.info("队列执行成功：{}，时间为：{}，广告id : {}", Constants.ADVERT_TOSTART_QUEUENAME, DateUtils.getTime(), advertId);
} else {
log.info("队列执行失败：{}，时间为：{}，广告id : {}，未获取到数据或者数据不符合消费规则", Constants.ADVERT_TOSTART_QUEUENAME, DateUtils.getTime(), advertId);
}
}
} catch (Exception e) {
log.info("队列执行失败：{}，时间为：{}", Constants.ADVERT_TOSTART_QUEUENAME, e.getMessage());
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException interruptedException) {
interruptedException.printStackTrace();
}
}
}
});
thread.setName(Constants.ADVERT_TOSTART_QUEUENAME + "-线程");
thread.setDaemon(true);
thread.start();
}
@Override
public void afterSingletonsInstantiated() {
delayQueueService = new RedisDelayQueueServiceImpl(Constants.ADVERT_TOSTART_QUEUEKEY);
listenerDelayQueueStart();
}
}
