前言

本篇将完成高并发优化，包括：

• Redis后端缓存优化
• 并发优化

一、高并发优化分析

在优化之前要明白高并发发生在哪

高并发位置分析

红色的部分是可能出现高并发的点，绿色部分则没有影响

当用户进入详情页的是时候，如果秒杀没有开始，在页面就会显示倒计时，如果开始了，就会显示执行秒杀操作的按钮，接着是执行秒杀操作，最后是返回结果

1、获取系统时间

在详情页，当秒杀还没开始的时候，用户会不断的刷新页面，这是很有可能的，这就会造成大量的请求，详情页实际应该部署到CDN节点上，CDN会把详情页做静态化处理，详情页当中还有许多获取静态资源的请求，这些静态资源也会部署到CDN上

CDN的全称是Content Delivery Network，即内容分发网络。其基本思路是尽可能避开互联网上有可能影响数据传输速度和稳定性的瓶颈和环节，使内容传输的更快、更稳定。通过在网络各处放置节点服务器所构成的在现有的互联网基础之上的一层智能虚拟网络，CDN系统能够实时地根据网络流量和各节点的连接、负载状况以及到用户的距离和响应时间等综合信息将用户的请求重新导向离用户最近的服务节点上。其目的是使用户可就近取得所需内容，解决 Internet网络拥挤的状况，提高用户访问网站的响应速度。 ——百度百科

当详情页部署在CDN上的时候，这些获取静态资源的请求是不用访问系统，这样也会导致获取不到系统时间，无法对是否开启秒杀做判断，所以要单独的获取时间

也就是，在详情页可能会有大量的获取静态资源的请求，无论是在秒杀开始前还是秒杀开始后，对于这种有可能出现高并发的地方，其中一种的措施是把详情页部署在CDN中，这样不同的用户会请求距离最近的CDN节点来获取这些静态资源，但是这些请求是不用访问系统的，为了时刻获取到系统时间，所以就要从系统中获取时间

那么从系统中获取时间的操作需不需要优化呢？
** 从系统中获取时间不需要优化 **

Java访问一次内存(Cacheline)大约10ns，1秒等于十亿纳秒，当我们访问系统时间，本质上也就是new了一个日期对象，然后返回给用户，在理想状态下，这个操作可以1秒中实现一亿次，没有后端的访问，所以这个操作可以不用优化

2、秒杀地址接口分析

秒杀地址接口无法使用CDN缓存，因为CDN适合对于请求的资源不宜变化的，但是秒杀地址的返回数据是在变化的，随着时间的推移，秒杀活动从未开始到结束，这都是不断变化的

但是适合使用服务器端缓存，比如Redis，先访问数据库，获取到了秒杀的数据，然后放到Redis缓存中，当下一次访问的时候，直接在缓存中查找，缓存中如果有相应的数据，直接返回，而不用再访问数据库

3、秒杀操作优化分析

该操作同样也不能使用CDN缓存，大部分的写操作和核心部分的请求一般无法使用CDN缓存，并且后端缓存困难，因为库存问题，极短的时间内都要对数据库进行更新操作，无法使用缓存技术，否则会出现不一致的错误

4、Java控制事务行为分析

Java控制事务行为

当事务开启的时候，另一个事务要对数据库的同一行进行操作，在当前事务没有commit或者rollback之前，其他事务是无法执行的，此时，正在执行的事务获得了数据库该行的行级锁

当然，如果一个事务持有行级锁的时间极短的话，也可以忽略，但是实际的情况并没有那么理想

事务行为延迟分析

客户端执行update减库存操作，获取到执行后的结果，然后SQL语句会通过网络把结果返回给MySQL的时候，网络延迟是必须考虑的，还要考虑的就是GC

Java GC：(Garbage Collection)垃圾回收机制

GC又分新生代GC和老生代GC，GC不一定每次出现，但是一定会出现

具体关于GC的内容有很多，这也是成为Java开发人员必须要理解掌握的，我作为初学者，也没有深入了解过GC，水平有限，这里只是先提到，要成为Java开发者，GC是避不开的

当Java客户端执行这个事务的时候，Java客户端和数据库之间的网络延迟和可能的GC会持续较长时间，特别对于秒杀系统来说，行级锁的持有时间是一定要优化的

对异地机房之间的通信做一个简单的分析

异地机房

也就是上海机房和北京机房之间的通讯(一来一回)大约需要13毫秒，这是理想状态，实际应该在20毫秒左右，也就是1秒最多执行50次相同的操作

