秒杀

Redis

发布日期: 2021-10-09

文章字数: 3.9k

阅读时长: 17 分

阅读次数:

SpringBoot+Redis+RabbitMq专题应用场景 - 秒杀业务

问题域

超卖问题 - redis单线程的模型
重复抢购问题.
秒杀 - 结算付款 - “削峰” - rabbitMQ

技术点

SpringBoot任务调度[定时器]
RedisTemplate的API
RabbitMQ[消息队列]

秒杀流程图

rabbitmq - 流量削峰

核心表介绍

drop table web_seckill;
drop table web_order;

-- 订单表
create table web_order(
    id          int auto_increment comment '主键' primary key,
    orderNo     varchar(100)           null comment '订单编号',
    uName       varchar(40)            null comment '用户名',
    orderStatus varchar(2) default '0' null comment '0已经创建 1未支付  2已经支付  3已经收货  4已经评价',
    secId       bigint                 null comment '秒杀ID'
);

-- 秒杀表 - 当有秒杀活动的时候,一定是要创建秒杀表,存放秒杀信息.
-- 核心字段status
create table web_seckill(
    id           bigint auto_increment comment '秒杀主键' primary key,
    productId    bigint                 null comment '商品id',
    num          int                    null comment '秒杀名额',
    seckillPrice float(10, 2)           null comment '秒杀价',
    startTime    datetime               null comment '开始时间',
    endTime      datetime               null comment '结束时间',
    status       varchar(2) default '0' null comment '0未开始，1已开始，2已经结束',
    href         varchar(200)           null comment '秒杀详情页'
);

-- 模拟数据
INSERT INTO web_seckill (id, productId, num, seckillPrice, startTime, endTime, status, href) VALUES (1, 3, 30, 569.9, '2021-10-09 14:00:50', '2021-10-09 16:55:10', '0', '88081.html');
INSERT INTO web_seckill (id, productId, num, seckillPrice, startTime, endTime, status, href) VALUES (2, 1, 5, 5, '2021-10-09 14:00:50', '2021-10-09 16:55:10', '0', '88082.html');

RabbitMQ介绍和使用

介绍 - 消息队列中间件

RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现。

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合[降低各个服务之间的耦合性]，异步消息，流量削锋,分布式事务等问题实现高性能，高可用，可伸缩和最终一致性[架构]

使用较多的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ。其使用场景为：异步处理，应用解耦，流量削锋和消息通讯四个场景。

下载和安装

Erlang - http://www.erlang.org/downloads

RabbitMQ - https://github.com/rabbitmq/rabbitmq-server/releases

注意一下两者的版本,需要相互兼容.否则安装运行失败!

默认的端口号:15672

浏览器输入localhost:15672

默认的账号和密码都是guest

SpringBoot整合rabbitmq

导入依赖

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

application-dev.yml

spring:
    rabbitmq:
    host: localhost

架构图

主要概念

RabbitMQ Server：也叫broker server，它是一种传输服务。他的角色就是维护一条从Producer到Consumer的路线，保证数据能够按照指定的方式进行传输。

Producer：消息生产者，如图A、B、C，数据的发送方。消息生产者连接RabbitMQ服务器然后将消息投递到Exchange。
Consumer：消息消费者，如图1、2、3，数据的接收方。消息消费者订阅队列，RabbitMQ将Queue中的消息发送到消息消费者。
Exchange：生产者将消息发送到Exchange（交换器），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。Exchange并不存储消息。RabbitMQ中的Exchange有direct、fanout、topic、headers四种类型，每种类型对应不同的路由规则。
Queue：（队列）是RabbitMQ的内部对象，用于存储消息。消息消费者就是通过订阅队列来获取消息的，RabbitMQ中的消息都只能存储在Queue中，生产者生产消息并最终投递到Queue中，消费者可以从Queue中获取消息并消费。多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。
RoutingKey：生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。RabbitMQ为routing key设定的长度限制为255bytes，
Connection：（连接）：Producer和Consumer都是通过TCP连接到RabbitMQ Server的。以后我们可以看到，程序的起始处就是建立这个TCP连接。
Channels：（信道）：它建立在上述的TCP连接中。数据流动都是在Channel中进行的。也就是说，一般情况是程序起始建立TCP连接，第二步就是建立这个Channel。
VirtualHost：权限控制的基本单位，一个VirtualHost里面有若干Exchange和MessageQueue，以及指定被哪些user使用.

