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

在这个项目里我用的是 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. 
   
7.        <dependency>
   
8.            <groupId>org.springframework.boot</groupId>
   
9.            <artifactId>spring-boot-starter-test</artifactId>
   
10.            <scope>test</scope>
    
11.        </dependency>
    
12.        <dependency>
    
13.            <groupId>org.springframework.boot</groupId>
    
14.            <artifactId>spring-boot-starter-web</artifactId>
    
15.        </dependency>
    
16.        <dependency>
    
17.            <groupId>mysql</groupId>
    
18.            <artifactId>mysql-connector-java</artifactId>
    
19.        </dependency>
    
20.        <dependency>
    
21.            <groupId>org.springframework.boot</groupId>
    
22.            <artifactId>spring-boot-starter-data-jpa</artifactId>
    
23.        </dependency>
    
24.        <dependency>
    
25.            <groupId>org.projectlombok</groupId>
    
26.            <artifactId>lombok</artifactId>
    
27.            <version>1.16.18</version>
    
28.        </dependency>
    
29.        <dependency>
    
30.            <groupId>org.springframework.boot</groupId>
    
31.            <artifactId>spring-boot-starter-amqp</artifactId>
    
32.        </dependency>
    
33.        <dependency>
    
34.            <groupId>com.alibaba</groupId>
    
35.            <artifactId>fastjson</artifactId>
    
36.            <version>RELEASE</version>
    
37.            <scope>compile</scope>
    
38.        </dependency>
    
39.    </dependencies>
    

1.2 application.properties

1. server.port=10000
   
2. 
   
3. spring.datasource.url=jdbc:mysql://xxxxx/xxxxx?characterEncoding=utf-8
   
4. spring.datasource.username=xxx
   
5. spring.datasource.password=xxxx
   
6. spring.datasource.driver-class-name=com.mysql.jdbc.Driver
   
7. 
   
8. spring.jpa.properties.hibernate.hbm2ddl.auto=update
   
9. spring.jpa.show-sql=true
   
10. 
    
11. spring.rabbitmq.host=localhost
    
12. spring.rabbitmq.username=root
    
13. spring.rabbitmq.password=root
    
14. spring.rabbitmq.port=5672
    

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

1.3 数据表

1. create table ifnot result
   
2. (
   
3.    id int auto_increment primary key,
   
4.    ticket_id intnull,
   
5.    user_id intnull
   
6. );
   
7. 
   
8. create table ifnot exists ticket
   
9. (
   
10.    id int auto_increment primary key,
    
11.    name varchar(255)null,
    
12.    content varchar(255)null,
    
13.    user_name varchar(20)null,
    
14.    count intdefault'6666'notnull
    
15. );
    

根据数据表可以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. publicclass AMQP {
   
5.    publicstaticvoid main(String[] args){
   
6.        SpringApplication.run(AMQP.class, args);
   
7.    }
   
8. }
   

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

3. controller

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

1. @RestController
   
2. @RequestMapping("/ticket")
   
3. publicclassTicketController{
   
4.    @Autowired
   
5.    privateTicketService ticketService;
   
6.    @Autowired
   
7.    privateMQSender mqSender;
   
8. 
   
9.    @RequestMapping("/get/{id}")
   
10.    publicTicket getByid(@PathVariableInteger id){
    
11.        return ticketService.findById(id);
    
12.    }
    
13. 
    
14.    @RequestMapping("/reduce/{id}/{userId}")
    
15.    publicString reduceCount(@PathVariableInteger id,
    
16.                              @PathVariableInteger userId){
    
17.        Message message =newMessage(id,userId);
    
18. 
    
19.        ticketService.reduceCount(id);
    
20.        mqSender.sendMessage(newMessage(message.getTicketId(),message.getUserId()));
    
21.        return"抢票成功!";
    
22.    }
    
23. }
    

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

4. Service

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

4.1 ResultServiceImpl.java

1. @Service
   
2. publicclassResultServiceImplimplementsResultService{
   
3. 
   
4.    @Autowired
   
5.    privateResultRepository resultRepository;
   
6.    @Override
   
7.    publicvoid add(Result result){
   
8.        resultRepository.add(result.getTicketId(), result.getUserId());
   
9. 
   
10.    }
    
11. 
    
12.    @Override
    
13.    publicResult findOneByUserId(Integer userId){
    
14.        return resultRepository.findByUserId(userId);
    
15.    }
    
16. }
    

4.2 TicketServiceImpl.java

1. @Service
   