所以优化的思路是：
** 把客户端逻辑放在MySQL服务端，避免网络延迟和GC的影响，从而减少行级锁的持有时间 **

二、Redis后端缓存优化

首先是Redis安装，这个就不多说了，Windows的最好下载MSI，Linux用户可以下载安装包，在控制台进入Redis目录，使用make、make install安装Redis

使用redis-server命令启动Redis，这里只是简单的使用Redis，所以就使用Redis默认的配置

使用redis-cli -p 6379命令连接Redis，默认的端口就是6379

要使用Java访问Redis，需要在pom.xml中添加相应的依赖，首先是引入的是Redis的客户端Jedis，在Redis官网中可以看到各种语言访问redis的客户端是什么

Redis客户端

星号代表的是推荐的客户端，接着引入Jedis

<!-- redis客户端：jedis -->
<dependency>
  <groupId>redis.clients</groupId>
  <artifactId>jedis</artifactId>
  <version>2.7.3</version>
</dependency>

因为要优化的是秒杀暴露接口，所以打开SeckillServiceImpl类，找到exportSeckillUrl方法

Seckill seckill = seckillDao.queryById(seckillId);

要优化的地方就是这一步，对数据库的操作，所有秒杀单都要请求暴露秒杀地址接口，可以使用Redis缓存起来，这样可以降低数据库的访问压力

使用Redis常用的基本的逻辑是：

先从cache中取数据，如果cache中有相应的数据，则直接返回，如果cache中没有数据，则从数据库中获取数据，获得的数据先put到cache中，然后返回给用户

为了后期维护和可扩展性，肯定不能把这些逻辑直接写在业务代码中，也就是不能直接现在SeckillServiceImpl类的exportSeckillUrl方法中

之前说过对数据库或者是其他用于存储的类所在的包是DAO，也就是数据访问对象，在Dao包下新建一个cache包，在cache包下新建一个RedisDao类

public class RedisDao {
    
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
    private final JedisPool jedisPool;
    
    public RedisDao(String ip, int port){
        jedisPool = new JedisPool(ip, port);
    }
}

要在后台打印信息，所以使用slf4j，之前已经引入了Jedis的依赖，所以可以直接使用，有点类似于数据库连接池的ConnectionPool

然后初始化一个构造方法，出入ip和port，用于连接Redis，因为本项目只是简单的运用Redis，不需要复杂的配置，所以直接初始化JedisPool即可

这个类是对Redis进行操作，我们使用Redis是要从缓存中获取Seckill对象，如果缓存中没有，还要对数据库进行操作，并把数据存放在缓存中

/**
 * 从Redis获取数据，由于在redis中的数据都是字节数组，所以需要对数据进行反序列化，转化为想要的类型对象
 * @param seckillId
 * @return
 */
