java redis实现消息队列功能

末蓝、 2024-03-17 13:09 73阅读 0赞

#### java redis实现消息队列功能 ####

*   *  背景：
     *   *  1.1.核心代码
         *  1.2.可以根据不同业务类型创建不同枚举（相当于mq名称）
         *  1.3.测试接口
     *  总结

### 背景： ###

**需求：** 业务中需要批量处理任务，且需要每个任务间隔一段时间。最好在不同服务器同时运行不影响每个任务间隔。  
**部署环境：** 没有mq队列，有redis。  
秉着尽量不多增加系统复杂度的情况，使用redis来实现队列功能。

首先看一下代码：

#### 1.1.核心代码 ####

import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang3.StringUtils;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.Resource;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @program: wys-service
     * @description:
     * @author: wuyuanshn
     * @create: 2023-06-07 14:20
     **/
    @Slf4j
    @Service
    public class ListRedisQueue {
        
        @Resource
        private RedisServiceUtils redisServiceUtils;
    
    
        //队列名 OBJECT可以替换为自己的项目名
        public static final String LOCK_KEY = "OBJECT_LIST_QUEUE_%s";
    
        @Resource
        private RedisTemplate redisTemplate;
    
        public void produce(String key, String message) {
        
            redisTemplate.opsForList().rightPush(key, message);
    
        }
    
        public void produce(String key, Object... vs) {
        
            redisTemplate.opsForList().rightPush(key, vs);
        }
    
        public void produce(String key, List<String> messageList) {
        
            redisTemplate.opsForList().rightPushAll(key, messageList);
        }
    
        public void consume(String key) {
        
            String lockKey = String.format(LOCK_KEY, key);
            try {
        
                boolean lock = redisServiceUtils.getLock(lockKey, 1800);
                if (!lock) {
        
                    log.info("ListRedisQueue consume lock ");
                    return;
                }
    
                while (true) {
        
                    String msg = (String) redisTemplate.opsForList().leftPop(key);
                    if (StringUtils.isBlank(msg)) {
        
                        log.info("ListRedisQueue consume 获取 end");
                        //执行队列结束触发任务
                        executeEndByKey(key);
                        return;
                    }
                    log.info("ListRedisQueue consume 获取消息：{}", msg);
                    //执行队列任务
                    executeByKey(key, msg);
    
                    //更新key过期时间
                    redisTemplate.expire(lockKey, 15, TimeUnit.MINUTES);
    
                    try {
        
                        Random random = new Random();
                        int i = random.nextInt(2000) + 1000;
                        log.info("ListRedisQueue consume sleep：{}", i);
                        Thread.sleep(i);
                    } catch (Exception e) {
        
                        log.error("ListRedisQueue consume sleep error", e);
                    }
                }
            } catch (Exception e) {
        
                log.error("ListRedisQueue consume error", e);
            } finally {
        
                redisServiceUtils.releaseLock(lockKey);
            }
    
        }
    
        /**
         * 执行队列任务
         *
         * @param key
         * @param value
         */
        public void executeByKey(String key, String value) {
        
            ListQueueNameEnum structureEnum = ListQueueNameEnum.getStructureEnum(key);
            if (structureEnum == null) {
        
                return;
            }
            switch (structureEnum) {
        
                case URL_BASE:
                    if (StringUtils.isNotBlank(value)) {
        
                        try {
        
                            //业务
                        } catch (Exception e) {
        
                            log.error("ListRedisQueue executeByKey ERROR ", e);
                        }
                    }
                    return;
                default:
            }
        }
    
        /**
         * 执行队列结束触发任务
         *
         * @param key
         */
        public void executeEndByKey(String key) {
        
            ListQueueNameEnum structureEnum = ListQueueNameEnum.getStructureEnum(key);
            if (structureEnum == null) {
        
                return;
            }
            switch (structureEnum) {
        
                case URL_BASE:
    				// 结束业务
                    return;
                default:
    
            }
    
    
        }
    
    
    }

**代码讲解：**

*  每个任务（相当于mq消息）接收到后需要如何处理，就在此方法中编辑 executeByKey 。
 *  任务（mq消息）处理完毕时，如果需要做相关逻辑操作，如通知某程序等，在executeEndByKey方法编写。
 *  睡眠时间 random.nextInt(2000) + 1000; 可以根据自己的业务时间情况进行修改。
 *  redis锁 代码没有粘贴，使用自己项目中的锁即可。如果项目中没有 网上也有很多。 RedisServiceUtils

#### 1.2.可以根据不同业务类型创建不同枚举（相当于mq名称） ####

import org.apache.commons.lang.StringUtils;
    
    /**
     * @program: wys-service
     * @description: redis队列名称列表
     * @author: wuyuanshn
     * @create: 2023-06-07 17:15
     **/
    public enum ListQueueNameEnum {
        
        TX_URL_BASE("URL_BASE", "信息"),
        ;
    
        private String code;
        private String value;
    
        ListQueueNameEnum(String code, String value) {
        
            this.code = code;
            this.value = value;
        }
    
        public String getValue() {
        
            return value;
        }
    
        public String getCode() {
        
            return code;
        }
    
        public void setCode(String code) {
        
            this.code = code;
        }
    
        public void setValue(String value) {
        
            this.value = value;
        }
    
        /**
         * 根据统计编号获取对应枚举实例
         *
         * @param value
         * @return
         */
        public static ListQueueNameEnum getStructureEnum(String value) {
        
            if (StringUtils.isBlank(value)) {
        
                return null;
            }
            for (ListQueueNameEnum result : ListQueueNameEnum.values()) {
        
                if (result.name().equals(value)) {
        
                    return result;
                }
            }
            return null;
        }
    
    }

#### 1.3.测试接口 ####

测试只需要调用consume接口 即可。

@Resource
        private ListRedisQueue listRedisQueue;
    
     @PostMapping(value = "/produce")
        public void produce() {
        
            for (int i = 0; i < 1; i++) {
        
                listRedisQueue.produce(ListQueueNameEnum.URL_BASE.getCode(), "xxxx", "1233", "345566");
            }
        }
    
    /** 测试只需要调用consume 此方法即可。
    */
        @PostMapping(value = "/consume")
        public void consume() {
        
          produce(); 
          log.info("生产消息完毕");
            listRedisQueue.consume(ListQueueNameEnum.URL_BASE.getCode());
        }

### 总结 ###

是使用redis代替mq功能，终究还是比较复杂，虽然实现了多项目间公用一个队列的需求。但是并不完美。如果可以还是尽量用合适的服务如mq等来实现业务。