SSM框架学习之高并发秒杀业务--笔记5-- 并发优化

时间：2022-03-22 14:17:11

前几节终于实现了这个高并发秒杀业务，现在问题是如何优化这个业务使其能扛住一定程度的并发量。

一. 优化分析

　　对于整个业务来说，首先是分析哪些地方会出现高并发，以及哪些地方会影响到了业务的性能。可能会出现高并发的地方：详情页，获取系统时间，地址暴露接口，执行秒杀操作。

　这个业务为什么要单独获取时间呢？用户会在详情页大量刷新，为了优化这里，将detal.jsp详情页和一些静态资源(css,js等）部署在CDN的节点上（至于这个CDN是什么，下面会说），也就是说用户访问详情页是不需要访问我们的系统的，这样就降低了服务器的负荷，但这个时候就拿不到系统的时间了，所以要单独做一个请求来获取当前的系统时间。

那么什么是CDN呢，content distribute network 内容分发网络，本质上是一种加速用户获取数据的系统，把一些用户频繁访问的静态资源部署在离用户最近的网络节点上，关于CDN的具体解释可以见这篇博文：http://blog.csdn.net/coolmeme/article/details/9468743。对于获取系统时间这个操作，因为java访问一次内存大约10ns，所以不需要优化。

对于秒杀地址接口，因为是经常变化的，所以不适合部署在CDN上，要部署在我们服务器的系统上，这里要用到服务器端缓存如：redis缓存服务器来进行优化。

redis缓存服务器：Redis是一个开源的使用ANSI C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库，并提供多种语言的API。关于Redis，他是一个内存数据库，即将硬盘上的部分数据缓存在内存中。对内度的读取的速度要远快于对硬盘的读取，记得我们数据库老师以前和我们说过，对于数据库的设计而言，优化的最核心的部分是如何减少对磁盘的IO操作，因为磁盘的IO是其实就是硬盘的磁头在读磁片上的磁道，这是个机械运动，速度要远慢于内存的读写。关于redis的一些知识，还要深入学习才行。

对于秒杀操作，这个是整个业务最核心的东西，不可能部署在CDN上，也不能使用redis缓存服务器，因为不可能在缓存中取减库存，要在mysql中操作，否则会产生数据不一致的情况。老师也说了一些其他的优化方案，不过我听不懂就是了，什么原子计数器，分布式MQ，消费消息并落地之类的。貌似和分布式系统有关？不明白啊，还得好好去学，先知道有这个东西先。

对于并发程序来说，拖慢速度的关键是事务控制，涉及到数据库中的行级锁，优化方向是：如何减少行级锁的持有时间。那么优化思路是：将客户端逻辑放到MySql服务端，同时避免网络延迟和GC(垃圾回收)的影响。具体说就是把在客户端中的事务控制放在MySql服务端。具体方式就是使用存储过程，使整个事务在到MySql端完成。什么是存储过程：在大型数据库系统中，一组为了完成特定功能的SQL 语句集，存储在数据库中，经过第一次编译后再次调用不需要再次编译，用户通过指定存储过程的名字并给出参数（如果该存储过程带有参数）来执行它。具体见：存储过程简介。

二. 具体优化

1.redis后端缓存优化编码

redis的下载和安装，以及如何使用Redis的官方首选Java开发包Jedis：Windows下Redis的安装使用。

在dao包下新建cache目录，新建RedisDao类，用于访问我们的redis。

RedisDao.java

 package org.seckill.dao.cache;

 import com.dyuproject.protostuff.LinkedBuffer;

 import com.dyuproject.protostuff.ProtobufIOUtil;

 import com.dyuproject.protostuff.ProtostuffIOUtil;

 import com.dyuproject.protostuff.runtime.RuntimeSchema;

 import org.seckill.entity.Seckill;

 import org.slf4j.Logger;

 import org.slf4j.LoggerFactory;

 import redis.clients.jedis.Jedis;

 import redis.clients.jedis.JedisPool;

 /**

  * Created by yuxue on 2016/10/22.

  */

 public class RedisDao {

     private final Logger logger= LoggerFactory.getLogger(this.getClass());

     private final JedisPool jedisPool;

     private RuntimeSchema<Seckill> schema=RuntimeSchema.createFrom(Seckill.class);

     public RedisDao(String ip, int port ){

         jedisPool=new JedisPool(ip,port);

     }

     public Seckill getSeckill(long seckillId) {

         //redis操作逻辑

         try{

             Jedis jedis=jedisPool.getResource();

             try {

                 String key="seckill:"+seckillId;

                 //并没有实现序列化机制

                 //get->byte[]->反序列化->Object(Seckill)

                 //采用自定义序列化

                 //protostuff : pojo.

                 byte[] bytes=jedis.get(key.getBytes());

                 //缓存获取到

                 if(bytes!=null){

                     //空对象

                     Seckill seckill=schema.newMessage();

                     ProtostuffIOUtil.mergeFrom(bytes,seckill,schema);

                     //seckill被反序列化

                     return seckill;

                 }

             }finally {

                 jedis.close();

             }

         }catch (Exception e){

             logger.error(e.getMessage(),e);

         }

         return null;

     }

     public String putSeckill(Seckill seckill){

         // set Object(Seckill) -> 序列化 ->发送给redis

         try{

             Jedis jedis=jedisPool.getResource();

             try{

                 String key="seckill:"+seckill.getSeckillId();

                 byte[] bytes=ProtostuffIOUtil.toByteArray(seckill,schema,

                         LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE));

                 //超时缓存

                 int timeout=60*60;//1小时

                 String result=jedis.setex(key.getBytes(),timeout,bytes);

                 return result;

             }finally{

                 jedis.close();

             }

         }catch (Exception e){

             logger.error(e.getMessage(),e);

         }

         return null;

     }

 }

