java秒杀高并发------秒杀接口高并发秒杀优化 RabbitMQ模式

谁践踏了优雅 2022-05-25 05:29 330阅读 0赞

## RabbitMQ ##

我在windows平台下安装

参考  
[https://blog.csdn.net/hzw19920329/article/details/53156015][https_blog.csdn.net_hzw19920329_article_details_53156015]

集成RabbitMQ

要先安装 erlang，要依赖他

[http://www.erlang.org/downloads][http_www.erlang.org_downloads]

启动：  
![这里写图片描述][70]

安装了管理界面后  
rabbitmq-plugins enable rabbitmq\_management  
![这里写图片描述][70 1]

## SpringBoot集成 RabbitMQ ##

1.添加依赖 spring-boot-starter-amqp

<dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

添加配置

#rabbitmq
    spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    spring.rabbitmq.virtual-host=/
    #消费者数量
    spring.rabbitmq.listener.simple.concurrency= 10
    spring.rabbitmq.listener.simple.max-concurrency= 10
    #消费者每次从队列获取的消息数量
    spring.rabbitmq.listener.simple.prefetch= 1
    #消费者自动启动
    spring.rabbitmq.listener.simple.auto-startup=true
    #消费失败，自动重新入队
    spring.rabbitmq.listener.simple.default-requeue-rejected= true
    #启用发送重试
    spring.rabbitmq.template.retry.enabled=true
    spring.rabbitmq.template.retry.initial-interval=1000
    spring.rabbitmq.template.retry.max-attempts=3
    spring.rabbitmq.template.retry.max-interval=10000
    spring.rabbitmq.template.retry.multiplier=1.0

配置类

@Configuration
    public class MQConfig { 
        public static final String QUEUE ="queue";
    
        //队列
        @Bean
        public Queue queue(){
            return new Queue("queue",true);
        }

2.创建消息接受者

@Service
    public class MQReceiver { 
    
        private static Logger log = LoggerFactory.getLogger(MQReceiver.class);
            @RabbitListener(queues = MQConfig.QUEUE)
        public void receiver(String message){
            log.info("recevier"+message);
        }
    }

3.创建消息发送者

public class MQSender { 
        @Autowired
        AmqpTemplate amqpTemplate;
    
        @Autowired
        RedisService redisService;
          private static Logger log = LoggerFactory.getLogger(MQReceiver.class);
    
        public void send(Object message){
            //将bean转为字符串
            String msg = redisService.beanToString(message);
            //发送
            amqpTemplate.convertAndSend(MQConfig.QUEUE,msg);
            log.info("send:"+msg);
    
        }
    }

使用 guest默认是不允许远程连接的。

## RabbitMQ4种交换机模式 ##

消息是先发到交换机上，再由交换机发送到队列

SirectExchange 安装routingkey分发到指定队列

**Direct模式**

配置

//队列
    @Bean
    public Queue queue(){
        return new Queue(QUEUE,true);
    }

发送

public void send(Object message){
        //将bean转为字符串
        String msg = redisService.beanToString(message);
        //发送
        amqpTemplate.convertAndSend(MQConfig.QUEUE,msg);
        log.info("send:"+msg);
    
    }

接收

@RabbitListener(queues = MQConfig.QUEUE)
    public void receiver(String message){
        log.info("recevier"+message);
    }

**Topic模式**

多关键字匹配  
配置：

public static final String TOPIC_QUEUE1 ="topic.queue1";public static final String TOPIC_QUEUE2 ="topic.queue2";public static final String TOPIC_EXCHANGE="topicExchange";public static final String ROUTING_KEY1="topic.key1";public static final String ROUTING_KEY2="topic.#";//# 通配符//Topic模式 交换机 Exchange@Beanpublic Queue topicQueue1(){ return new Queue(TOPIC_QUEUE1,true);}@Beanpublic Queue topicQueue2(){ return new Queue(TOPIC_QUEUE2,true);}//topic excahnge@Beanpublic TopicExchange topicExchange(){ return new TopicExchange(TOPIC_EXCHANGE);}//绑定@Beanpublic Binding topicBinding1(){ return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with(ROUTING_KEY1);}@Beanpublic Binding topicBinding2(){ return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with(ROUTING_KEY2);}
    
    
    发送
    
    
    
    public void sendTpoic(Object message){
        //将bean转为字符串
        String msg = redisService.beanToString(message);
        //发送
        amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,MQConfig.ROUTING_KEY1,msg+":1");
        amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,MQConfig.ROUTING_KEY2,msg+":2");
        log.info("send topic:"+msg);
    
    }

接收

@RabbitListener(queues = MQConfig.TOPIC_QUEUE1)
    public void receiverTopic1(String message){
        log.info("recevier topic 1:"+message);
    }
    
    @RabbitListener(queues = MQConfig.TOPIC_QUEUE2)
    public void receiverTopic2(String message){
        log.info("recevier topic 2:"+message);
    }

![这里写图片描述][70 2]

**Fanout 广播模式**  
将消息分发到所有的绑定队列，无routingkey的概念  
还是绑定的之前两个主题模式创建的队列

配置

//Fanout模式 交换机Exchange（广播）
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange(FANOUT_EXCHANGE);
    }
    
    //绑定
    @Bean
    public Binding fanoutBinding1(){
        return BindingBuilder.bind(topicQueue1()).to(fanoutExchange());
    }
    @Bean
    public Binding fanoutBinding2(){
        return BindingBuilder.bind(topicQueue2()).to(fanoutExchange());
    }