public Seckill getSeckill(long seckillId){
    //redis操作逻辑
    try{
        Jedis jedis = jedisPool.getResource();
        try{
            String key = "seckill: " + seckillId;
            //没有实现内部序列化操作，采用自定义序列化
            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;
}

这个方法用于从Redis中获取数据

首先是获取到Jedis对象，使用了Jedis就要关闭，所以在接下来的逻辑还要再try/catch，在finally中关闭Jedis

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

由于Redis并没有实现内部序列化操作，所以在Redis中存储的数据均为字节码数组，所以就要通过反序列化获取到Object数据，也就是Seckill对象

在GitHub上专门有一个对Java序列化技术的性能比对

本项目采用自定义序列化，使用的是protostuff，需要两个依赖

<!-- 自定义序列化技术 -->
<dependency>
  <groupId>com.dyuproject.protostuff</groupId>
  <artifactId>protostuff-core</artifactId>
  <version>1.0.12</version>
</dependency>
    
<dependency>
  <groupId>com.dyuproject.protostuff</groupId>
  <artifactId>protostuff-runtime</artifactId>
  <version>1.0.12</version>
</dependency>

这样就可以实现把一个对象转换成二进制数组，然后传到Redis中

使用protostuff的API，全局的定义一个动态的schema，是由protostuff自定转换的，对性能几乎没影响

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

class对象代表类的字节码对象，也就是这个类的类类型，可以通过反射获取到这个类有哪些属性和方法，RuntimeSchema就是基于传递参数的字节码对象来做一个模式，当创建对象的时候，会根据不同的模式赋予相应的值

序列化就是根据字节码和字节码所对应的对象有哪些属性，把字节码的数据传递给那些属性，这样就可以序列化好一个对象

传递的参数是这个对象的class，也就是这个对象的类类型，再传递一个类似schema的内容，会描述这个对象的结构

回到getSeckill方法中

//Redis没有实现内部序列化操作，采用自定义序列化
byte[] bytes = jedis.get(key.getBytes());   
if(bytes != null){
    Seckill seckill = schema.newMessage();
    ProtostuffIOUtil.mergeFrom(bytes, seckill, schema);
    return seckill;
}

使用Jedis的get方法，返回的是字节码数组，然后对字节码数组进行判断，如果bytes不为空，则是从Redis中获取到了数据，然后使用protostuff进行转换

protostuff提供了一个Util工具类来进行转换，需要传递的参数有这个对象的字节码数组，空的对象，和自定义的模式schema，然后protostuff就会把字节码数组按照传入的模式，即schema，将数据传入空对象中，这样Seckill对象就已经被赋值了，这样就是把一个字节数组转化为我们想要的对象，也就是对象反序列化

当Redis缓存中有数据的时候，使用getSeckill方法可以获取到数据，当Redis缓存中没有数据的时候，还需要向Redis缓存中存放数据

/**
 * 把数据放入Redis中时，需要对数据进行序列化，转化为字节数组
 * @param seckill
 * @return
 */
public String putSeckill(Seckill seckill){
    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;//一小时
            String result = jedis.setex(key.getBytes(), timeout, bytes);
            return result;
        } finally {
            jedis.close();
        }
    } catch(Exception e) {
        logger.error(e.getMessage(), e);
    }
    return null;
}

大部分代码和getSeckill方法相同，主要分析不同部分

try{
    String key = "seckill: " + seckill.getSeckillId();
    byte[] bytes = ProtostuffIOUtil.toByteArray(seckill, schema, LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE));
                
    //超时缓存
    int timeout = 60 * 60;//一小时
    String result = jedis.setex(key.getBytes(), timeout, bytes);
    return result;
} finally {
    jedis.close();
}

我们需要把Object对象，这里是Seckill对象转换成字节码数组，然后put到Redis中，这个过程也就是序列化的过程

首先设置一个key，然后使用protostuff的IOUtil工具来得到相应对象的字节码数组，传递的对象有Seckill对象本身，之前自定义的模式schema，需要一个缓存器，缓存器的大小直接设置默认的大小即可，当对象数据特别大的时候，会有缓冲的过程

接着使用Jedis的setex方法，也就是超时缓存，返回的是String类型，如果错误，则返回的是错误信息，如果正确，则会返回OK

然后回到SeckillServiceImpl类，既然有了RedisDao，通过注解注入到Spring容器中

@Autowired
private RedisDao redisDao;

然后把exportSeckillUrl方法中

Seckill seckill = seckillDao.queryById(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);
    }
}

首先通过RedisDao从Redis中获取数据，然后对获取到的数据进行判断，如果为空，说明缓存中没有相应的数据，这时就需要从数据库中获取数据，获取到的数据还要再放入Redis缓存中

三、秒杀操作并发优化

事务执行流程

当一个事务在执行对数据库的更改操作的时候，会获得该行的行级锁，然后通过update返回的结果来决定是否进行insert操作，最后是commit或者rollback，这时行级锁也被释放，期间肯定会有网络延迟和GC的影响

1、简单优化

Paste_Image.png

主要的目的是减少行级锁的持有时间，通过对事务执行的流程进行下调换就可以使行级锁的持有时间大幅度减少

把insert操作放在前面，我们在创建success_killed表的时候就设置了seckilId和userPhone为联合主键，这个主键冲突的概率并不是很高

然后才是update减库存操作，这时候会获取到该行的行级锁，通过insert的返回结果来决定是否进行update操作，最后是commit或者rollback

