Redis应用(一) --消息队列

Spring-boot 集成Redis应用(一) --消息队列

一.基础环境

jdk 1.8
maven 3.5.3
spring-boot 2.0.4
redis 4.0.11

二.基本介绍

Spring MVC 3.2 之后引入了基于Servlet 3的异步请求处理。因此使用了DeferredResult 相关使用来实现异步处理，从而扩大请求吞吐量。

Redis使用 LPUSH 和RPOP命令实现队列的概念。只需要让生产者将任务使用LPUSH 命令加入到某个键中，另一边让消费者不断地使用RPOP命令从该键中取出任务即可。

三.流程介绍

流程图

四.相关代码示例

POM文件

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
 <groupId>org.apache.commons</groupId>
 <artifactId>commons-pool2</artifactId>
 <version>2.0</version>
</dependency>

application.properties 简单相关redis配置

注意如果redis没有安装到本机，那么必须设置访问密码否则会有连接报错异常。

#Redis server host.
spring.redis.host=192.168.56.101
# Redis server port.
spring.redis.port=6379
#Login password of the redis server.
spring.redis.password=123456
# Maximum number of connections that can be allocated
# by the pool at a given time. Use a negative value for no limit.
spring.redis.lettuce.pool.max-active=8
# Maximum number of "idle" connections in the pool.
# Use a negative value to indicate an unlimited number of idle connections.
spring.redis.lettuce.pool.max-idle=8
# Maximum amount of time a connection allocation should
# block before throwing an exception when the pool is exhausted.
# Use a negative value to block indefinitely.
spring.redis.lettuce.pool.max-wait=-1ms
# Target for the minimum number of idle connections to maintain in the pool. T
# his setting only has an effect if it is positive.
spring.redis.lettuce.pool.min-idle=0
# Shutdown timeout.
spring.redis.lettuce.shutdown-timeout=0ms

自定义StringRedisTemplate 类,保证采用string的序列序列化,具体配置在代码中有原文链接，如果有疑问可以看看。

@Configuration
public class RedisConfig {
    /**
     *  定义 StringRedisTemplate ，指定序列化和反序列化的处理类
     * @param redisConnectionFactory
     * @return
     */
    @Bean
    public RedisTemplate<String,String> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        StringRedisTemplate stringRedisTemplate = new StringRedisTemplate(redisConnectionFactory);
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        //配置过滤类型
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        //默认允许序列化类型
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        //序列化 值时使用此序列化方法
        stringRedisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        stringRedisTemplate.afterPropertiesSet();
        return stringRedisTemplate;
    }

编写DeferredResult 的管理类，方便在异步方法中找到任务ID对应的DeferredResult对象。

@Component
public class DeferredResultHolder {

    private Map<String, DeferredResult<String>> map = new HashMap<String,DeferredResult<String>>();

    public Map<String, DeferredResult<String>> getMap() {
        return map;
    }

    public void setMap(Map<String, DeferredResult<String>> map) {
        this.map = map;
    }
}

编写消息队列监听,监听处理任务程序的消费队列,如果队列里有值，则代表任务处理完成，并找到对应的DeferredResult 对象，进行赋值结果返回。

/**
 * 设置消息队列监听
 * ContextRefreshedEvent spring 在初始化完毕后的事件
 * @author Neal
 */
@Component
public class QueueListener implements ApplicationListener<ContextRefreshedEvent> {

    //日志
    private Logger logger = LoggerFactory.getLogger(getClass());

    //redis完成队列KEY
    private static String REDIS_COMPLATE = "complete";


    @Autowired
    private RedisTemplate redisTemplate;

