【IDEA+SpringBoot+Java商城秒杀实战20】高并发秒杀接口优化

Ona ·

更新时间:2024-09-21

· 772 次阅读

高并发秒杀接口优化 接口优化（核心思路：减少对数据库的访问） Redis预减库存减少对数据库的访问内存标记减少Redis的访问请求先入队缓冲，异步下单，增强用户体验 RabbitMQ安装与SpringBoot的集成（目的：同步下单改成异步下单） Nginx水平拓展压测

项目迭代是一个优化和调整的过程，发现问题解决问题，不断优化。
秒杀业务场景，并发量很大，瓶颈在数据库，怎么解决，加缓存。用户发起请求时，从浏览器开始,在浏览器上做页面静态化直接将页面缓存到用户的浏览器端，然后请求到达网站之前可以部署CDN节点，让请求先访问CDN，到达网站时候使用页面缓存。页面缓存再进一步的话，粒度再细一点的话就是对象缓存。缓存层依次请求完之后，才是数据库。通过一层一层的访问缓存逐步的削减到达数据库的请求数量，这样才能保证网站在大并发之下抗住压力。

但是仅仅依靠缓存还不够，还需要进行接口优化。

秒杀接口优化大致实现步骤：

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

收到请求，Redis预减库存（先减少Redis里面的库存数量，库存不足，直接返回），如果库存已经到达临界值的时候，即=0，就不需要继续往下走，直接返回失败，否正进入3

请求入队，立即返回排队中

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

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

1.商品库存数量预加载库存到Redis上 MiaoshaController实现InitializingBean接口，重写afterPropertiesSet方法。 目的： 在容器启动的时候，检测到了实现了接口InitializingBean之后，就回去回调afterPropertiesSet方法。将每种商品的库存数量加载到redis里面去。

@RequestMapping("/miaosha")
@Controller
public class MiaoshaController implements InitializingBean{	
	public void afterPropertiesSet() throws Exception {
		List goodslist=goodsService.getGoodsVoList();
		if(goodslist==null) {
			return;
		}
		for(GoodsVo goods:goodslist) {
			//如果不是null的时候，将库存加载到redis里面去 prefix---GoodsKey:gs ,	 key---商品id,	 value
			redisService.set(GoodsKey.getMiaoshaGoodsStock, ""+goods.getId(), goods.getStockCount());
		}
	}
}

@Service
public class GoodsService {
	public static final String COOKIE1_NAME_TOKEN="token";	
	@Autowired
	GoodsDao goodsDao;
	@Autowired
	RedisService redisService;
	public List  getGoodsVoList() {
		return goodsDao.getGoodsVoList();
	}
}

@Mapper
public interface GoodsDao {
	//两个查询
	@Select("select g.*,mg.stock_count,mg.start_date,mg.end_date,mg.miaosha_price from miaosha_goods mg left join goods g on 								mg.goods_id=g.id")  
	public List getGoodsVoList();
}

2.收到请求后预减库存

后端接收秒杀请求的接口doMiaosha，收到请求，Redis预减库存（先减少Redis里面的库存数量，库存不足，直接返回），如果库存已经到达临界值的时候，即=0，就不需要继续往下走，直接返回失败