2. publicclassTicketServiceImplimplementsTicketService{
   
3. 
   
4.    @Autowired
   
5.    privateTicketRepository repository;
   
6.    @Override
   
7.    publicTicket findById(Integer id){
   
8.        return repository.findTicketById(id);
   
9.    }
   
10. 
    
11.    @Override
    
12.    publicTicket reduceCount(Integer id){
    
13.        repository.reduceCount(id);
    
14.        return findById(id);
    
15.    }
    
16. }
    

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

5. Dao

5.1 ResultRepository.java

1. @Repository
   
2. publicinterfaceResultRepositoryextendsJpaRepository<Result,Integer>{
   
3. 
   
4.    @Transactional
   
5.    @Modifying
   
6.    @Query(value ="insert into result(ticket_id,user_id) values(?1,?2) ",nativeQuery =true)
   
7.    void add(@Param("ticketId")Integer ticketId,@Param("userId")Integer userId);
   
8. 
   
9.    Result findByUserId(Integer userId);
   
10. }
    

5.2 TicketRepository.java

1. @Repository
   
2. publicinterfaceTicketRepositoryextendsJpaRepository<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. publicclassMessageimplementsSerializable{
   
3.    privateInteger ticketId;
   
4.    privateInteger userId;
   
5. 
   
6.    publicMessage(){
   
7.    }
   
8.    publicMessage(Integer ticketId,Integer userId){
   
9.        this.ticketId = ticketId;
   
10.        this.userId = userId;
    
11.    }
    
12. }
    

6.2 MQConstants.java

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

1. publicclassMQConstants{
   
2.    publicstaticfinalString QUEUE="qiangpiao";
   
3. }
   

6.3 MQSender.java

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

1. @Service
   
2. @Slf4j
   
3. publicclassMQSender{
   
4.    @Autowired
   
5.    privateAmqpTemplate amqpTemplate;
   
6. 
   
7.    publicvoid sendMessage(Message message){
   
8.        String msg =JSONObject.toJSONString(message);
   
9.        log.info("send message : "+msg);
   
10.        amqpTemplate.convertAndSend(MQConstants.QUEUE,msg);
    
11.    }
    
12. }
    




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

6.4 MQReceiver.java

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

1. @Service
   
2. @Slf4j
   
3. publicclassMQReceiver{
   
4. 
   
5.    @Autowired
   
6.    privateTicketService ticketService;
   
7.    @Autowired
   
8.    privateResultService resultService;
   
9. 
   
10.    @RabbitListener(queues =MQConstants.QUEUE)
    
11.    publicvoid receive(String message){
    
12.        log.info("receive msg : "+message);
    
13.        JSONObject jsonObject =JSONObject.parseObject(message);
    
14.        System.out.println(jsonObject);
    
15.        Message msg =JSONObject.toJavaObject(jsonObject,Message.class);
    
16.        Integer ticketId = msg.getTicketId();
    
17.        Integer userId = msg.getUserId();
    
18.        // 减库存
    
19.        Ticket ticket = ticketService.reduceCount(ticketId);
    
20.        if(ticket.getCount()<=0){
    
21.            return;
    
22.        }
    
23.        // 判断是否已经抢过
    
24.        Result oneByUserId = resultService.findOneByUserId(userId);
    
25.        if(oneByUserId !=null){
    
26.            return;
    
27.        }
    
28.        resultService.add(newResult(ticketId,userId));
    
29.    }
    
30. }
    

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

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

运行效果

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

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

压测报告:

1. ServerSoftware:        
   
2. ServerHostname:        127.0.0.1
   
3. ServerPort:            10000
   
4. 
   
5. DocumentPath:          /ticket/reduce/1/10
   
6. DocumentLength:        13 bytes
   
7. 
   
8. ConcurrencyLevel:      1000
   
9. Time taken for tests:   423.101 seconds
   
10. Complete requests:      9999
    
11. Failed requests:        0
    
12. Total transferred:      1459854 bytes
    
13. HTML transferred:       129987 bytes
    
14. Requests per second:    23.63[#/sec] (mean)
    
15. Time per request:       42314.334[ms](mean)
    
16. Time per request:       42.314[ms](mean, across all concurrent requests)
    
17. Transfer rate:          3.37[Kbytes/sec] received
    
18. 
    
19. ConnectionTimes(ms)
    
20.              min  mean[+/-sd] median   max
    
21. Connect:        0    2   6.8      0      29
    
22. Processing:   217401977390.7  41984   58488
    
23. Waiting:      217401977390.8  41984   58488
    
24. Total:        246401997384.8  41985   58488
    
25. 
    
26. Percentage of the requests served within a certain time (ms)
    
27.  50%  41984
    
28.  66%  42670
    
29.  75%  42744
    
30.  80%  42758
    
31.  90%  42801
    
32.  95%  42828
    
33.  98%  42850
    
34.  99%  42868
    
35. 100%  58488(longest request)
    

注意

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

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

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