RabbitMQ三种模式

1. 直接模式

我们需要将消息发给唯一一个节点时使用这种模式，这是最简单的一种形式。

任何发送到Direct Exchange的消息都会被转发到RouteKey中指定的Queue。

一般情况可以使用rabbitMQ自带的Exchange：””(该Exchange的名字为空字符串，下文称其为default Exchange)。

这种模式下不需要将Exchange进行任何绑定(binding)操作
消息传递时需要一个“RouteKey”，可以简单的理解为要发送到的队列名字。
如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。

具体操作

新建一个队列 - j03s_direct

发送方

package tech.aistar.rabbitmq;

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import tech.aistar.mapper.UsersMapper;
import tech.aistar.model.entity.Users;

/**
 * 本类用来演示:
 *
 * @author: success
 * @date: 2021/10/9 9:24 上午
 */
@SpringBootTest
public class TestRabbitMq {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private UsersMapper usersMapper;

    /**
     * 直接模式 - 模拟的是生产者
     */
    @Test
    public void testDirect(){
        Users u = usersMapper.findByEmail("849962874@qq.com");

        //秒杀 - 从redis中删除一条数据,需要把删除的数据保存到rabbitmq中.
        //rabbitTemplate.convertAndSend("j03s_direct","我直接一个直接模式!");
        rabbitTemplate.convertAndSend("j03s_direct",u);

    }
}

接收方

package tech.aistar.handler;
   
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import tech.aistar.model.entity.Users;
   
import java.io.IOException;
import java.util.Map;
   
/**
 * 本类用来演示: 直接模式 1:1 - 只要j03s_direct队列上一旦被生产者发送了消息
 *
 * 那么此处将会进行消费
 *
 * @author: success
 * @date: 2021/10/9 9:33 上午
 */
@Component
@RabbitListener(queues = "j03s_direct")
public class RabbitMqService {
   
    /**
     * 参数需要和发送方发送的参数保持一致即可
     * @param msg
     */
//    @RabbitHandler
//    public void getDirect01(String msg){
//        System.out.println("msg01:"+msg);
//    }
   
//    /**
//     * 可以进行删除删除的操作
//     * @param u
//     */
//    @RabbitHandler
//    public void getDirect01(Users u){
//        System.out.println("msg01:"+u);
//    }
   
