java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > SpringBoot Redis不重复消费队列

SpringBoot+Redis实现不重复消费的队列的示例代码

作者：雷袭月启

本文主要介绍了SpringBoot+Redis实现不重复消费的队列的示例代码,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

背景

最近我们新研发了一个“年夜饭订购”功能（没想到吧，雷袭在是一个程序猿的同时，也是一名优秀的在厨子）。用户使用系统选择年夜饭，点击“下单”时，后台首先会生成一条订单数据，返回消息给用户：“您已成功下单，后厨正在准备菜品！”。同时，以线程的方式指挥各个厨子按菜单联系供应商准备食材，制作菜品，最后打包寄给客户。但是，用户在使用这个功能时，系统却有一定的机率卡死，这个问题极大的影响了用户的体验。年关将近，这个功能也显得越发重要，客户要求我们限期整改，三天内必须解决该问题。

我首先对这个功能进行了分析，很明显，这是一个使用频次不高，但是使用时间比较集中的功能。在大量用户同时使用时，会导致后台的厨师，食材，供应商等全面告警（用程序员语言翻译一下，这个功能耗CPU，耗内存，耗IO）。但用户对于实时性的要求并不高。下单之后，订购的菜品是一天内完成，还是两天完成并没有关系，只要年前能做完就可以。

因此，我们决定采用消息中间件的方式，以队列的形式逐次的执行“年夜饭制作”的操作， 来缓解服务器的各种资源的压力。

之所以采用Redis来实现消息队列，而不是使用更为成熟的ONS，Kafka。不是因为ONS用不起，而是Redis更有性价比（用户只允许使用ONS中间件，但ONS会带来额外的网络开销，学习成本和风险都更大，这个功能使用频度并不高，没有必要为了它而引入一个重量级的中间件。）

代码实践

说干就干，咱们先看看源码，如下：

// 订单实体类
@Data
public class OrderEntity implements Serializable {

    /**
     * 客户姓名
     */
    private String customerName;

    /**
     * 订单号
     */
    private String orderCode;

    /**
     * 菜单
     */
    List<String> menus;
}

@Slf4j
@Service
public class DinnerService {
    
    /**
     * 年夜饭下单
     *
     * @param req 订单信息
     * @return
     */
    public Object orderNewYearEveDinner(OrderEntity entity) {
        // 存储订单信息
        saveOrder(entity);
        // 异步开始做菜
        CompletableFuture.runAsync(() -> doNewYearEveDinner(entity));
        return "您已成功下单，后厨正在准备预制菜!";
    }

    /**
     * 这里模拟的是做年夜饭的过程方法，该方法用时较长，整个过程需要10秒。
     * 这个过程中存在多种意外，可能导致该方法执行失败
     *
     * @param req 订单信息
     */
    public void doNewYearEveDinner(OrderEntity entity) {
        System.out.println("开始做订单 " + entity.getOrderCode() + " 的年夜饭");
        try {
            Thread.sleep(10000);
        }catch (Exception e ) {
            e.printStackTrace();
            System.out.println("厨子跑了，厨房着火了，供应商堵路上了");
        }
        System.out.println("订单 " + entity.getOrderCode() + " 的年夜饭已经完成");
    }
    
    private void saveOrder(OrderEntity req) {
        //这里假设做的是订单入库操作
        System.out.println("订单 " + req.getOrderCode() + " 已经入库， 做饭开始时间为 "+ new Date());
    }

}

1、引入maven依赖，在application.yml中添加redis配置

 <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
spring:
  redis:
    database: 9
    host: 127.0.0.1
    port: 6379
    password: 
    jedis:
      pool:
        max-active: 8
        max-wait: -1
        max-idle: 8
        min-idle: 0

2、添加Redis队列监听，添加Redis配置文件注册监听

// 监听类
@Component
public class DinnerListener implements MessageListener {

    @Autowired
    private DinnerService service;

    @Override
    public void onMessage(Message message, byte[] pattern)  {
        OrderEntity entity= JSON.parseObject(message.toString(), OrderEntity.class);
        service.doNewYearEveDinner(entity);
    }
}