发送

public void sendFanout(Object message){
        //将bean转为字符串
        String msg = redisService.beanToString(message);
        //发送
        amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE,"",msg);
        log.info("send fanout: :"+msg);
    
    }

接收  
还是对那两个主题的队列进行监听

![这里写图片描述][70 3]

## Headers ##

通过添加属性 key-value匹配

配置

//Hander模式
    
    
    @Bean
    public HeadersExchange headerstExchange(){
        return new HeadersExchange(HEADERS_EXCHANGE);
    }
    
    @Bean
    public Queue headerqueue(){
        return new Queue(HEADERS_QUEUE,true);
    }
    
    
    @Bean
    public Binding headersBinding(){
        Map<String,Object> map  = new HashMap<String, Object>();
        map.put("header1","value1");
        map.put("header2","value2");
        //满足key-value才会往队列中放东西
        return BindingBuilder.bind(headerqueue()).to(headerstExchange()).whereAll(map).match();
    }

发送

public void sendHeader(Object message){
        //将bean转为字符串
        String msg = redisService.beanToString(message);
        //发送
        MessageProperties properties = new MessageProperties();
        properties.setHeader("header1","value1");
        properties.setHeader("header2","value2");
        Message obj = new Message(msg.getBytes(),properties);
        amqpTemplate.convertAndSend(MQConfig.HEADERS_EXCHANGE,"",obj);
        log.info("send header: :"+msg);
    
    }

接收

@RabbitListener(queues = MQConfig.HEADERS_QUEUE)
    public void receiverHeader(byte[] message){
        log.info("recevier header :"+new String (message));
    }

![这里写图片描述][70 4]

## 使用RabbitMQ改写秒杀功能 ##

思路：减少数据库访问

1.系统初始化，把商品库存数量加载到Redis
    
    2.收到请求，Redis预减库存，库存不足，直接返回，否则3
    
    3.请求入队，立即返回排队中
    
    4.请求出队，生成订单，减少库存
    
    5.客户端轮询，是否秒杀成功

**系统初始化，把商品库存数量加载到Redis**

如何在初始化的时候就将库存数据存入缓存中

实现这个借口 InitializingBean

public class MiaoshaController implements InitializingBean{

然后  
实现方法

//在系统初始化的时候调用
    @Override
    public void afterPropertiesSet() throws Exception {
        List<GoodsVo> goodsVoList =  goodsService.listGoodsVo();
        if (goodsVoList==null){
            return ;
             }
        for (GoodsVo goodsVo : goodsVoList){
            //将商品库存加载到redis中 
              redisService.set(GoodsKey.getMiaoshaGoodsStock,""+goodsVo.getId(),goodsVo.getStockCount());
        }
    }

## 收到请求，Redis预减库存，库存不足，直接返回，否则3 请求入队，立即返回排队中 ##

//减少库存
    long stock = redisService.decr(GoodsKey.getMiaoshaGoodsStock,""+goodsId);
    if (stock<0){
       //如果没有库存了就返回秒杀失败了。
        return Result.error(CodeMsg.MIAO_SHA_OVER);
    }
    //判断是否秒杀到了。
    MiaoshaOrder order  = orderService.getMiaoshaOrderByUserIdGoodsId(user.getId(),goodsId);
        if(order!=null) {
       //已经秒杀到了
            return Result.error(CodeMsg.REPEATE_MIAOSHA);
        }
    //入队
    MiaoshaMessage mm = new MiaoshaMessage();
    mm.setGoodsId(goodsId);
    mm.setUser(user);
    mqSender.sendMiaoshaMessage(mm);
    return  Result.success(0);//排队中
    }

