查看原文
其他

SpringBoot整合RabbitMQ(二)单机抢票系统

微笑面对生活 Java知音 2020-08-20

点击上方“Java知音”,选择“置顶公众号”

技术文章第一时间送达!


上一章节讲了基本的整合和各种Exchange的使用,这章主要来实现一个单机的简单的抢票系统,麻雀虽小但五脏俱全,为什么用它做抢票系统大家应该也懂,为了削峰和异步处理。

在这个项目里我用的是 springboot的2版本,ORM选用 JPA快速开发,JSON工具使用阿里的 fastjson,当然,mq用的是 rabbitMQ。导入的是 springboot集成的依赖。

1. 配置部分

1.1 pom.xml
  1.    <dependencies>

  2.        <dependency>

  3.            <groupId>org.springframework.boot</groupId>

  4.            <artifactId>spring-boot-starter</artifactId>

  5.        </dependency>


  6.        <dependency>

  7.            <groupId>org.springframework.boot</groupId>

  8.            <artifactId>spring-boot-starter-test</artifactId>

  9.            <scope>test</scope>

  10.        </dependency>

  11.        <dependency>

  12.            <groupId>org.springframework.boot</groupId>

  13.            <artifactId>spring-boot-starter-web</artifactId>

  14.        </dependency>

  15.        <dependency>

  16.            <groupId>mysql</groupId>

  17.            <artifactId>mysql-connector-java</artifactId>

  18.        </dependency>

  19.        <dependency>

  20.            <groupId>org.springframework.boot</groupId>

  21.            <artifactId>spring-boot-starter-data-jpa</artifactId>

  22.        </dependency>

  23.        <dependency>

  24.            <groupId>org.projectlombok</groupId>

  25.            <artifactId>lombok</artifactId>

  26.            <version>1.16.18</version>

  27.        </dependency>

  28.        <dependency>

  29.            <groupId>org.springframework.boot</groupId>

  30.            <artifactId>spring-boot-starter-amqp</artifactId>

  31.        </dependency>

  32.        <dependency>

  33.            <groupId>com.alibaba</groupId>

  34.            <artifactId>fastjson</artifactId>

  35.            <version>RELEASE</version>

  36.            <scope>compile</scope>

  37.        </dependency>

  38.    </dependencies>

1.2 application.properties
  1. server.port=10000


  2. spring.datasource.url=jdbc:mysql://xxxxx/xxxxx?characterEncoding=utf-8

  3. spring.datasource.username=xxx

  4. spring.datasource.password=xxxx

  5. spring.datasource.driver-class-name=com.mysql.jdbc.Driver


  6. spring.jpa.properties.hibernate.hbm2ddl.auto=update

  7. spring.jpa.show-sql=true


  8. spring.rabbitmq.host=localhost

  9. spring.rabbitmq.username=root

  10. spring.rabbitmq.password=root

  11. spring.rabbitmq.port=5672

我只是很有针对性的对 mq和 datasource进行了配置。

1.3 数据表
  1. create table if not result

  2. (

  3.    id int auto_increment primary key,

  4.    ticket_id int null,

  5.    user_id int null

  6. );


  7. create table if not exists ticket

  8. (

  9.    id int auto_increment primary key,

  10.    name varchar(255) null,

  11.    content varchar(255) null,

  12.    user_name varchar(20) null,

  13.    count int default '6666' not null

  14. );

根据数据表可以Generate出JavaBean,不贴JavaBean了。 ##### 1.4 项目架构

  1. ├── src

  2. │   ├── main

  3. │   │   ├── java

  4. │   │   │   └── com

  5. │   │   │       └── fantj

  6. │   │   │           └── springbootjpa

  7. │   │   │               ├── AMQP.java

  8. │   │   │               ├── controller

  9. │   │   │               │   └── TicketController.java

  10. │   │   │               ├── mq

  11. │   │   │               │   ├── Message.java

  12. │   │   │               │   ├── MQConstants.java

  13. │   │   │               │   ├── MQReceiver.java

  14. │   │   │               │   └── MQSender.java

  15. │   │   │               ├── pojo

  16. │   │   │               │   ├── Result.java

  17. │   │   │               │   └── Ticket.java

  18. │   │   │               ├── repostory

  19. │   │   │               │   ├── ResultRepository.java

  20. │   │   │               │   └── TicketRepository.java

  21. │   │   │               └── service

  22. │   │   │                   ├── ResultServiceImpl.java

  23. │   │   │                   ├── ResultService.java

  24. │   │   │                   ├── TicketServiceImpl.java

  25. │   │   │                   └── TicketService.java

  26. │   │   └── resources

  27. │   │       ├── application.properties

  28. │   │       └── rebel.xml