这里有个优化点是：redis并没有实现对象的序列化，需要我们自己手动去序列化对象，当然这里可以让对象实现Serializable接口，也就是用jdk提供的对象序列化机制。但是这里为了优化这个目的，我们需要一个速度更快得序列化机制，所以老师这里用的是基于谷歌Protobuff的ProtoStuff序列化机制。

ProtoStuff的依赖

<!--protostuff序列化依赖-->

      <dependency>

          <groupId>com.dyuproject.protostuff</groupId>

          <artifactId>protostuff-core</artifactId>

          <version>1.0.8</version>

      </dependency>

      <dependency>

          <groupId>com.dyuproject.protostuff</groupId>

          <artifactId>protostuff-runtime</artifactId>

          <version>1.0.8</version>

      </dependency>

在spring-dao.xml配置RedisDao

 <!--RedisDao-->

     <bean id="redisDao" class="org.seckill.dao.cache.RedisDao">

         <constructor-arg index="0" value="localhost"/>

         <constructor-arg index="1" value="6379"/>

     </bean>

修改SeckillServiceImpl.java为

 package org.seckill.service.impl;

 import org.apache.commons.collections.MapUtils;

 import org.seckill.dao.SeckillDao;

 import org.seckill.dao.SuccesskilledDao;

 import org.seckill.dao.cache.RedisDao;

 import org.seckill.dto.Exposer;

 import org.seckill.dto.SeckillExecution;

 import org.seckill.entity.Seckill;

 import org.seckill.entity.SuccessKilled;

 import org.seckill.enums.SeckillStatEnum;

 import org.seckill.exception.RepeatKillException;

 import org.seckill.exception.SeckillCloseException;

 import org.seckill.exception.SeckillException;

 import org.seckill.service.SeckillService;

 import org.slf4j.Logger;

 import org.slf4j.LoggerFactory;

 import org.springframework.beans.factory.annotation.Autowired;

 import org.springframework.stereotype.Service;

 import org.springframework.transaction.annotation.Transactional;

 import org.springframework.util.DigestUtils;

 import java.util.Date;

 import java.util.HashMap;

 import java.util.List;

 import java.util.Map;

 /**

  * Created by yuxue on 2016/10/15.

  */

 @Service

 public class SeckillServiceImpl implements SeckillService {

     private Logger logger = LoggerFactory.getLogger(this.getClass());

     @Autowired

     private SeckillDao seckillDao;

     @Autowired

     private SuccesskilledDao successkilledDao;

     @Autowired

     private RedisDao redisDao;

     //md5盐值字符串，用于混淆MD5

     private final String salt = "fsladfjsdklf2jh34orth43hth43lth3";

     public List<Seckill> getSeckillList() {

         return seckillDao.queryAll(0, 4);

     }

     public Seckill getById(long seckillId) {

         return seckillDao.queryById(seckillId);

     }

     public Exposer exportSeckillUrl(long seckillId) {

         //优化点：缓存优化，超时的基础上维护一致性

         //1.访问redis

         Seckill seckill = redisDao.getSeckill(seckillId);

         if (seckill == null) {

             //2.若缓存中没有则访问数据库

             seckill = seckillDao.queryById(seckillId);

             if (seckill == null) {

                 return new Exposer(false, seckillId);

             } else {

                 //3.放入redis

                 redisDao.putSeckill(seckill);

             }

         }

         Date startTime = seckill.getStartTime();

         Date endTime = seckill.getEndTime();

         Date nowTime = new Date();

         if (nowTime.getTime() < startTime.getTime() ||

                 nowTime.getTime() > endTime.getTime()) {

             return new Exposer(false, seckillId, nowTime.getTime(), startTime.getTime(), endTime.getTime());

         }

         //转化特定字符串的过程，不可逆

         String md5 = getMD5(seckillId);

         return new Exposer(true, md5, seckillId);

     }

     private String getMD5(long seckillId) {

         String base = seckillId + "/" + salt;

         String md5 = DigestUtils.md5DigestAsHex(base.getBytes());

         return md5;

     }

     @Transactional

     /*

     * 使用注解控制事务方法的优点：

     * 1：开发团队一致的约定

     * 2：保证事务方法的执行时间经可能的短，不要穿插其他网络操作RPC/HTTP请求或者剥离到事务方法外部，使得

     *    这个事务方法是个比较干净的对数据库的操作

     * 3：不是所有的方法都需要事务，如只有一条修改操作，只读操作不需要事务控制

     * */

     public SeckillExecution executeSeckill(long seckillId, long userPhone, String md5) throws SeckillException,

             RepeatKillException, SeckillCloseException {

         if (md5 == null || !md5.equals(getMD5(seckillId))) {

             throw new SeckillException("seckill data rewrite");

         }

         //执行秒杀逻辑：减库存+记录购买行为

         Date nowTime = new Date();

         try {

             //记录购买行为

             int insertCount = successkilledDao.insertSucessSeckilled(seckillId, userPhone);

             //唯一：seckillId,userPhone

             if (insertCount <= 0) {

                 //重复秒杀

                 throw new RepeatKillException("seckill repeated");

             } else {

                 //减库存,热点商品竞争

                 int updateCount = seckillDao.reduceNumber(seckillId, nowTime);

                 if (updateCount <= 0) {

                     //没有更新到记录，秒杀结束，rollback

                     throw new SeckillCloseException("seckill is close");

                 } else {

                     //秒杀成功,commit

                     SuccessKilled successKilled = successkilledDao.queryByIdWithSeckill(seckillId, userPhone);

                     return new SeckillExecution(seckillId, SeckillStatEnum.SUCCESS, successKilled);

                 }

             }

         } catch (SeckillCloseException e1) {

             throw e1;

         } catch (RepeatKillException e2) {

             throw e2;

         } catch (Exception e) {

             logger.error(e.getMessage());

             throw new SeckillException("seckill inner error" + e.getMessage());

         }

     }

     public SeckillExecution executeSeckillProdure(long seckillId, long userPhone, String md5){

         if (md5 == null || !md5.equals(getMD5(seckillId))) {

             return new SeckillExecution(seckillId, SeckillStatEnum.DATA_REWRITE);

         }

         Date killTime=new Date();

         Map<String,Object> map=new HashMap<String, Object>();

         map.put("seckillId",seckillId);

         map.put("phone",userPhone);

         map.put("killTime",killTime);

         map.put("result",null);

         try{

             seckillDao.killByProcedure(map);

             //获取result

             int result= MapUtils.getInteger(map,"result",-2);

             if(result==1){

                 SuccessKilled sk=successkilledDao.

                         queryByIdWithSeckill(seckillId,userPhone);

                 return new SeckillExecution(seckillId,SeckillStatEnum.SUCCESS,sk);

             }else{

                 return new SeckillExecution(seckillId,SeckillStatEnum.stateOf(result));

             }

         }catch (Exception e){

             logger.error(e.getMessage(),e);

             return new SeckillExecution(seckillId,SeckillStatEnum.INNER_ERROR);

         }

     }

 }

