redis的list，redis用list做消息隊列

2023-10-18 阅读 25 评论 0

摘要：leftPush消息入隊，rightPop對應，消息出隊。 rightPop(RedisConstant.MQ_LIST, 0L, TimeUnit.SECONDS)阻塞出隊，0表示永久阻塞生產消息服務 @Service public class RedisService {@Autowiredprivate RedisTemplate<String, String> redisT

leftPush消息入隊，rightPop對應，消息出隊。
rightPop(RedisConstant.MQ_LIST, 0L, TimeUnit.SECONDS)阻塞出隊，0表示永久阻塞

生產消息服務

@Service
public class RedisService {@Autowiredprivate 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對應，左邊入隊，右邊出隊listOperations.leftPush(RedisConstant.MQ_LIST, s);//因為出隊是阻塞讀取的，所以上一步入隊后，數據立刻就被驅走了，下一步size=0Long 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;}
}

測試

@RestController
@RequestMapping("redisList")
public class RedisListController {@Autowiredprivate RedisService redisService;@GetMapping("publish")public Object publish() {return redisService.publish();}
}

消費消息服務，定時任務

@Component
public class RedisConsumeTask {@Autowiredprivate RedisService redisService;@TaskLock(RedisConstant.CONSUME_REDIS_LIST)@Scheduled(cron = "0/10 * * * * ?")public void consumeMqList() {redisService.consumeMqList();}
}@Service
@Slf4j
public class RedisService {@Autowiredprivate RedisTemplate<String, String> redisTemplate;public void consumeMqList() {ListOperations<String, String> listOperations = redisTemplate.opsForList();//0時間，表示阻塞永久//待機一小時后，再次發消息，消費不了了，阻塞有問題啊。還得輪尋啊//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 {@Autowiredprivate 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("任務開始, " + lockKey + ", " + new Date());return pjp.proceed();} finally {lock.unlock();System.out.println("任務結束, " + lockKey + ", " + new Date());}}}