    //DeferredResult管理类
    @Autowired
    private DeferredResultHolder deferredResultHolder;



    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent){
        //由于处理队列方法是一个无限循环，需要单起一个线程，防止阻塞系统启动
        new Thread(()->{
            while(true) {
                logger.info("读取消息队列完成订单 ");
                //从完成的队列中按顺序取出完成的任务ID
                Object uuid = redisTemplate.opsForList().rightPop(REDIS_COMPLATE,5000,TimeUnit.SECONDS);
                //为空判断
                if(null == uuid) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(500);
                        continue;
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                logger.info("返回订单处理结果: " + uuid);
                //返回处理结果
                deferredResultHolder.getMap().get(uuid).setResult("success");
            }
        }).start();
    }
}

编写模拟任务处理方法 ,在DEMO中同样使用监听机制来实现,当从生产队列中取到非NULL值的任务ID时，则进行1秒钟的休眠，方便后期调试查看。然后将该ID放入消费队列。

/**
 * 模拟另一个程序去处理消息队列里的任务
 * ContextRefreshedEvent spring 在初始化完毕后的事件
 * @author Neal
 */
@Component
public class ResolveListener implements ApplicationListener<ContextRefreshedEvent> {


    private Logger logger = LoggerFactory.getLogger(getClass());

    //redis 完成队列 KEY
    private static String REDIS_COMPLATE = "complete";

    //redis 准备队列 KEY
    private static String REDIS_MESSAGE = "prepare";

    @Autowired
    private RedisTemplate redisTemplate;


    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        //由于处理队列方法是一个无限循环，需要单起一个线程，防止阻塞系统启动
        new Thread(()-> {
            while(true) {
                //获取任务队列中的任务ID
                Object prepareduuid = redisTemplate.opsForList().rightPop(REDIS_MESSAGE, 5000, TimeUnit.SECONDS);
                //非空判断
                if(null == prepareduuid) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(500);
                        continue;
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                logger.info("读取消息队列待处理ID ; " + prepareduuid);
                /**
                 * 模拟任务处理过程 begin
                 */
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                /**
                 * end
                 */
                logger.info("完成订单处理，把处理ID放入完成队列");
                //将完成后的任务放入 任务结束队列
                redisTemplate.opsForList().leftPush(REDIS_COMPLATE, prepareduuid);
            }
        }).start();

    }
}

接着写一下测试的controller层，就是简单的请求。

@RestController
@RequestMapping("/redis")
public class RedisController {

    private Logger logger = LoggerFactory.getLogger(getClass());
    //redis 准备队列
    private static String REDIS_MESSAGE = "prepare";

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private DeferredResultHolder deferredResultHolder;

    /**
     * 测试消息队列入口接口
     * @return
     */
    @GetMapping("/async")
    public DeferredResult<String> async() {
        logger.info("主线程开始");
        //生成唯一值 模拟任务ID初始化
        String uuid = UUID.randomUUID().toString();
        //将要任务的ID放入redis 待处理任务消息队列
        redisTemplate.opsForList().leftPush(REDIS_MESSAGE,uuid);

        DeferredResult<String> deferredResult = new DeferredResult<>();

        //将任务ID和 DeferredResult 对象绑定
        deferredResultHolder.getMap().put(uuid,deferredResult);

        logger.info("主线程返回");

        return deferredResult;

    }
}

启动spring-boot。应用postman发送GET请求调试。

postman

控制台

从postman请求以及控制台输出日志可以看到,postman发起请求是先请求主线程，然后主线程调用结束，紧接着是模拟的任务处理ResolveListener线程进行处理操作，最后监听队列QueueListener线程监听到任务处理结束并返回结果给前台，打印出success。

 因为步骤5和6都是在启动监听，所以单独启用一个线程，防止方法内的WHILE循环阻塞容器启动。

 在此DEMO中，如果程序运行时间长会报出 Redis command timed out; nested exception is io.lettuce.core.RedisCommandTimeoutException: Command timed out 异常。具体原因就是连接REDIS时间设置过短，修改相关配置即可。

以上的消息队列思路是通过学习慕课网JoJozhai老师的 spring security相关课程写的DEMO

DEMO代码地址