您现在的位置是：网站首页> 编程资料编程资料

redis用list做消息队列的实现示例_Redis_

2023-05-27 404人已围观

简介 redis用list做消息队列的实现示例_Redis_

生产消息服务

@Service public class RedisService {     @Autowired     private RedisTemplate 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 listOperations = redisTemplate.opsForList();         //leftPush和rightPop对应，左边入队，右边出队         listOperations.leftPush(RedisConstant.MQ_LIST, s);         //因为出队是阻塞读取的，所以上一步入队后，数据立刻就被驱走了，下一步size=0         Long size = listOperations.size(RedisConstant.MQ_LIST);         List 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 {     @Autowired     private RedisService redisService;     @GetMapping("publish")     public Object publish() {         return redisService.publish();     } }

消费消息服务，定时任务

@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 redisTemplate;     public void consumeMqList() {         ListOperations 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 {     @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("任务开始, " + lockKey + ", " + new Date());             return pjp.proceed();         } finally {             lock.unlock();             System.out.println("任务结束, " + lockKey + ", " + new Date());         }     } }