## 请求出队，生成订单，减少库存 ##

接收

@RabbitListener(queues = MQConfig.MIAOSHA_QUEUE)
        public void receiver(String message) {
            log.info("recevier" + message);
            //将消息还原为bean对象
            MiaoshaMessage mm = redisService.stringToBean(message, MiaoshaMessage.class);
            MiaoshaUser user = mm.getUser();
            long goodsId = mm.getGoodsId();
            //减库存，订单
            GoodsVo goods = goodsService.getGoodsVoByGoodsId(goodsId);
            int stock = goods.getGoodsStock();
            if (stock <= 0) {
       //没有库存了.
                return;
            }
                  //判断是否重复秒杀
            MiaoshaOrder order = orderService.getMiaoshaOrderByUserIdGoodsId(user.getId(), goodsId);
            if (order != null) {
                return;
            } else {
                //秒杀
                //减库存 下订单 写入秒杀订单
                 miaoshaService.miaosha(user, goods);
            }
    
        }

## 客户端轮询，是否秒杀成功 ##

定义个接口

//秒杀排队后轮询
        //秒杀成功返回订单id
        //秒杀失败返回 -1
        //排队中 返回 0
        @RequestMapping("/result")
        public Result miaoshResult(Model model, MiaoshaUser user, @RequestParam("goodsId") long goodsId) {
            //没登录就跳转到登录页面
            model.addAttribute("user", user);
            if (user == null) {
                return Result.error(CodeMsg.SESSION_ERROR);
            }
    //查询下是否生成了订单
            long rsult = miaoshaService.getMiaoshResult(user.getId(), goodsId);
            return Result.success(rsult);
    
    
        }

miaoshaService.getMiaoshResult

public long getMiaoshResult(Long userId, long goodsId) {
        MiaoshaOrder order = orderService.getMiaoshaOrderByUserIdGoodsId(userId,goodsId);
        if (order!=null){
            return order.getId();
        }else {
            boolean isOver = getGoodsOver(goodsId);
            if (isOver){
       //卖完了还在没有订单
                return -1;
            }else {
                return 0;
            }
    
        }

没有订单有两种情况，卖完了失败，和排队中

在上面的秒杀那做个标记。这个商品是否秒杀完了。存入redis中。  
之后去判断是否存在这个key就知道是哪种情况

@Transactional
    public OrderInfo miaosha(MiaoshaUser user, GoodsVo goods) {
        //减少库存 下订单 写入秒杀订单
        boolean success = goodsServic.reduceStock(goods);
        //减库存成功了才进行下订单
        if (success) {
            //订单表，秒杀表
            OrderInfo orderInfo = orderService.createOrder(user, goods);
            return orderInfo;
        } else {
       //说明商品秒杀完了。做一个标记
            setGoodsOver(goods.getId());
            return null;
        }
    
    }

public void setGoodsOver(Long goodsId) {
        redisService.set(MiaoshaKey.isGoodsOver,""+goodsId,true);
    
    
    }
    
    private boolean getGoodsOver(long goodsId) {
        return redisService.exists(MiaoshaKey.isGoodsOver,""+goodsId);
    }

前端请求做判断。

[https_blog.csdn.net_hzw19920329_article_details_53156015]: https://blog.csdn.net/hzw19920329/article/details/53156015
[http_www.erlang.org_downloads]: http://www.erlang.org/downloads
[70]: /images/20220525/c85b0fb9e6fc4aeeae5fd092dafc0459.png
[70 1]: /images/20220525/a57f6cb9c6b14112b109c9156d4f9a5f.png
[70 2]: /images/20220525/9540f02da1f544b0853125ce5fe894ed.png
[70 3]: /images/20220525/a5c9bea543ab419f88129771a6d7b693.png
[70 4]: /images/20220525/439ae7eca5754a0389d3d998431daaab.png