2. 启动类

  1. @SpringBootApplication

  2. @EntityScan("com.fantj.springbootjpa.pojo")

  3. @EnableRabbit

  4. public class AMQP {

  5.    public static void main(String[] args) {

  6.        SpringApplication.run(AMQP.class, args);

  7.    }

  8. }

注意这个 @EnableRabbit注解,它会开启对rabbit注解的支持。

3. controller

很简单的一个controller类,实现查询和抢票功能。

  1. @RestController

  2. @RequestMapping("/ticket")

  3. public class TicketController {

  4.    @Autowired

  5.    private TicketService ticketService;

  6.    @Autowired

  7.    private MQSender mqSender;


  8.    @RequestMapping("/get/{id}")

  9.    public Ticket getByid(@PathVariable Integer id){

  10.        return ticketService.findById(id);

  11.    }


  12.    @RequestMapping("/reduce/{id}/{userId}")

  13.    public String reduceCount(@PathVariable Integer id,

  14.                              @PathVariable Integer userId){

  15.        Message message = new Message(id,userId);


  16.        ticketService.reduceCount(id);

  17.        mqSender.sendMessage(new Message(message.getTicketId(),message.getUserId()));

  18.        return "抢票成功!";

  19.    }

  20. }

注意 privateMQSendermqSender;这是我的 rabbit发送消息的类。

4. Service

接口我就不再这里贴出,直接贴实现类。

4.1 ResultServiceImpl.java
  1. @Service

  2. public class ResultServiceImpl implements ResultService{


  3.    @Autowired

  4.    private ResultRepository resultRepository;

  5.    @Override

  6.    public void add(Result result) {

  7.        resultRepository.add(result.getTicketId(), result.getUserId());


  8.    }


  9.    @Override

  10.    public Result findOneByUserId(Integer userId) {

  11.        return resultRepository.findByUserId(userId);

  12.    }

  13. }

4.2 TicketServiceImpl.java
  1. @Service

  2. public class TicketServiceImpl implements TicketService{


  3.    @Autowired

  4.    private TicketRepository repository;

  5.    @Override

  6.    public Ticket findById(Integer id) {

  7.        return repository.findTicketById(id);

  8.    }


  9.    @Override

  10.    public Ticket reduceCount(Integer id) {

  11.        repository.reduceCount(id);

  12.        return findById(id);

  13.    }

  14. }

这两个都是很普通的service实现类,没有新加入的东西。

5. Dao

5.1 ResultRepository.java
  1. @Repository

  2. public interface ResultRepository extends JpaRepository<Result,Integer> {


  3.    @Transactional

  4.    @Modifying

  5.    @Query(value = "insert into result(ticket_id,user_id) values(?1,?2) ",nativeQuery = true)

  6.    void add(@Param("ticketId") Integer ticketId,@Param("userId") Integer userId);


  7.    Result findByUserId(Integer userId);

  8. }

5.2 TicketRepository.java
  1. @Repository

  2. public interface TicketRepository extends JpaRepository<Ticket,Integer>{

  3.    /**

  4.     *  减少库存

  5.     */

  6.    @Transactional

  7.    @Modifying

  8.    @Query(value = "update ticket t set t.count=t.count+(-1) where id=?1",nativeQuery = true)

  9.    int reduceCount(Integer id);

  10.    /**

  11.     * 查询信息

  12.     */

  13.    Ticket findTicketById(Integer id);

  14. }

到了这里,你会发现,md哪里有用mq的痕迹...

6. MQ

剩下的全是mq的处理。

6.1 Message.java

这个类用来封装mq传输的消息对象,我们使用它来对传输的byte进行编解码,得到我们想要的数据。

  1. @Data

  2. public class Message implements Serializable {

  3.    private Integer ticketId;

  4.    private Integer userId;


  5.    public Message() {

  6.    }

  7.    public Message(Integer ticketId, Integer userId) {

  8.        this.ticketId = ticketId;

  9.        this.userId = userId;

  10.    }

  11. }

6.2 MQConstants.java

这是一个常量类,用来定义和保存 queue的名字,虽然里面只有一个常量,好习惯要从小事做起。

  1. public class MQConstants {

  2.    public static final String QUEUE= "qiangpiao";

  3. }