对于暴露秒杀接口exportSeckillUrl这个方法，原本是直接从数据库中取Seckill对象的，现在优化为先在Redis缓存服务器中取，如果没有则去数据库中取，并将其放入Redis缓存中。这里还有个优化点就是在执行秒杀executeSeckill方法中

将insert操作放到了update之前。

2. 利用存储过程

对于现在的update操作，还是在客户端控制事务的，为了进一步优化，现在将update的操作逻辑放在Mysql端来执行，也就是利用存储过程来完成商品更新操作，减少行级锁的持有时间。

在src/main/sql目录下新建seckill.sql, 编写存储过程

 -- 秒杀执行存储过程

 DELIMITER $$ -- console ; 转换为 $$

 -- 定义存储过程

 -- 参数：in 输入参数；out 输出参数

 -- row_count():返回上一条修改类型sql(delete, insert,update)的影响行数

 -- row_count: 0:未修改；>0:表示修改的行数；<0:sql错误/未执行

 CREATE PROCEDURE `seckill`.`execute_seckill`

   (in v_seckill_id bigint, in v_phone bigint,

     in v_kill_time timestamp,out r_result int)

   BEGIN

     DECLARE insert_count int DEFAULT 0;

     START TRANSACTION ;

     insert ignore into seccess_killed

     (seckill_id,user_phone,create_time)

       values (v_seckill_id,v_phone,v_kill_time);

     select row_count() into insert_count;

     IF (insert_count=0) THEN

       ROLLBACK ;

       set r_result=-1;

     ELSEIF (insert_count<0) THEN

       ROLLBACK ;

       set r_result=-2;

     ELSE

       update seckill

       set number=number-1

       where seckill_id=v_seckill_id

         and end_time>v_kill_time

         and start_time<v_kill_time

         and number>0;

       select row_count() into insert_count;

       IF (insert_count=0) THEN

         ROLLBACK ;

         set r_result=0;

       ELSEIF (insert_count<0) THEN

         ROLLBACK ;

         set r_result=-2;

       ELSE

         COMMIT;

         set r_result=1;

       END IF;

     END IF;

   END;

 $$

 -- 存储过程定义结束

 DELIMITER ;

 --

 set @r_result=-3;

 -- 执行存储过程

 call execute_seckill(1004,13225534035,now(),@r_result);

 -- 获取结果

 select @r_result;

 -- 存储过程

 -- 1：存储过程优化：事务行级锁持有时间

 -- 2：不要过度依赖存储过程

 -- 3：简单的逻辑可以应用存储过程

 -- 4：QPS:一个秒杀单6000/qps