@Transactional
public SeckillExecution executeSeckill(long seckillId, long userPhone, String md5) {
    if(md5 == null || !md5.equals(getMD5(seckillId))){
        throw new SeckillException("seckill data rewrite");
    }
    //执行秒杀逻辑：减库存 + 记录购买行为
    Date nowTime = new Date();
    try {
        //记录购买行为
        int insertCount = successKilledDao.insertSuccessKilled(seckillId, userPhone);
        if(insertCount <= 0){
            //重复秒杀
            throw new RepeatKillException("seckill repeated");
        } else {
            //减库存
            int updateCount = seckillDao.reduceNumber(seckillId, nowTime);
            if(updateCount <= 0){
                //没有更新记录，即秒杀活动结束
                throw new SeckillCloseException("seckill is closed");
            } else {
                //秒杀成功
                SuccessKilled successKilled = successKilledDao.queryByIdWithSeckill(seckillId, userPhone);
                return new SeckillExecution(seckillId, SeckillStateEnum.SUCCESS, successKilled);
            }
        }
    } catch(SeckillCloseException e1) {
        throw e1;
    } catch(RepeatKillException e2){
        throw e2;
    } catch (Exception e) {
        logger.error(e.getMessage(), e);
        //所有编译期异常转化为运行期异常
        throw new SeckillException("seckill inner error: " + e.getMessage());
    }
}

主要改变的是中间try/catch部分

2、深度优化

同样的目的是降低行级锁到commit或者rollback之间的时间，同时还要让MySQL获得更高的QPS，所以将事务SQL放在MySQL端来执行，之前是在MyBatis的配置文件中编写SQL语句的，这里使用存储过程，将编写的SQL语句直接在MySQL端执行

在sql目录下新建一个seckill.sql文件

