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: 02、添加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不重复消费队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
这篇文章主要介绍了java交换排序之鸡尾酒排序实现方法,实例分析了排序的原理与相关的实现技巧,需要的朋友可以参考下2015-02-02
本文主要介绍了maven中更改jdk版本的方法实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧2024-07-07
这篇文章主要介绍了java list集合排序按某一属性排序操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-08-08
这篇文章主要为大家介绍了分布式面试难题，ElasticSearch解决大数据量检索的问题分析回答，让面试官无话可说，帮助大家实现面试开薪自由2022-03-03
这篇文章主要介绍了Java for循环Map集合优化实现解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2020-01-01
springboot整合shardingsphere和seata实现分布式事务的实践
本文主要介绍了springboot整合shardingsphere和seata实现分布式事务的实践，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2022-07-07
这篇文章主要为大家详细介绍了java实现五子棋大战，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2022-03-03
springboot统一返回json数据格式并配置系统异常拦截方式
这篇文章主要介绍了springboot统一返回json数据格式并配置系统异常拦截方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2024-08-08
打包的方式有打包成jar包或者打包成war包发布，区别在于jar包内置了tomcat、netty等服务器，更改只需要修改pom.xml的坐标即可，war不内置服务器，需要上传到服务器tomcat解压后运行，本文分析Springboot打包成jar发布，感兴趣的朋友一起看看吧2023-02-02
idea显示springboot多服务启动界面service操作
这篇文章主要介绍了idea显示springboot多服务启动界面service操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2020-09-09
最新评论