分析下单时，应用的模型如下：
访问数据库的次数如下：（访问6次数据库；是否能对访问数据库的次数进行一定程序的优化，采用缓存的方式）
原来对用户，商品进行校验的方式:都需要访问数据库
1.校验下单状态,下单的商品是否存在，用户是否合法，购买数量是否正确 ItemModel itemModel = itemService.getItemById(itemId); if(itemModel == null){ throw new BusinessException(EmBusinessError.PARAMETER_VALIDATION_ERROR,"商品信息不存在"); }可以将商品的详情以及用户的详情存入到redis缓存中，之后用户再次下单的时候可以从缓存中取出数据
问题是：这里的商品信息若在之后用户下过单之后，可能进行更改就存在一个redis缓存与数据库缓存不一致的情况。
应用
//使用缓存校验itemModel ItemModel itemModel = itemService.getItemByIdInCache(itemId); if(itemModel == null){ throw new BusinessException(EmBusinessError.PARAMETER_VALIDATION_ERROR,"商品信息不存在"); } UserModel userModel = userService.getUserByIdInCache(userId); if(userModel == null){ throw new BusinessException(EmBusinessError.PARAMETER_VALIDATION_ERROR,"用户信息不存在"); }ItemServiceImpl
@Override public ItemModel getItemByIdInCache(Integer id) { ItemModel itemModel = ( ItemModel)redisTemplate.opsForValue().get("item_validate_"+id); if(itemModel==null){ itemModel = this.getItemById(id); redisTemplate.opsForValue().set("item_validate_"+id,itemModel); //设置失效时间 redisTemplate.expire("item_validate_"+id,10, TimeUnit.MINUTES); } return itemModel; }UserServiceImpl
@Override public UserModel getUserByIdInCache(Integer id) { UserModel userModel = (UserModel)redisTemplate.opsForValue().get("user_validate_"+id); if(userModel==null){ userModel=this.getUserById(id); redisTemplate.opsForValue().set("user_validate_"+id,userModel); redisTemplate.expire("user_validate_"+id,10, TimeUnit.MINUTES); } return userModel; }在访问数据库进行库存扣减的问题时，由于数据库行锁的存在，可能存在数据库瓶颈的问题
可以将数据库中的stock缓存到缓存中，库存缓存化，之后再采用消息队列将其异步执行更新mysql数据库中的内容；
代码
活动发布同步库存进缓存
/活动函发布 void publishPromo(Integer promoId); @Override public void publishPromo(Integer promoId) { //通过活动Id获取活动 PromoDO promoDO = promoDOMapper.selectByPrimaryKey(promoId); if(promoDO.getItemId()==null||promoDO.getItemId().intValue()==0){ return; } ItemModel itemModel = itemService.getItemById(promoDO.getItemId()); //将库存同步到redis内 redisTemplate.opsForValue().set("promo_item_stock_"+itemModel.getId(),itemModel.getStock()); }交易下单减掉缓存库存
//2.落单减库存 boolean result = itemService.decreaseStock(itemId,amount); //减去缓存中的库存 long result = redisTemplate.opsForValue().increment("promo_item_stock_"+itemId,amount.intValue()*-1);异步消息扣减数据库内存库
代码 生产者
package com.imooc.miaoshaproject.mq; import com.alibaba.fastjson.JSON; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.exception.RemotingException; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import java.nio.charset.Charset; import java.util.HashMap; import java.util.Map; /** * Created by hzllb on 2019/2/23. */ @Component public class MqProducer { private DefaultMQProducer producer; @Value("${mq.nameserver.addr}") private String nameAddr; @Value("${mq.topicname}") private String topicName; @PostConstruct public void init() throws MQClientException { //做mq producer的初始化 producer = new DefaultMQProducer("producer_group"); producer.setNamesrvAddr(nameAddr); producer.start(); } //同步库存扣减消息 public boolean asyncReduceStock(Integer itemId,Integer amount) { Map<String,Object> bodyMap = new HashMap<>(); bodyMap.put("itemId",itemId); bodyMap.put("amount",amount); Message message = new Message(topicName,"increase", JSON.toJSON(bodyMap).toString().getBytes(Charset.forName("UTF-8"))); try { producer.send(message); } catch (MQClientException e) { e.printStackTrace(); return false; } catch (RemotingException e) { e.printStackTrace(); return false; } catch (MQBrokerException e) { e.printStackTrace(); return false; } catch (InterruptedException e) { e.printStackTrace(); return false; } return true; } }消息消费者:数据库中库存的扣减操作
package com.imooc.miaoshaproject.mq; import com.alibaba.fastjson.JSON; import com.imooc.miaoshaproject.dao.ItemDOMapper; import com.imooc.miaoshaproject.dao.ItemStockDOMapper; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageExt; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import java.util.List; import java.util.Map; /** * Created by hzllb on 2019/2/23. */ @Component public class MqConsumer { private DefaultMQPushConsumer consumer; @Value("${mq.nameserver.addr}") private String nameAddr; @Value("${mq.topicname}") private String topicName; @Autowired private ItemStockDOMapper itemStockDOMapper; @PostConstruct public void init() throws MQClientException { consumer = new DefaultMQPushConsumer("stock_consumer_group"); consumer.setNamesrvAddr(nameAddr); consumer.subscribe(topicName,"*"); consumer.registerMessageListener(new MessageListenerConcurrently() { @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { // //实现库存真正到数据库内扣减的逻辑 Message msg = msgs.get(0); String jsonString = new String(msg.getBody()); Map<String,Object>map = JSON.parseObject(jsonString, Map.class); Integer itemId = (Integer) map.get("itemId"); Integer amount = (Integer) map.get("amount"); itemStockDOMapper.decreaseStock(itemId,amount); return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } }); consumer.start(); } }在缓存中加入一个标志，进行判断：当缓存中的库存小于0时，则在缓存中设置一个标志，每次用户下单的时候都需要先判断这个标志是否存在，若标志存在，则说明没有库存，直接返回不存在
//防止商品售罄 //判断商品是否已经售罄，若对应的售罄key存在则直接返回下单失败 if(redisTemplate.hasKey("promo_item_stock_invalid_"+itemId)){ throw new BusinessException(EmBusinessError.STOCK_NOT_ENOUGH); }mq异步事务
if(!mqProducer.transactionAsyncReduceStock(userModel.getId(),promoId,itemId,amount,stockLogId)){ throw new BusinessException(EmBusinessError.UNKNOWN_ERROR,"下单失败"); }mqproducer
//事务型异步库存扣减消息 public boolean transactionAsyncReduceStock(Integer userId,Integer promoId,Integer itemId,Integer amount,String stockLogId){ Map<String,Object> bodyMap = new HashMap<>(); bodyMap.put("itemId",itemId); bodyMap.put("amount",amount); bodyMap.put("stockLogId",stockLogId); Map<String,Object> argsMap = new HashMap<>(); argsMap.put("itemId",itemId); argsMap.put("amount",amount); argsMap.put("userId",userId); argsMap.put("promoId",promoId); argsMap.put("stockLogId",stockLogId); Message message = new Message(topicName,"increase" , JSON.toJSON(bodyMap).toString().getBytes(Charset.forName("UTF-8"))); TransactionSendResult transactionSendResult=null; try { //发送的是事务型消息 //发送完成之后事务处于prepare状态 //1、往消息队列中投递prepare消息，维护在message broker的中间件上 //2、在本地执行localTRranscatopn transactionSendResult = transactionMQProducer.sendMessageInTransaction(message,argsMap); } catch (MQClientException e) { e.printStackTrace(); return false; } if(transactionSendResult.getLocalTransactionState()==LocalTransactionState.ROLLBACK_MESSAGE){ return false; }else if(transactionSendResult.getLocalTransactionState()==LocalTransactionState.COMMIT_MESSAGE){ return true; }else{ return false; } } @PostConstruct public void init() throws MQClientException { //做mq producer的初始化 producer = new DefaultMQProducer("producer_group"); producer.setNamesrvAddr(nameAddr); producer.start(); transactionMQProducer = new TransactionMQProducer("tran_producer_group"); transactionMQProducer.setNamesrvAddr(nameAddr); transactionMQProducer.start(); transactionMQProducer.setTransactionListener(new TransactionListener() { /** * COMMIT_MESSAGE, ROLLBACK_MESSAGE, 将之前prepare消息发送 UNKNOW; * @param message * @param arg * @return */ @Override public LocalTransactionState executeLocalTransaction(Message message, Object arg) { //真正要做的事，创建订单 Integer itemId = (Integer) ((Map)arg).get("itemId"); Integer promoId = (Integer) ((Map)arg).get("promoId"); Integer userId = (Integer) ((Map)arg).get("userId"); Integer amount = (Integer) ((Map)arg).get("amount"); String stockLogId = (String) ((Map)arg).get("stockLogId"); try { //在这里真正调用createOrder方法，完成订单的创建以及对应redis orderService.createOrder(userId, itemId,promoId,amount,stockLogId); } catch (BusinessException e) { e.printStackTrace(); //发生异常，事务回滚 return LocalTransactionState.ROLLBACK_MESSAGE; } //只有当COMMIT完成之后，才发生后消息的回调 return LocalTransactionState.COMMIT_MESSAGE; } @Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { //根据是否扣减库存成功来判断要返回COMMIT,ROLLBACK还是继续UNKWON //实现库存真正到数据库内扣减的逻辑 String jsonString = new String(msg.getBody()); Map<String,Object>map = JSON.parseObject(jsonString, Map.class); Integer itemId = (Integer) map.get("itemId"); Integer amount = (Integer) map.get("amount"); //查询流水 String stockLogId = (String) map.get("stockLogId"); StockLogDO stockLogDO = stockLogDOMapper.selectByPrimaryKey(stockLogId); if(stockLogDO==null) return LocalTransactionState.UNKNOW; if(stockLogDO.getStatus().intValue()==2){ return LocalTransactionState.COMMIT_MESSAGE; }else if(stockLogDO.getStatus().intValue()==1){ return LocalTransactionState.UNKNOW; } return LocalTransactionState.ROLLBACK_MESSAGE; } }); }创建订单creatOrder
@Override @Transactional public OrderModel createOrder(Integer userId, Integer itemId, Integer promoId, Integer amount,String stockLogId) throws BusinessException { // 1.校验下单状态,下单的商品是否存在，用户是否合法，购买数量是否正确 // ItemModel itemModel = itemService.getItemById(itemId); // if(itemModel == null){ // throw new BusinessException(EmBusinessError.PARAMETER_VALIDATION_ERROR,"商品信息不存在"); // } //使用缓存校验itemModel ItemModel itemModel = itemService.getItemByIdInCache(itemId); if(itemModel == null){ throw new BusinessException(EmBusinessError.PARAMETER_VALIDATION_ERROR,"商品信息不存在"); } //缓存 UserModel userModel = userService.getUserByIdInCache(userId); if(userModel == null){ throw new BusinessException(EmBusinessError.PARAMETER_VALIDATION_ERROR,"用户信息不存在"); } if(amount <= 0 || amount > 99){ throw new BusinessException(EmBusinessError.PARAMETER_VALIDATION_ERROR,"数量信息不正确"); } //校验活动信息 if(promoId != null){ //（1）校验对应活动是否存在这个适用商品 if(promoId.intValue() != itemModel.getPromoModel().getId()){ throw new BusinessException(EmBusinessError.PARAMETER_VALIDATION_ERROR,"活动信息不正确"); //（2）校验活动是否正在进行中 }else if(itemModel.getPromoModel().getStatus().intValue() != 2) { throw new BusinessException(EmBusinessError.PARAMETER_VALIDATION_ERROR,"活动信息还未开始"); } } //2.落单减库存：这里扣减的是缓存中的库存（只有当这些操作全部成功时，事务型的消息才会被消费者销毁） boolean result = itemService.decreaseStock(itemId,amount); if(!result){ throw new BusinessException(EmBusinessError.STOCK_NOT_ENOUGH); } //3.订单入库：可以将其插入刀片 OrderModel orderModel = new OrderModel(); orderModel.setUserId(userId); orderModel.setItemId(itemId); orderModel.setAmount(amount); if(promoId != null){ orderModel.setItemPrice(itemModel.getPromoModel().getPromoItemPrice()); }else{ orderModel.setItemPrice(itemModel.getPrice()); } orderModel.setPromoId(promoId); orderModel.setOrderPrice(orderModel.getItemPrice().multiply(new BigDecimal(amount))); //生成交易流水号,订单号 //这里都可以进行改进，将其发送到队列中， orderModel.setId(generateOrderNo()); OrderDO orderDO = convertFromOrderModel(orderModel); orderDOMapper.insertSelective(orderDO); //加上商品的销量 itemService.increaseSales(itemId,amount); //设置库存流水状态位成功 StockLogDO stockLogDO = stockLogDOMapper.selectByPrimaryKey(stockLogId); if(stockLogDO==null){ throw new BusinessException(EmBusinessError.UNKNOWN_ERROR); }else{ stockLogDO.setStatus(2); stockLogDOMapper.updateByPrimaryKeySelective(stockLogDO); } //4.返回前端 return orderModel; }