在Service层和dao层分别定义调用存储过程的接口，然后在Mybatis中配置调用存储过程

 package org.seckill.service;

 import org.seckill.dto.Exposer;

 import org.seckill.dto.SeckillExecution;

 import org.seckill.entity.Seckill;

 import org.seckill.exception.RepeatKillException;

 import org.seckill.exception.SeckillCloseException;

 import org.seckill.exception.SeckillException;

 import java.util.List;

 /**

  * 业务接口:站在"使用者"角度设计接口

  * 三个方面：方法定一粒度，参数，返回类型/异常

  * Created by yuxue on 2016/10/15.

  */

 public interface SeckillService {

     /**

      * 查询所有秒杀记录

      * @return

      */

     List<Seckill> getSeckillList( );

     /**

      * 查询单个秒杀记录

      * @param seckillId

      * @return

      */

     Seckill getById(long seckillId);

     /**

      * 秒杀开启时输出秒杀接口地址

      * 否则输出系统时间和秒杀时间

      * @param seckillId

      */

     Exposer exportSeckillUrl(long seckillId);

     /**

      * 执行秒杀操作

      * @param seckillId

      * @param userPhone

      * @param md5

      */

     SeckillExecution executeSeckill(long seckillId, long userPhone, String md5)

         throws SeckillException,RepeatKillException,SeckillCloseException;

     /**

      * 执行秒杀操作by 存储过程

      * @param seckillId

      * @param userPhone

      * @param md5

      */

     SeckillExecution executeSeckillProdure(long seckillId, long userPhone, String md5);

 }
 package org.seckill.dao;

 import org.apache.ibatis.annotations.Param;

 import org.seckill.entity.Seckill;

 import java.util.Date;

 import java.util.List;

 import java.util.Map;

 /**

  * Created by yuxue on 2016/10/12.

  */

 public interface SeckillDao {

     /**

      * 减库存

      * @param seckillId

      * @param killTime

      * @return 如果影响的行数大于1，表示更新记录行数

      */

     int reduceNumber(@Param("seckillId") long seckillId,@Param("killTime") Date killTime);

     /**

      * 根据id查询秒杀对象

      * @param seckillId

      * @return

      */

     Seckill queryById(long seckillId);

     /**

      * 根据偏移量查询秒杀商品列表

      * @param offset

      * @param limit

      * @return

      */

     List<Seckill> queryAll(@Param("offset") int offset, @Param("limit") int limit);

     /**

      * 使用存储过程执行秒杀

      * @param paramMap

      */

     void killByProcedure(Map<String,Object> paramMap);

 }