//配置类，用于注册监听
@Configuration
public class RedisConfig {
    @Bean
    public ChannelTopic topic() {
        return new ChannelTopic("NEW_YEAR_DINNER");
    }

    @Bean
    public MessageListenerAdapter messageListenerAdapter(DinnerListener listener) {
        return new MessageListenerAdapter(listener);
    }

    @Bean
    public RedisMessageListenerContainer redisContainer(RedisConnectionFactory redisConnectionFactory,
                                                        MessageListenerAdapter messageListenerAdapter,
                                                        ChannelTopic topic) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        container.addMessageListener(messageListenerAdapter, topic);
        return container;
    }

}

3、修改原方法，以及Controller调用

// DinnerService中的方法修改   
 /**
     * 年夜饭下单
     *
     * @param req 订单信息
     * @return
     */
    public Object orderNewYearEveDinner(OrderEntity entity) {
        // 存储订单信息
        saveOrder(entity);
        // 异步开始做菜
        redisTemplate.convertAndSend("NEW_YEAR_DINNER", JSON.toJSONString(entity));
        return "您已成功下单，后厨正在准备预制菜!";
    }


@RestController
public class DinnerController {

    private int i = 0;
    
    @Autowired
    private DinnerService service;

    @GetMapping("/orderDinner")
    public Object orderDinner() {
        OrderEntity entity = new OrderEntity();
        entity.setOrderCode("Order" + (++i));
        entity.setCustomerName("第"+i+"位客户");
        return service.orderNewYearEveDinner(entity);
    }
}

4、通过postman调用四次请求，测试结果如下：

5、Listener中添加同步锁

细看上文中打出来的注释，我发现这和我设想的不一样啊。原定的计划是先做完第一份年夜饭，再做第二份，做完第二份再做第三份，为什么第一次没执行完就开始执行第二次了？

在网上查了些资料后我才知道，要达到我想要的效果，得在Listener中添加上同步锁，如下：

@Component
public class DinnerListener implements MessageListener {

    @Autowired
    private DinnerService service;

    private final Object lock = new Object();

    @Override
    public void onMessage(Message message, byte[] pattern)  {
        synchronized (lock) {
            OrderEntity entity = JSON.parseObject(message.toString(), OrderEntity.class);
            service.doNewYearEveDinner(entity);
        }
    }
}

再次执行测试用例，结果如下：

6、多服务不重复消费消息

上面的结果已经满足了我们的要求，但是，客户考虑到我们只有一个厨房，的确影响效率，决定给我们扩建一个厨房（添加服务器），希望能达到厨房A做第一份订单，厨房B做第二份订单，以上的代码能实现吗？我们把刚才的项目拷贝一份，修改端口，启动后测试。结果如下：

从上面的日志可以看出来，两个服务都做了订单1的年夜饭，消息被重复消费了。但是根据业务需求，我们不需要重复消费消息，我们想达到的效果是多服务实现负载均衡，本服务在处理的数据，其他服务不需要再处理了，应该怎么实现呢？咱们依然可以运用Redis，对代码做如下调整：

@Component
public class DinnerListener implements MessageListener {

    @Autowired
    private DinnerService service;
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    private final Object lock = new Object();

    @Override
    public void onMessage(Message message, byte[] pattern)  {
        synchronized (lock) {
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(message.toString(), "1", 1, TimeUnit.DAYS);
            // 加锁失败，已有消费端在此时对此消息进行处理，这里不再做处理
            if (!flag) {
                return;
            }
            OrderEntity entity = JSON.parseObject(message.toString(), OrderEntity.class);
            service.doNewYearEveDinner(entity);
        }
    }
}

从测试结果来看，这么调整解决达到了我们的效果。

7、添加日志监控

仔细检查，发现上面的代码虽然满足了我们的业务需求，但是在安全方面仍然没有得到一定的保障，方法doNewYearEveDinner存在很多不可预见的隐患，如厨师跑了，厨房着了，供应商堵路上了，这些都会导致方法执行失败，那么，我们怎么知道这个订单执行成功或者失败了呢？看日志吗？成百上千条数据堆起来，通过看日志来看结果多不方便啊？咱们是否可以对代码做一下调整？基于这方面考虑，我对代码做了以下调整