6.3 MQSender.java

这是消息发送类,用来给queue发送数据。

  1. @Service

  2. @Slf4j

  3. public class MQSender {

  4.    @Autowired

  5.    private AmqpTemplate amqpTemplate;


  6.    public void sendMessage(Message message){

  7.        String msg = JSONObject.toJSONString(message);

  8.        log.info("send message : "+msg);

  9.        amqpTemplate.convertAndSend(MQConstants.QUEUE,msg);

  10.    }

  11. }

AmqpTemplate是springboot框架提供给我们使用的amqp操作模板,利用它我们能更方便的调用和处理业务。 我们在Controller层调用它,来完成消息入队的操作,完成削峰和异步处理,大大增加了系统并发和强健性。

6.4 MQReceiver.java

这是消息接收类,用来从queue里获取数据。

  1. @Service

  2. @Slf4j

  3. public class MQReceiver {


  4.    @Autowired

  5.    private TicketService ticketService;

  6.    @Autowired

  7.    private ResultService resultService;


  8.    @RabbitListener(queues = MQConstants.QUEUE)

  9.    public void receive(String message){

  10.        log.info("receive msg : "+message);

  11.        JSONObject jsonObject = JSONObject.parseObject(message);

  12.        System.out.println(jsonObject);

  13.        Message msg = JSONObject.toJavaObject(jsonObject, Message.class);

  14.        Integer ticketId = msg.getTicketId();

  15.        Integer userId = msg.getUserId();

  16.        // 减库存

  17.        Ticket ticket = ticketService.reduceCount(ticketId);

  18.        if (ticket.getCount() <= 0){

  19.            return;

  20.        }

  21.        // 判断是否已经抢过

  22.        Result oneByUserId = resultService.findOneByUserId(userId);

  23.        if (oneByUserId != null){

  24.            return;

  25.        }

  26.        resultService.add(new Result(ticketId,userId));

  27.    }

  28. }

在这个类中, @RabbitListener(queues=MQConstants.QUEUE)标记的是监听方法,该方法会从queue中获取到String数据。

之后我们需要将其复原为JavaBean,取出我们该要的属性,继续处理业务: 查询票剩余量-> 判断是否已抢到过-> 减库存 -> 增加抢票数据。 (我这里写的有点草率,应该先查余量...,不过不重要,本章重点在过一遍springboot与rabbitmq的整合)。

运行效果

我对该抢票功能做了一个9999请求,我本来做3k并发,电脑没那么多句柄,实现不了,最后做了1k并发的压测。

这是rabbitMQ 自带Managerment模板上的截图:

压测报告:

  1. Server Software:        

  2. Server Hostname:        127.0.0.1

  3. Server Port:            10000


  4. Document Path:          /ticket/reduce/1/10

  5. Document Length:        13 bytes


  6. Concurrency Level:      1000

  7. Time taken for tests:   423.101 seconds

  8. Complete requests:      9999

  9. Failed requests:        0

  10. Total transferred:      1459854 bytes

  11. HTML transferred:       129987 bytes

  12. Requests per second:    23.63 [#/sec] (mean)

  13. Time per request:       42314.334 [ms] (mean)

  14. Time per request:       42.314 [ms] (mean, across all concurrent requests)

  15. Transfer rate:          3.37 [Kbytes/sec] received


  16. Connection Times (ms)

  17.              min  mean[+/-sd] median   max

  18. Connect:        0    2   6.8      0      29

  19. Processing:   217 40197 7390.7  41984   58488

  20. Waiting:      217 40197 7390.8  41984   58488

  21. Total:        246 40199 7384.8  41985   58488


  22. Percentage of the requests served within a certain time (ms)

  23.  50%  41984

  24.  66%  42670

  25.  75%  42744

  26.  80%  42758

  27.  90%  42801

  28.  95%  42828

  29.  98%  42850

  30.  99%  42868

  31. 100%  58488 (longest request)

注意

  1. 本项目没有考虑线程安全的问题,事实上线程是不安全的,线程安全问题后面会说。

  2. 本项目只是为了mq的削峰和异步处理,最直观的就是数据库可以称住高并发,一般来讲,数据库连接这块是称不住的。

  3. mq在分布式下的问题后面会说。



如果你觉得文章不错,欢迎点赞分享到朋友圈

    您可能也对以下帖子感兴趣

    文章有问题?点此查看未经处理的缓存