RabbitMQ+Redis集群+Quartz实现简单高并发秒杀

小鱼儿 2022-06-05 12:25 424阅读 0赞

花了两天时间实现了一个使用rabbitMQ队列和redis集群存取数据以及使用Quartz触发添加秒杀商品。  
这一块小功能很早就想做的，自从自学了redis的命令，发现了expire能够设置自动消亡的时候，我就已经开始蠢蠢欲动了，接着在接触rabbitMQ工作模式（多个消费者争抢数据）的时候，我已经下决心要实现秒杀了。  
上个项目是9月底和朋友做完的，一个高并发分布式的项目，开6台centOS虚拟机搭建nginx、主从服务器、redis集群，rabbitMQ队列，amoeba实现读写分离和主从数据库，以及Solr搜索。这个项目是用来练手linux与分布式的，大多数精力都花在搭环境上了，基本步骤也都能百度到，不想写到博客。正好这个秒杀的功能不多不少，思路还有点意思，所以写一下与大家分享。

**秒杀的设计理念：**

**限流**： 鉴于只有少部分用户能够秒杀成功，所以要限制大部分流量，只允许少部分流量进入服务后端。前台页面控制

**削峰**：对于秒杀系统瞬时会有大量用户涌入，所以在抢购一开始会有很高的瞬间峰值。高峰值流量是压垮系统很重要的原因，所以如何把瞬间的高流量变成一段时间平稳的流量也是设计秒杀系统很重要的思路。实现削峰的常用的方法有利用缓存和消息中间件(RabbitMQ)等技术。

**异步处理**：秒杀系统是一个高并发系统，采用异步处理模式可以极大地提高系统并发量，其实异步处理就是削峰的一种实现方式。(RabbitMQ实现)

**内存缓存**：秒杀系统最大的瓶颈一般都是数据库读写，由于数据库读写属于磁盘IO，性能很低，如果能够把部分数据或业务逻辑转移到内存缓存，效率会有极大地提升。

**我的第一次尝试：**  
**纯粹使用一台redis实现秒杀，是有同步安全问题的**  
因为redis是支持高并发的，一秒可以承受10000次的请求，所以暂且使用一台redis试试效果，毕竟单台redis是单线程，并发安全问题会少一点。

首先创建秒杀商品表，习惯使用PowerDesigner画表格，因为只是简单的一个Demo，只需要id, title, price, num, KillTime 五个属性，分别指代商品id，商品标题，商品价格，秒杀商品的数量，以及秒杀开始的时间。

这个我是模仿淘宝的整点抢购，每一个小时扫描一次秒杀商品表，将商品按抢购时间发布出来，将id作为key，num作为value写入redis，并设置消亡时间为1s（为了测试方便设了5秒）。

当用户点击抢购按钮，首先在前端进行控制，如果时间还没到整点前后两分钟的区间，直接在前端拦截（没写），else才发送请求，使用ajax与restful方式发送请求的url，根据接收的参数，反馈不一样的信息。

后台收到请求之后，首先根据id从redis中get对应的num，如果为null，返回”notbegin”，判断num>0则decr自减，返回true，否则返回finished，如果catch到了错误，返回false。  
前端function：

Quartz的配置：

<bean name="secKillJobDetail" class="org.springframework.scheduling.quartz.JobDetailFactoryBean">
            
            <property name="jobClass" value="com.jt.manage.job.SecKillJob" />
            
            <property name="name" value="SecKillJob" />
            
            <property name="group" value="SecKillJob" />
            
            <property name="durability" value="true"/>
            
            <property name="applicationContextJobDataKey" value="applicationContext"/>
        </bean>
    
        
        <bean class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
            <property name="triggers">
                <list>
                    <ref bean="cronTrigger" />
                </list>
            </property>
        </bean>
    
        
        <bean id="cronTrigger" class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
            <property name="jobDetail" ref="secKillJobDetail" />
            
            <property name="cronExpression" value="0/30 * * * * ?" />
        </bean>

Quartz按时执行的job

Controller层：

@Controller
    @RequestMapping("/SecKill")
    public class SecKillController { 
    
        @Autowired
        private SecKillItemService secKillItemService;
        @Autowired
        private RabbitSecKillService rabbitSecKillService;
    
        @RequestMapping("/list")
        public String listSecKillItem(Model model) {
            List<SecKillItem> secKillItemList = secKillItemService.findSecKillItemList();
            model.addAttribute("secKillItemList", secKillItemList);
            return "seckillitem-list";
        }
    
        @RequestMapping("/startKill/{id}")
        @ResponseBody
        public String startKill(@PathVariable Long id){
            String result = secKillItemService.getSecKillResult(id);
            System.out.println("result:"+result);
            return result;
    
        }
    
    }

