快捷導(dǎo)航

redis用list做消息隊(duì)列的實(shí)現(xiàn)示例

更新時(shí)間：2022年02月18日 11:28:48 作者：絕世奇才

本文主要介紹了redis用list做消息隊(duì)列的實(shí)現(xiàn)示例，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

生產(chǎn)消息服務(wù)

@Service
public class RedisService {
? ? @Autowired
? ? private RedisTemplate<String, String> redisTemplate;


? ? public Object publish() {
? ? ? ? OrderDTO dto = new OrderDTO();
? ? ? ? dto.setId(1);
? ? ? ? dto.setCreateTime(new Date());
? ? ? ? dto.setMoney("12.34");
? ? ? ? dto.setOrderNo("orderNo1");
? ? ? ? String s = JSON.toJSONString(dto);

? ? ? ? ListOperations<String, String> listOperations = redisTemplate.opsForList();
? ? ? ? //leftPush和rightPop對(duì)應(yīng)，左邊入隊(duì)，右邊出隊(duì)
? ? ? ? listOperations.leftPush(RedisConstant.MQ_LIST, s);

? ? ? ? //因?yàn)槌鲫?duì)是阻塞讀取的，所以上一步入隊(duì)后，數(shù)據(jù)立刻就被驅(qū)走了，下一步size=0
? ? ? ? Long size = listOperations.size(RedisConstant.MQ_LIST);
? ? ? ? List<String> list = new ArrayList<>();
? ? ? ? if (size != null && size > 0) {
? ? ? ? ? ? ?list = listOperations.range(RedisConstant.MQ_LIST, 0, size - 1);
? ? ? ? }
? ? ? ? return list;

? ? }
}

測(cè)試

@RestController
@RequestMapping("redisList")
public class RedisListController {

? ? @Autowired
? ? private RedisService redisService;

? ? @GetMapping("publish")
? ? public Object publish() {
? ? ? ? return redisService.publish();
? ? }
}

消費(fèi)消息服務(wù)，定時(shí)任務(wù)

@Component
public class RedisConsumeTask {
? ? @Autowired
? ? private RedisService redisService;

? ? @TaskLock(RedisConstant.CONSUME_REDIS_LIST)
? ? @Scheduled(cron = "0/10 * * * * ?")
? ? public void consumeMqList() {
? ? ? ? redisService.consumeMqList();
? ? }
}

@Service
@Slf4j
public class RedisService {

? ? @Autowired
? ? private RedisTemplate<String, String> redisTemplate;

? ? public void consumeMqList() {
? ? ? ? ListOperations<String, String> listOperations = redisTemplate.opsForList();
? ? ? ? //0時(shí)間，表示阻塞永久
? ? ? ? //待機(jī)一小時(shí)后，再次發(fā)消息，消費(fèi)不了了，阻塞有問題啊。還得輪尋啊
? ? ? ? //String s = listOperations.rightPop(RedisConstant.MQ_LIST, 0L, TimeUnit.SECONDS);
? ? ? ? String s = listOperations.rightPop(RedisConstant.MQ_LIST);
? ? ? ? if (s == null) {
? ? ? ? ? ? return;
? ? ? ? }

? ? ? ? log.info("{} = {}", RedisConstant.MQ_LIST, s);

? ? ? ? OrderDTO dto = JSON.parseObject(s, OrderDTO.class);
? ? ? ? log.info("dto = {}", dto);
? ? }
}

日志

@Component
@Aspect
public class TaskLockAop {

? ? @Autowired
? ? private RedisLockRegistry redisLockRegistry;

? ? @Around("execution(@TaskLock * * (..))")
? ? public Object taskAround(ProceedingJoinPoint pjp) throws Throwable {

? ? ? ? TaskLock taskAnnotation = ((MethodSignature)pjp.getSignature()).getMethod().getAnnotation(TaskLock.class);

? ? ? ? String lockKey = taskAnnotation.value();
? ? ? ? Lock lock = redisLockRegistry.obtain(lockKey);
? ? ? ? try {
? ? ? ? ? ? lock.tryLock(30L, TimeUnit.SECONDS);
? ? ? ? ? ? System.out.println("任務(wù)開始, " + lockKey + ", " + new Date());

? ? ? ? ? ? return pjp.proceed();

? ? ? ? } finally {
? ? ? ? ? ? lock.unlock();
? ? ? ? ? ? System.out.println("任務(wù)結(jié)束, " + lockKey + ", " + new Date());
? ? ? ? }
? ? }
}