//订单类进行调整
@Data
public class OrderEntity implements Serializable {

    /**
     * 客户姓名
     */
    private String customerName;

    /**
     * 订单号
     */
    private String orderCode;

    /**
     * 菜单
     */
    List<String> menus;

    /**
     * 出餐状态
     */
    private String dinnerState;
    
    /**
     * 做饭开始时间
     */
    private String dinnerStartTime;

    /**
     * 做饭结束时间
     */
    private String dinnerEndTime;

    /**
     * 备注
     */
    private String remark;
}

// DinnerService做如下调整， 添加一个订单信息更新的方法
@Slf4j
@Service
public class DinnerService {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 年夜饭下单
     *
     * @param req 订单信息
     * @return
     */
    public Object orderNewYearEveDinner(OrderEntity req) {
        // 存储订单信息
        saveOrder(req);
        // 异步开始做菜
        redisTemplate.convertAndSend("NEW_YEAR_DINNER", JSON.toJSONString(req));
        return "您已成功下单，订单号为"+ req.getOrderCode()+"，后厨正在准备预制菜!";
    }
    /**
     * 这里模拟的是做年夜饭的过程方法，该方法用时较长，整个过程需要10秒，但是，这个过程中存在多种意外，该方法可能失败
     *
     * @param req 订单信息
     */
    public void doNewYearEveDinner(OrderEntity req) throws Exception {
        System.out.println("开始做订单 " + req.getOrderCode() + " 的年夜饭");
        Thread.sleep(10000);
        System.out.println("订单 " + req.getOrderCode() + " 的年夜饭已经完成");
    }

    private void saveOrder(OrderEntity req) {
        //这里假设做的是订单入库操作
        System.out.println("订单 " + req.getOrderCode() + " 已经入库， 做饭开始时间为 "+ new Date());
    }

    /**
     * 根据订单编号修改订单信息
     *
     * @param orderCode 订单编号
     * @param dinnerStatus
     * @param remark
     */
    public void updateOrder(String orderCode, String dinnerStatus, String remark) {
        // 根据订单编号修改订单的出餐结束时间，出餐状态，备注等信息。
        System.out.println("更新订单 "+ orderCode +" 信息，做饭结束时间为 "+ new Date() + ", 出餐状态为"+ dinnerStatus +", 备注为 " +remark);
    }
}

// Listener中做如下调整
    @Override
    public void onMessage(Message message, byte[] pattern)  {
        synchronized (lock) {
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(message.toString(), "1", 1, TimeUnit.DAYS);
            // 加锁失败，已有消费端在此时对此消息进行处理，这里不再做处理
            if (!flag) {
                return;
            }
            OrderEntity param = JSON.parseObject(message.toString(), OrderEntity.class);
            try {
                service.doNewYearEveDinner(param);
                service.updateOrder(param.getOrderCode(), "SUCCESS", "成功");
            }catch (Exception e) {
                e.printStackTrace();
                service.updateOrder(param.getOrderCode(), "FAIL", e.getMessage());
            }
        }
    }

这部分代码就不贴测试结果了，与上一次的测试结果一致，只不过提升了功能的可测试性，扩展一下，这个结果能否达到我们的要求呢？其实仍然没有，对于执行失败的订单，我们需要一个机制来处理，根据报错信息决定是重新执行还是直接报警，人为介入处理，由此才能实现整个事务的闭环。

这是一次简单的SpringBoot+Redis实现队列的实践，个人觉得这个过程比较有趣，分析问题出现的原因，需求的潜在归约，根据业务的需要、当前的条件选择合适的方法和组件，快而有效的解决问题，所以我将它记录了下来，供大家参考。实际上，已经有大神对于Redis实现队列的方法进行了完整细致的归纳，如果想深入的了解这部分的知识，推荐你们看看这篇博客： Redis队列详解（springboot实战）

到此这篇关于SpringBoot+Redis实现不重复消费的队列的示例代码的文章就介绍到这了,更多相关SpringBoot Redis不重复消费队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文