Service层

@Service
    public class SecKillItemServiceImpl implements SecKillItemService { 
        @Autowired
        private JedisCluster jedisCluster;
        @Autowired
        private SecKillItemMapper secKillItemMapper;
    
        private static Connection connection = null;
    
        @Override
        public List<SecKillItem> findSecKillItemList() {
            SecKillItem secKillItem = new SecKillItem();
            List<SecKillItem> secKillItemList = secKillItemMapper.select(secKillItem);
            return secKillItemList;
        }
    
        @Override
        public String getSecKillResult(Long id) {
            try {
                System.out.println("id:"+id);
                System.out.println("剩余存活时间"+jedisCluster.ttl(id+""));
                if(jedisCluster.get(id+"")==null){
                    System.out.println("活动还没开始");
                    return "notbegin";
                }
                String num = jedisCluster.get(id+"");
                Integer Num = Integer.parseInt(num);
                System.out.println("商品"+id+"当前有"+Num+"个");
                if(Num>0){
                    Num-=1;
                    jedisCluster.decr(id+"");
                    System.out.println("商品"+id+"剩余"+jedisCluster.get(id+"")+"个");
                    return "true";
                }else{
                    return "finished";
                }
            } catch (Exception e) {
                return "false";
            }
        }
    }

高并发测试我使用的是Apache的jmeter  
[http://jmeter.apache.org/][http_jmeter.apache.org]

![这里写图片描述][SouthEast]  
![这里写图片描述][SouthEast 1]

并发100次发现有3次为true，出现了安全问题，分析了一下，是因为service存在if判断。  
百度了下redis有自带的getset方法可以进行同步操作，不过只用getset，还是要判断返回的数量，这个不太靠谱。

**我的第二次尝试：**  
**使用rabbitMQ的阻塞队列和redis集群来实现**  
因为rabbitMQ的排队入库，真的是很好的削峰技术，我觉得是可行的。

这里有一个点：当rabbitMQ用于数据库削峰的时候，在配置文件中配置了接收数据的方法，这个方法默认打开且被动接受；  
然而秒杀的接收数据方是用户，当用户主动点击秒杀时，用户才从队列中获取数据，所以不能使用配置文件。。。可能是rabbitmq没有这样一个倒过来的处理逻辑，那我只能手写队列的生产者和消费者了。

而且鉴于第一次尝试中的get 和 set 操作同时存在导致redis不安全的问题，这次不从redis中get数据进行判断，将redis只作为数据的提供者。

还有一个点：单台redis是单线程的，是安全的，只要你正确使用它的线程安全的方法；但是redis集群是多线程的，是不安全的，即使有hash一致性算法，即使redis集群可以整体操作，但是底层还是单独的redis在工作，所以事务也是不支持的。

这次我每次初始化provider的时候，只在队列中存储一个token，用户点击后consumer方法和provider共用一个connection，注意不能使用ThreadLocal，因为他们是不同的线程的数据，会get到null。自己手写一个普通的静态类来维护connection就好，因为一件商品只需要在一个队列中存取

新job：

public class SecKillJob extends QuartzJobBean{ 
        private static Connection connection = null;
    
        @Override
        protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
            System.out.println("开始加入缓存");
            ApplicationContext applicationContext = (ApplicationContext) context.getJobDetail().getJobDataMap().get("applicationContext");
            SecKillItemMapper mapper = applicationContext.getBean(SecKillItemMapper.class);
            JedisCluster jedisCluster = applicationContext.getBean(JedisCluster.class);
            SecKillItem secKillItem = new SecKillItem();
            //只秒杀第一件商品
            secKillItem.setId(1L);
            List<SecKillItem> list = mapper.select(secKillItem);
            for (SecKillItem item : list) {
                //纯使用redis存取数据，使用num，get和set会有线程安全问题
                /** * 使用rabbitMQ+redis，在redis中存入(key,value) id--token,token根据商品id计算， * 并将其存入rabbitMQ，存入的数量表示可以被抢购的数量 */
                String id = item.getId().toString();
                String token = id+"-token";
                jedisCluster.set(id, token);
                jedisCluster.expire(id, 15);
                //每种商品只添加一件
                initConnection();
                provider(token);
            }
            System.out.println("加入缓存与Rabbitmq成功");
        }
    
        public static void initConnection(){
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.196.137");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("/SecKillHost");
            connectionFactory.setUsername("SecKillAdmin");
            connectionFactory.setPassword("626316");
    
            try {
                connection = connectionFactory.newConnection();
                //ConnThreadLocal.setConn(connection);
                Conn.setConn(connection);
                System.out.println("这是SecKillJob的initConnection(), Connection==null??"+(connection==null));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        public static void provider(String token){
            try {
                //1.只有通过通道才能连接rabbitMQ
                Channel channel = connection.createChannel();
                //每次存入redis前先删除这个队列
                channel.queueDelete("SecKillQueue");
                //2.定义队列的名称
                String queue = "SecKillQueue";
                //3.声明队列
                channel.queueDeclare(queue, false, false, false, null);
                channel.basicPublish("", queue, null, token.getBytes());
                System.out.println("一件商品已存入rabbitMq，token="+token);
                channel = connection.createChannel();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

新service

public class SecKillItemServiceImpl implements SecKillItemService { 
        @Autowired
        private JedisCluster jedisCluster;
        @Autowired
        private SecKillItemMapper secKillItemMapper;
    
        private static Connection connection = null;
    
        @Override
        public List<SecKillItem> findSecKillItemList() {
            SecKillItem secKillItem = new SecKillItem();
            List<SecKillItem> secKillItemList = secKillItemMapper.select(secKillItem);
            return secKillItemList;
        }
    
    
        public String getSecKillResult(Long id){
            try {
                String token = consumer();
                Long tokenId = Long.parseLong(token.substring(0, token.indexOf("-")));
                if(id.equals(tokenId)){
                    return "true";
                }else{
                    System.out.println("活动还没开始/已经抢购完毕");
                    return "notbegin";
                }
            } catch (Exception e) {
                e.printStackTrace();
                return "false";
            }
        }
        public static String consumer() throws Exception{
            //connection=ConnThreadLocal.getConn();
            connection = Conn.getConn();
            System.out.println("这是SecKillServiceImpl 的 consumer() , 从ConnThreadLocal获取的Connection==null??"+(connection==null));
            Channel channel = connection.createChannel();
            String queue = "SecKillQueue";
            channel.queueDeclare(queue, false, false, false, null);
            channel.basicQos(1);
            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(queue, false, consumer);//当开启autoAsk时，若处理出错，则生产者还会将此消息给别的消费者使用，所以，要关掉自动应答
            System.out.println("准备好接收了");
    
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String token = new String(delivery.getBody());
            System.out.println("接收到的token"+token);
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            return token;
        }
    
    }

用来维护connection的Conn类

public class Conn {
        private static Connection connection;
        public static void setConn(Connection connection){
            Conn.connection = connection;
        }
        public static Connection getConn(){
            return Conn.connection;
        }
    
    }

rabbitMQ的工作模式：

![这里写图片描述][SouthEast 2]  
原理说明:  
生产者为消息队列中生产消息,多个消费者争抢执行权利,谁抢到谁执行.

**初步实现的效果：**  
rabbitMQ后台：  
![rabbitMQ后台][rabbitMQ]

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

并发一百个，的确只有一个获取到了

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

有一个问题是：除第一个显示抢购成功，其余的在设置的15秒内是没有反应的，因为一直在等待rabbitMQ的数据，直到我销毁queue。在实际业务中，可以采用页面动画来缓解用户情绪，比如大转盘之类的，先转个十几秒哈哈。

还有，可以在前台直接产生随机数，比如100以内的随机数，只有随机数为1的可以访问后台，其余直接抢购失败，也是实际业务中可能使用的

**其实还有分布式锁的解决办法，不难，而且可以不使用队列**

不过暂时不研究了，因为我要开始找实习了，在实践中发掘更有用的知识。快12月了，校招都没了，不好找啊。。。。  
祝自己架构师之路顺利^\_^

[http_jmeter.apache.org]: http://jmeter.apache.org/
[SouthEast]: /images/20220605/9f96960e9dc94b45ae8ad864dfd0400e.png
[SouthEast 1]: /images/20220605/9c4b40b7df644b78a8b200c77fd798ca.png
[SouthEast 2]: /images/20220605/db6302c2b17343ac868493298439ae2e.png
[rabbitMQ]: /images/20220605/1c6b688b302c4d568ba772da21f05704.png
[SouthEast 3]: /images/20220605/5b98319cc9644b8a99c9f14fb0b7009d.png
[SouthEast 4]: /images/20220605/a0288d553acc42c5b4c0c61c79d658ec.png