下面的要点便是如何在Mybatis中配置killByProcedure这个接口，存储过程的调用本质上是Mybatis在调用它，那么就得配置配置才行

 <?xml version="1.0" encoding="UTF-8" ?>

 <!DOCTYPE mapper

         PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"

         "http://mybatis.org/dtd/mybatis-3-mapper.dtd">

 <mapper namespace="org.seckill.dao.SeckillDao">

     <select id="queryById" resultType="Seckill" parameterType="long">

         select seckill_id,name,number,start_time, end_time, create_time

         from seckill

         where seckill_id=#{seckillId}

     </select>

     <update id="reduceNumber">

         update

           seckill

         set

           number=number -1

         where seckill_id = #{seckillId}

         and start_time <![CDATA[ <= ]]>   #{killTime}

         and end_time >= #{killTime}

         and number > 0;

     </update>

     <select id="queryAll" resultType="Seckill">

         select seckill_id,name,number,start_time,end_time,create_time

         from seckill

         order by create_time DESC

         limit #{offset},#{limit}

     </select>

     <!--mybatis调用存储过程-->

     <select id="killByProcedure" statementType="CALLABLE">

         call execute_seckill（

         #｛seckillId,jdbcType=BIGINT,mode=IN｝,

         #｛phone,jdbcType=BIGINT,mode=IN｝,

         #｛killTime,jdbcType=TIMESTAMP,mode=IN｝,

         #｛result,jdbcType=INTEGER ,mode=OUT｝

         ）

     </select>

 </mapper>

这里要记住配置这个存储过程的一些参数。

<!--mybatis调用存储过程-->

33     <select id="killByProcedure" statementType="CALLABLE">

34         call execute_seckill（

35         #｛seckillId,jdbcType=BIGINT,mode=IN｝,

36         #｛phone,jdbcType=BIGINT,mode=IN｝,

37         #｛killTime,jdbcType=TIMESTAMP,mode=IN｝,

38         #｛result,jdbcType=INTEGER ,mode=OUT｝

39         ）

40     </select>

SeckillServiceImpl里调用存储过程来执行秒杀的方法executeSeckillProdure具体实现如下：

 public SeckillExecution executeSeckillProdure(long seckillId, long userPhone, String md5){

         if (md5 == null || !md5.equals(getMD5(seckillId))) {

             return new SeckillExecution(seckillId, SeckillStatEnum.DATA_REWRITE);

         }

         Date killTime=new Date();

         Map<String,Object> map=new HashMap<String, Object>();

         map.put("seckillId",seckillId);

         map.put("phone",userPhone);

         map.put("killTime",killTime);

         map.put("result",null);

         try{

             seckillDao.killByProcedure(map);

             //获取result

             int result= MapUtils.getInteger(map,"result",-2);

             if(result==1){

                 SuccessKilled sk=successkilledDao.

                         queryByIdWithSeckill(seckillId,userPhone);

                 return new SeckillExecution(seckillId,SeckillStatEnum.SUCCESS,sk);

             }else{

                 return new SeckillExecution(seckillId,SeckillStatEnum.stateOf(result));

             }

         }catch (Exception e){

             logger.error(e.getMessage(),e);

             return new SeckillExecution(seckillId,SeckillStatEnum.INNER_ERROR);

         }

     }

至此，优化篇就写完了，写得略微粗糙了点，好多细节都没有具体分析，主要是快考试了，也没什么时间写博客了，忙死，估计这篇写完后就好好复习准备期末考试了吧。至此，终于写完了整个项目的过程，伴随着自己的一些理解，以后还要修改修改。结尾老师总结了下这个课程涉及到的知识点又介绍了下一般网站的系统部署架构，什么Nginx，Jetty，rabbitmq之类的。。。不得不感叹技术世界真是深似海，越学就会觉得自己不会的东西越多，不管怎样，慢慢来吧，在技术的道路上前进着！！！

相关文章