-- 秒杀执行存储过程
DELIMITER $$ --console ; 转换为 $$
-- 定义存储过程
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
            success_killed (seckill_id,user_phone,create_time)
            value (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 
$$
-- 存储过程定义结束 

首先要知道的是MySQL中的console是用分号来隔离的，但是在存储过程中也是通过分号来决定是否换行，所以使用MySQL的DELIMITER暂时地把MySQL的隔离符号改为$$

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

接着创建存储过程

CREATE PROCEDURE `seckill`.`execute_seckill`
    (in v_seckill_id bigint, in v_phone bigint,
     in v_kill_time timestamp, out r_result int)

同时定义一些变量：

• in：输入参数，在存储过程中可以被使用
• out：输出参数，在存储过程中不能被使用，但是可以被赋值

然后开始编写存储过程的逻辑

DECLARE insert_count int DEFAULT 0;
START TRANSACTION;

先是定义一个变量insert_count，默认设置为0，然后开启事务

先执行insert语句，插入用户的购买明细

insert ignore into
    success_killed (seckill_id,user_phone,create_time)
    value (v_seckill_id,v_phone,v_kill_time);
select row_count() into insert_count;

不要忘记ignore关键字，然后使用** MySQL的内置函数row_count()，用于显示上一条修改类型SQL语句执行后被影响的行数 **，把row_count()的值赋值给insert_count变量，接着对insert_count进行判断

IF (insert_count = 0) THEN
    ROLLBACK;
    set r_result = -1;-- 重复秒杀
ELSEIF (insert_count < 0) THEN
    ROLLBACK;
    set r_result = -2;-- 系统错误
ELSE

• insert_count = 0：未修改语句，设置输出参数r_result的值为-1，对应之前在Java中创建的数据字典就是重复秒杀
• insert_count < 0：SQL错误/未执行修改SQL，设置输出参数r_result的值为-2，对应之前在Java中创建的数据字典就是系统错误

如果insert_count的值大于0，则执行update操作

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;

同样最后使用row_count()函数来判断是否commit或者rollback

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;

• insert_count = 0：未更改数据，设置输出参数r_result的值为0，对应之前在Java中创建的数据字典就是秒杀结束
• insert_count < 0：SQL错误/未执行修改SQL，设置输出参数r_result的值为-2，对应之前在Java中创建的数据字典就是系统错误
• insert_count > 0：成功更改数据，设置输出参数r_result的值为1，对应之前在Java中创建的数据字典就是秒杀成功

最后结尾的时候使用$$结束存储过程，然后把换行符再改为分号

DELIMITER ;

整个过程就是把insert插入购买明细及update减库存操作放入存储过程中，在MySQL端直接执行这些语句，然后就可以直接调用这个存储过程

set @r_result = -3;
-- 执行存储过程
call execute_seckill(1003,13522233356,now(),@r_result);
-- 获取结果
select @r_result;

首先是定义一个变量，在console中定义变量使用@，通过call执行存储过程

然后在Service层的SeckillService类中添加通过使用存储过程执行秒杀操作的方法，通过Java客户端来调用存储过程，这时只要获取到返回值，通过数据字典就可以判断执行结果

/**
 * 执行秒杀操作by存储过程
 * @param seckillId
 * @param userPhone
 * @param md5
 * @return
 */
SeckillExecution executeSeckillProcedure(long seckillId, long userPhone, String md5);

接着在SeckillServiceimpl中复写这个方法

public SeckillExecution executeSeckillProcedure(long seckillId, long userPhone, String md5) {
    if(md5 == null || !md5.equals(getMD5(seckillId))){
        return new SeckillExecution(seckillId, SeckillStateEnum.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);
        int result = MapUtils.getInteger(map, "result", -2);
        if(result == 1){
            SuccessKilled sk = successKilledDao.queryByIdWithSeckill(seckillId, userPhone);
            return new SeckillExecution(seckillId, SeckillStateEnum.SUCCESS, sk);
        }else{
            return new SeckillExecution(seckillId, SeckillStateEnum.stateOf(result));
        }
    } catch (Exception e) {
        logger.error(e.getMessage(), e);
        return new SeckillExecution(seckillId, SeckillStateEnum.INNER_ERROR);
    }
}

首先要验证的是MD5，如果MD5值错误，就返回SeckillExecution对象，数据字典中传递数据被篡改

if(md5 == null || !md5.equals(getMD5(seckillId))){
    return new SeckillExecution(seckillId, SeckillStateEnum.DATA_REWRITE);
}

重新获取系统时间

Date killTime = new Date();

接着要使用SeckillDao获取到数据库的执行结果，所以在SeckillDao类中创建一个方法

/**
 * 使用存储过程执行秒杀
 * @param paramMap
 */
void killByProcedure(Map<String, Object> paramMap);

传入的参数是Map，这个方法要在SeckillDao.xml中使用，也就是MyBatis的配置文件，还是要通过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>

CALLABLE是jdbc专门为调用存储过程而开发的，使用call，后面跟存储过程的名称

每个传递的参数都包括要传递的值，jdbc的类型，参数的模式

回到SeckillServiceImpl的executeSeckillProcedure中，要使用killByProcedure，需要传入Map类型的参数，所以要先声明Map对象，

Map<String, Object> map = new HashMap<String, Object>();
map.put("seckillId", seckillId);
map.put("phone", userPhone);
map.put("killTime", killTime);
map.put("result", null);

Result现在为空，所以赋值null

然后就可以使用SeckillDao来获取数据库执行的结果，因为使用SeckillDao的方法的时候可能会出现异常，所以try/catch

try {
    seckillDao.killByProcedure(map);
    int result = MapUtils.getInteger(map, "result", -2);
    if(result == 1){
        SuccessKilled sk = successKilledDao.queryByIdWithSeckill(seckillId, userPhone);
        return new SeckillExecution(seckillId, SeckillStateEnum.SUCCESS, sk);
    }else{
        return new SeckillExecution(seckillId, SeckillStateEnum.stateOf(result));
    }
} catch (Exception e) {
    logger.error(e.getMessage(), e);
    return new SeckillExecution(seckillId, SeckillStateEnum.INNER_ERROR);
}

然后是获取result，使用MapUtil，要使用MapUtil，需要在pom.xml中引入依赖

<dependency>
  <groupId>commons-collections</groupId>
  <artifactId>commons-collections</artifactId>
  <version>3.2</version>
</dependency>

使用MapUtil传递的参数需要传递map，获取指定属性的值，如果没有获取到该属性，则赋值-2，表示系统内部错误

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

接着就可以通过result进行判断

if(result == 1){
    SuccessKilled sk = successKilledDao.queryByIdWithSeckill(seckillId, userPhone);
    return new SeckillExecution(seckillId, SeckillStateEnum.SUCCESS, sk);
}else{
    return new SeckillExecution(seckillId, SeckillStateEnum.stateOf(result));
}

如果等于1，获取用户购买明细，返回SeckillExecution对象，否则也返回SeckillExecution对象，但是可能是各种各样的异常，所以使用SeckillStateEnum的stateOf方法

public static SeckillStateEnum stateOf(int index){
    for(SeckillStateEnum state : values()){
        if(state.getState() == index){
            return state;
        }
    }
    return null;
}

通过传递的数字对应相应的字符串，如果不在数据字典中的异常，就要在catch中再返回一次，传递的参数是系统错误

logger.error(e.getMessage(), e);
return new SeckillExecution(seckillId, SeckillStateEnum.INNER_ERROR);

至此，高并发优化完成了