@RequestMapping(value="/{path}/do_miaosha",method=RequestMethod.POST)
@ResponseBody
public Result doMiaosha(Model model,MiaoshaUser user,
		@RequestParam(value="goodsId",defaultValue="0") long goodsId,
		@PathVariable("path")String path) {
	model.addAttribute("user", user);
	//1.如果用户为空，则返回至登录页面
	if(user==null){
		return Result.error(CodeMsg.SESSION_ERROR);
	}	
	//2.预减少库存，减少redis里面的库存
	long stock=redisService.decr(GoodsKey.getMiaoshaGoodsStock,""+goodsId);
	//3.判断减少数量1之后的stock，区别于查数据库时候的stock<=0
	if(stock是一个事务
	OrderInfo orderinfo=miaoshaService.miaosha(user,goodsvo);
	return Result.success(orderinfo);
}

RedisService里面的decr方法：减少key对应的值

/**
	 * 减少值
	 * @param prefix
	 * @param key
	 * @return
	 */
	public  Long decr(KeyPrefix prefix,String key){
		Jedis jedis=null;
		try {
			jedis=jedisPool.getResource();
			String realKey=prefix.getPrefix()+key;
			return jedis.decr(realKey);
		}finally {
			returnToPool(jedis);
		}
	}

3.消息入队（并将用户信息和商品信息封装起来传入队列）

再次改造doMiaosha接口方法，在收到请求之后，请求入队：

@RequestMapping(value="/{path}/do_miaosha",method=RequestMethod.POST)
	@ResponseBody
	public Result doMiaosha(Model model,MiaoshaUser user,
			@RequestParam(value="goodsId",defaultValue="0") long goodsId,
			@PathVariable("path")String path) {
		model.addAttribute("user", user);
		//1.如果用户为空，则返回至登录页面
		if(user==null){
			return Result.error(CodeMsg.SESSION_ERROR);
		}	
		//2.预减少库存，减少redis里面的库存
		long stock=redisService.decr(GoodsKey.getMiaoshaGoodsStock,""+goodsId);
		//3.判断减少数量1之后的stock，减少到0一下，则代表之后的请求都失败，直接返回
		if(stock<0) {
			return Result.error(CodeMsg.MIAOSHA_OVER_ERROR);
		}
		//4.判断这个秒杀订单形成没有，判断是否已经秒杀到了，避免一个账户秒杀多个商品
		MiaoshaOrder order = orderService.getMiaoshaOrderByUserIdAndCoodsId(user.getId(), goodsId);
		if (order != null) {// 查询到了已经有秒杀订单，代表重复下单
			return Result.error(CodeMsg.REPEATE_MIAOSHA);
		}
		//5.正常请求，入队，发送一个秒杀message到队列里面去，入队之后客户端应该进行轮询。
		MiaoshaMessage mms=new MiaoshaMessage();
		mms.setUser(user);
		mms.setGoodsId(goodsId);
		mQSender.sendMiaoshaMessage(mms);
		//返回0代表排队中
		return Result.success(0);
	}

MiaoshaMessage 消息的封装类:

//MiaoshaMessage 消息的封装  MiaoshaMessage Bean
@Getter
@Serter
public class MiaoshaMessage {
	private MiaoshaUser user;
	private long goodsId;
}

注意：消息队列这里，消息只能传字符串，MiaoshaMessage 这里是个Bean对象，是先用beanToString方法，将转换为String，放入队列，使用AmqpTemplate传输。

	@Autowired
	RedisService redisService;
	@Autowired
	AmqpTemplate amqpTemplate;
	public void sendMiaoshaMessage(MiaoshaMessage mmessage) {
		// 将对象转换为字符串
		String msg = RedisService.beanToString(mmessage);
		log.info("send message:" + msg);
		// 第一个参数队列的名字，第二个参数发出的信息
		amqpTemplate.convertAndSend(MQConfig.MIAOSHA_QUEUE, msg);
	}
	/**
	 * 将Bean对象转换为字符串类型
	 * @param 
	 */
	public static  String beanToString(T value) {
		//如果是null
		if(value==null) return null;
		//如果不是null
		Class clazz=value.getClass();
		if(clazz==int.class||clazz==Integer.class) {
			return ""+value;
		}else if(clazz==String.class) {
			return ""+value;
		}else if(clazz==long.class||clazz==Long.class) {
			return ""+value;
		}else {
			return JSON.toJSONString(value);
		}		
	}

redis多线程情况下是否安全？

1.新建RedisConcurrentTestUtil类来测试redis多线程是否安全，如果安全，那么应该只有10个线程能通过if(stock>=0)的判断，进行后面的秒杀操作!

@Service
public class RedisConcurrentTestUtil {
	@Autowired
	RedisService redisService;	//会出现循环依赖---Circular reference
	class ThreadTest implements Runnable{
		@Override
		public void run() {
			long stock=redisService.get("GoodsKey:gs1",Long.class);
			String name=Thread.currentThread().getName();
			System.out.println("当前线程 ："+name+"  stock："+stock);
			//2.预减少库存，减少redis里面的库存
			//stock最初为10，100个线程同时去减少1次，最终stock应该为-90
			stock=redisService.decr(GoodsKey.getMiaoshaGoodsStock,""+1);
			//是否线程安全？
			if(stock<0) {
				System.out.println("结束!!!");
				return;
			}
			//应该只有10个线程能从这里通过
			System.out.println("验证当前有几个线程通过if(stock<0)  当前线程 ："+name+"  减1之后的stock："+stock);
		}
	}
	public  void test(){
		ThreadTest t1=new ThreadTest();
		//开启50个线程
		for(int i=1;i<=100;i++){
			new Thread(t1,"Thread-"+i).start();
		}
	}
}

31
2.在某一个类（DemoController）里面定义一个接收请求的方法。

3.进入redis，设置GoodsKey:gs1这个键的值为10。

几个常用的redis命令：
auth [password] 输入密码
keys *查询所有键
flushdb 清空数据

4.发送请求，验证结果！

5.结果如下：
可以得出多线程下redis的操作是线程安全的：

redis里面库存结果：

RabbitMQ在Windows上面的安装

安装rabbitmq首先要首先安装erlang。

1.在Win环境下安装Erlang 2.在Win环境下安装RabbitMQ SpringBoot集成RabbitMQ 添加依赖spring-boot-starter-amqp

	
      org.springframework.boot
      spring-boot-starter-amqp

在application.properties配置文件里面添加RabbitMQ配置信息

#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.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
#最大间隔 10s
spring.rabbitmq.template.retry.max-interval=10000
spring.rabbitmq.template.retry.multiplier=1.0

创建消息发送者与接收者

//发送者
@Service
public class MQSender {
	private static Logger log=LoggerFactory.getLogger(MQSender.class);
	@Autowired
	RedisService redisService;
	@Autowired
	AmqpTemplate amqpTemplate;
	/**
	 * 发送秒杀信息，使用derict模式的交换机。（包含秒杀用户信息，秒杀商品id）
	 */
	public void sendMiaoshaMessage(MiaoshaMessage mmessage) {
		// 将对象转换为字符串
		String msg = RedisService.beanToString(mmessage);
		log.info("send message:" + msg);
		// 第一个参数队列的名字，第二个参数发出的信息
		amqpTemplate.convertAndSend(MQConfig.MIAOSHA_QUEUE, msg);
	}	
}

创建MQ的config，然后创建队列

@Configuration
public class MQConfig {
	public static final String QUEUE="queue";
	public static final String MIAOSHA_QUEUE="miaosha.queue";
	public static final String TOPIC_QUEUE1="topic.queue1";
	public static final String TOPIC_QUEUE2="topic.queue2";
	public static final String HEADER_QUEUE="header.queue";
	public static final String TOPIC_EXCHANGE="topic.exchange";
	public static final String FANOUT_EXCHANGE="fanout.exchange";
	public static final String HEADER_EXCHANGE="header.exchange";
	public static final String ROUTINIG_KEY1="topic.key1";
	public static final String ROUTINIG_KEY2="topic.#";
	/**
	 * Direct模式，交换机Exchange:
	 * 发送者，将消息往外面发送的时候，并不是直接投递到队列里面去，而是先发送到交换机上面，然后由交换机发送数据到queue上面去，
	 * 做了依次路由。
	 */
	@Bean
	public Queue queue() {
		//名称，是否持久化
		return new Queue(QUEUE,true);
	}
	@Bean
	public Queue miaoshaqueue() {
		//名称，是否持久化
		return new Queue(MIAOSHA_QUEUE,true);
	}
}

MiaoshaMessage类
public class MiaoshaMessage {
	private MiaoshaUser user;
	private long goodsId;
	public MiaoshaUser getUser() {
		return user;
	}
	public void setUser(MiaoshaUser user) {
		this.user = user;
	}
	public long getGoodsId() {
		return goodsId;
	}
	public void setGoodsId(long goodsId) {
		this.goodsId = goodsId;
	}
}