    /**
     * 可以进行删除删除的操作
     * @param u
     */
    @RabbitHandler
    public void getDirect01(Users u, @Headers Map<String,Object> header, Channel channel){
   
        System.out.println("msg01:"+u);
   
        //手动删除rabbitmq队列中的数据的.
        long ack = (long) header.get(AmqpHeaders.DELIVERY_TAG);
        try {
            channel.basicAck(ack,false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

配置一个手动ack - 手动删除rabbitmq队列中的数据

spring:
    rabbitmq:
    localhost: localhost
    listener:
      simple:
        # 手动ack - 手动删除
        acknowledge-mode: manual

2. 分裂模式

当我们需要将消息一次发给多个队列时，需要使用这种模式。如下图：

任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上。

可以理解为路由表的模式

这种模式不需要RouteKey

这种模式需要提前将Exchange与Queue进行绑定，一个Exchange可以绑定多个Queue，一个Queue可以同多个Exchange进行绑定。

如果接受到消息的Exchange没有与任何Queue绑定，则消息会被抛弃.

具体操作

新建一个Exchange - 交换器并且要设置它的type - fanout

一个交换器是需要和队列进行绑定的 - routingKey - “”

2-1. 新建了俩个队列e1,e2

2-2. 交换器和队列e1,e2进行绑定操作

发送方

package tech.aistar.rabbitmq;

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import tech.aistar.mapper.UsersMapper;
import tech.aistar.model.entity.Users;

/**
 * 本类用来演示:
 *
 * @author: success
 * @date: 2021/10/9 9:24 上午
 */
@SpringBootTest
public class TestRabbitMq {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private UsersMapper usersMapper;

    /**
     * 直接模式 - 模拟的是生产者
     */
    @Test
    public void testDirect(){
        Users u = usersMapper.findByEmail("849962874@qq.com");

        //秒杀 - 从redis中删除一条数据,需要把删除的数据保存到rabbitmq中.
        //rabbitTemplate.convertAndSend("j03s_direct","我直接一个直接模式!");
        rabbitTemplate.convertAndSend("j03s_direct",u);

    }

    /**
     * 分裂模式 - 发送方把数据发送交换器,由交换器把数据路由[转发]到各个队列上.
     * 交换器是不会存储消息[数据]的
     */
    @Test
    public void testFanout(){
        rabbitTemplate.convertAndSend("j03s_fanout","","我是分裂模式!");
    }
}

接收方

package tech.aistar.handler;
   
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
   
import java.io.IOException;
import java.util.Map;
   
/**
 * 本类用来演示: 分裂模式 - 一个生产者具有多个消费者
 *
 * @author: success
 * @date: 2021/10/9 10:25 上午
 */
@Component
public class RabbitMQFanout {
   
    @RabbitListener(queues = "e1")
    @RabbitHandler
    public void getE1(String msg, @Headers Map<String,Object> header, Channel channel){
        System.out.println("msg01-"+msg);
   
        //手动删除rabbitmq队列中的数据的.
        long ack = (long) header.get(AmqpHeaders.DELIVERY_TAG);
        try {
            channel.basicAck(ack,false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
   
    @RabbitListener(queues = "e2")
    @RabbitHandler
    public void getE2(String msg, @Headers Map<String,Object> header, Channel channel){
        System.out.println("msg02-"+msg);
   
        //手动删除rabbitmq队列中的数据的.
        long ack = (long) header.get(AmqpHeaders.DELIVERY_TAG);
        try {
            channel.basicAck(ack,false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

3. 主题模式

任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上

如上图所示此类交换器使得来自不同的源头的消息可以到达一个对列，其实说的更明白一点就是模糊匹配的意思，例如：上图中红色对列的routekey为usa.#，#代表匹配任意字符，但是要想消息能到达此对列，usa.必须匹配后面的#好可以随意。图中usa.newsusa.weather,都能找到红色队列，符号 # 匹配一个或多个词，符号 * 匹配不多不少一个词。因此 usa.# 能够匹配到 usa.news.XXX ，但是 usa. 只会匹配到 usa.XXX 。*

注：
交换器说到底是一个名称与队列绑定的列表。当消息发布到交换器时，实际上是由你所连接的信道，将消息路由键同交换器上绑定的列表进行比较，最后路由消息。任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上

这种模式较为复杂，简单来说，就是每个队列都有其关心的主题，所有的消息都带有一个“标题”(RouteKey)，Exchange会将消息转发到所有关注主题能与RouteKey模糊匹配的队列。
这种模式需要RouteKey，也许要提前绑定Exchange与Queue。
在进行绑定时，要提供一个该队列关心的主题，如“#.log.#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。
“#”表示0个或若干个关键字，“”表示一个关键字。如“log.”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.#”能与上述两者匹配。
同样，如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息

具体操作

新建一个交换器j03s_topic - type设置成topic

新建俩个队列top1,top2,top3,但是需要设置routingkey - 满足上面的正则

绑定

发送方

package tech.aistar.rabbitmq;

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import tech.aistar.mapper.UsersMapper;
import tech.aistar.model.entity.Users;

/**
 * 本类用来演示:
 *
 * @author: success
 * @date: 2021/10/9 9:24 上午
 */
@SpringBootTest
public class TestRabbitMq {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private UsersMapper usersMapper;

    /**
     * 直接模式 - 模拟的是生产者
     */
    @Test
    public void testDirect(){
        Users u = usersMapper.findByEmail("849962874@qq.com");

        //秒杀 - 从redis中删除一条数据,需要把删除的数据保存到rabbitmq中.
        //rabbitTemplate.convertAndSend("j03s_direct","我直接一个直接模式!");
        rabbitTemplate.convertAndSend("j03s_direct",u);

    }

    /**
     * 分裂模式 - 发送方把数据发送交换器,由交换器把数据路由[转发]到各个队列上.
     * 交换器是不会存储消息[数据]的
     */
    @Test
    public void testFanout(){
        rabbitTemplate.convertAndSend("j03s_fanout","","我是分裂模式!");
    }


    /**
     * 主题模式 - 通用,灵活性更强.
     */
    @Test
    public void testTopic(){
        rabbitTemplate.convertAndSend("j03s_topic","good.success.log","我是主题模式01!");
    }

}

接收方 - 和分裂模式的代码一样

package tech.aistar.handler;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
 * 本类用来演示: 主题模式
 *
 * @author: success
 * @date: 2021/10/9 10:25 上午
 */
@Component
public class RabbitMQTopic {

    @RabbitListener(queues = "top1")
    @RabbitHandler
    public void getE1(String msg, @Headers Map<String,Object> header, Channel channel){
        System.out.println("msg01-"+msg);

        //手动删除rabbitmq队列中的数据的.
        long ack = (long) header.get(AmqpHeaders.DELIVERY_TAG);
        try {
            channel.basicAck(ack,false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @RabbitListener(queues = "top2")
    @RabbitHandler
    public void getTop2(String msg, @Headers Map<String,Object> header, Channel channel){
        System.out.println("msg02-"+msg);

        //手动删除rabbitmq队列中的数据的.
        long ack = (long) header.get(AmqpHeaders.DELIVERY_TAG);
        try {
            channel.basicAck(ack,false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

利用Jmeter工具进行压测

压测 - 压力测试 - 测试系统的吞吐量,响应时间,错误率

下载以及安装步骤 - https://my.oschina.net/niepanLs/blog/3012286

具体操作

新建线程组

设置线程组的参数

添加聚合报告 - 观察吞吐量,响应时间,请求错误率信息 - 优化

添加取样器 - http请求

http请求设置

5-1. 模拟users.csv文件 - 存放若干个用户名

5-2. csv界面信息的设置

5-3. http请求界面的设置信息

SpringBoot整合Druid

导入依赖

<!--        druid连接池-->
<dependency>
  <groupId>com.alibaba</groupId>
  <artifactId>druid-spring-boot-starter</artifactId>
  <version>1.1.10</version>
</dependency>

application.yml文件中进行配置

server:
  port: 8088
spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/j13_test?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8
    username: root
    password: root
    druid:
      # 下面为连接池的补充设置，应用到上面所有数据源中
      # 初始化大小，最小，最大
      initial-size: 5
      min-idle: 5
      max-active: 20
      # 配置获取连接等待超时的时间
      max-wait: 60000
      # 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
      time-between-eviction-runs-millis: 60000
      # 配置一个连接在池中最小生存的时间，单位是毫秒
      min-evictable-idle-time-millis: 300000
      validation-query: SELECT 1 FROM DUAL
      test-while-idle: true
      test-on-borrow: false
      test-on-return: false
      # 打开PSCache，并且指定每个连接上PSCache的大小
      pool-prepared-statements: true
      #   配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
      max-pool-prepared-statement-per-connection-size: 20
      filters: stat,wall
      use-global-data-source-stat: true
      # 通过connectProperties属性来打开mergeSql功能；慢SQL记录
      connect-properties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=500
      stat-view-servlet:
        login-username: admin
        login-password: admin
  rabbitmq:
    host: localhost
    listener:
      simple:
#        设置成手动删除方式
        acknowledge-mode: manual
  mvc:
    view:
      prefix: /WEB-INF/
      suffix: .jsp
  redis:
    host: localhost
    port: 6379
    password: success
    database: 10

配置连接池过滤器

主程序上需要加入@ServletComponentScan//开启servlet注解使用

package tech.aistar.filter;
      
import com.alibaba.druid.support.http.WebStatFilter;
      
import javax.servlet.annotation.WebFilter;
import javax.servlet.annotation.WebInitParam;
      
/**
 * 本类用来演示: 连接池过滤器
 *
 * @author: success
 * @date: 2021/5/13 2:40 下午
 */
@WebFilter(filterName="druidWebStatFilter",urlPatterns="/*",
        initParams={
                @WebInitParam(name="exclusions",value="*.js,*.gif,*.jpg,*.bmp,*.png,*.css,*.ico,/druid/*")//忽略资源
        }
)
public class DruidFilter extends WebStatFilter {
}

查看Druid界面

http://localhost:8088/druid/

SpringBoot任务调度器

https://cron.qqe2.com/

cron在线表达式生成

用来定时执行任务的

在主程序上进行配置

@EnableScheduling
public class SpringbootJ03sApplication {
    public static void main(String[] args) {
        SpringApplication.run(SpringbootJ03sApplication.class, args);
    }
}

具体的任务

package tech.aistar.task;

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

/**
 * 本类用来演示: SpringBoot定时任务调度器
 *
 * @author: success
 * @date: 2021/10/9 1:44 下午
 */
@Component
public class HelloTask {
    /**
     * 每间隔2秒执行一次
     */
    @Scheduled(cron = "0/2 * * * * *" )
    public void testTask01(){
        System.out.println("task01...");
    }
}