投稿
  1. Java入门教程_springboot基础教程_springcloud_jvm学习交流网站【Java实验室】首页
  2. SprintBoot课程

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

超, 哥 • • SprintBoot课程 • 阅读 10833

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

1. 配置部分

1.1 pom.xml
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
 
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jpa</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.16.18</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>RELEASE</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>

        <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.16.18</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>RELEASE</version>
<scope>compile</scope>
</dependency>
</dependencies>

1.2 application.properties
1
2
3
4
5
6
7
8
9
10
11
12
13
14

server.port=10000
 
spring.datasource.url=jdbc:mysql://xxxxx/xxxxx?characterEncoding=utf-8
spring.datasource.username=xxx
spring.datasource.password=xxxx
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
 
spring.jpa.properties.hibernate.hbm2ddl.auto=update
spring.jpa.show-sql=true
 
spring.rabbitmq.host=localhost
spring.rabbitmq.username=root
spring.rabbitmq.password=root
spring.rabbitmq.port=5672

spring.datasource.url=jdbc:mysql://xxxxx/xxxxx?characterEncoding=utf-8
spring.datasource.username=xxx
spring.datasource.password=xxxx
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

spring.jpa.properties.hibernate.hbm2ddl.auto=update
spring.jpa.show-sql=true

spring.rabbitmq.host=localhost
spring.rabbitmq.username=root
spring.rabbitmq.password=root
spring.rabbitmq.port=5672

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

1.3 数据表
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

create table if not result
(
    id int auto_increment primary key,
    ticket_id int null,
    user_id int null
);
 
create table if not exists ticket
(
    id int auto_increment primary key,
    name varchar(255) null,
    content varchar(255) null,
    user_name varchar(20) null,
    count int default '6666' not null
);

create table if not exists ticket
(
id int auto_increment primary key,
name varchar(255) null,
content varchar(255) null,
user_name varchar(20) null,
count int default '6666' not null
);

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

├── src
│   ├── main
│   │   ├── java
│   │   │   └── com
│   │   │       └── fantj
│   │   │           └── springbootjpa
│   │   │               ├── AMQP.java
│   │   │               ├── controller
│   │   │               │   └── TicketController.java
│   │   │               ├── mq
│   │   │               │   ├── Message.java
│   │   │               │   ├── MQConstants.java
│   │   │               │   ├── MQReceiver.java
│   │   │               │   └── MQSender.java
│   │   │               ├── pojo
│   │   │               │   ├── Result.java
│   │   │               │   └── Ticket.java
│   │   │               ├── repostory
│   │   │               │   ├── ResultRepository.java
│   │   │               │   └── TicketRepository.java
│   │   │               └── service
│   │   │                   ├── ResultServiceImpl.java
│   │   │                   ├── ResultService.java
│   │   │                   ├── TicketServiceImpl.java
│   │   │                   └── TicketService.java
│   │   └── resources
│   │       ├── application.properties
│   │       └── rebel.xml

2. 启动类

1
2
3
4
5
6
7
8

@SpringBootApplication
@EntityScan("com.fantj.springbootjpa.pojo")
@EnableRabbit
public class AMQP {
    public static void main(String[] args) {
        SpringApplication.run(AMQP.class, args);
    }
}

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

3. controller

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

@RestController
@RequestMapping("/ticket")
public class TicketController {
    @Autowired
    private TicketService ticketService;
    @Autowired
    private MQSender mqSender;
 
    @RequestMapping("/get/{id}")
    public Ticket getByid(@PathVariable Integer id){
        return ticketService.findById(id);
    }
 
    @RequestMapping("/reduce/{id}/{userId}")
    public String reduceCount(@PathVariable Integer id,
                              @PathVariable Integer userId){
        Message message = new Message(id,userId);
 
        ticketService.reduceCount(id);
        mqSender.sendMessage(new Message(message.getTicketId(),message.getUserId()));
        return "抢票成功!";
    }
}

    @RequestMapping("/get/{id}")
public Ticket getByid(@PathVariable Integer id){
return ticketService.findById(id);
}

@RequestMapping("/reduce/{id}/{userId}")
public String reduceCount(@PathVariable Integer id,
@PathVariable Integer userId){
Message message = new Message(id,userId);

ticketService.reduceCount(id);
mqSender.sendMessage(new Message(message.getTicketId(),message.getUserId()));
return "抢票成功!";
}
}

注意private MQSender mqSender;这是我的rabbit发送消息的类。

4. Service

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

4.1 ResultServiceImpl.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

@Service
public class ResultServiceImpl implements ResultService{
 
    @Autowired
    private ResultRepository resultRepository;
    @Override
    public void add(Result result) {
        resultRepository.add(result.getTicketId(), result.getUserId());
 
    }
 
    @Override
    public Result findOneByUserId(Integer userId) {
        return resultRepository.findByUserId(userId);
    }
}

    @Autowired
private ResultRepository resultRepository;
@Override
public void add(Result result) {
resultRepository.add(result.getTicketId(), result.getUserId());

}

@Override
public Result findOneByUserId(Integer userId) {
return resultRepository.findByUserId(userId);
}
}

4.2 TicketServiceImpl.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

@Service
public class TicketServiceImpl implements TicketService{
 
    @Autowired
    private TicketRepository repository;
    @Override
    public Ticket findById(Integer id) {
        return repository.findTicketById(id);
    }
 
    @Override
    public Ticket reduceCount(Integer id) {
        repository.reduceCount(id);
        return findById(id);
    }
}

    @Autowired
private TicketRepository repository;
@Override
public Ticket findById(Integer id) {
return repository.findTicketById(id);
}

@Override
public Ticket reduceCount(Integer id) {
repository.reduceCount(id);
return findById(id);
}
}

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

5. Dao

5.1 ResultRepository.java
1
2
3
4
5
6
7
8
9
10

@Repository
public interface ResultRepository extends JpaRepository<Result,Integer> {
 
    @Transactional
    @Modifying
    @Query(value = "insert into result(ticket_id,user_id) values(?1,?2) ",nativeQuery = true)
    void add(@Param("ticketId") Integer ticketId,@Param("userId") Integer userId);
 
    Result findByUserId(Integer userId);
}

    @Transactional
@Modifying
@Query(value = "insert into result(ticket_id,user_id) values(?1,?2) ",nativeQuery = true)
void add(@Param("ticketId") Integer ticketId,@Param("userId") Integer userId);

Result findByUserId(Integer userId);
}

5.2 TicketRepository.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14

@Repository
public interface TicketRepository extends JpaRepository<Ticket,Integer>{
    /**
     *  减少库存
     */
    @Transactional
    @Modifying
    @Query(value = "update ticket t set t.count=t.count+(-1) where id=?1",nativeQuery = true)
    int reduceCount(Integer id);
    /**
     * 查询信息
     */
    Ticket findTicketById(Integer id);
}

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

6. MQ

剩下的全是mq的处理。

6.1 Message.java

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

1
2
3
4
5
6
7
8
9
10
11
12

@Data
public class Message implements Serializable {
    private Integer ticketId;
    private Integer userId;
 
    public Message() {
    }
    public Message(Integer ticketId, Integer userId) {
        this.ticketId = ticketId;
        this.userId = userId;
    }
}

    public Message() {
}
public Message(Integer ticketId, Integer userId) {
this.ticketId = ticketId;
this.userId = userId;
}
}

6.2 MQConstants.java

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

1
2
3

public class MQConstants {
    public static final String QUEUE= "qiangpiao";
}
6.3 MQSender.java

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

1
2
3
4
5
6
7
8
9
10
11
12

@Service
@Slf4j
public class MQSender {
    @Autowired
    private AmqpTemplate amqpTemplate;
 
    public void sendMessage(Message message){
        String msg = JSONObject.toJSONString(message);
        log.info("send message : "+msg);
        amqpTemplate.convertAndSend(MQConstants.QUEUE,msg);
    }
}

    public void sendMessage(Message message){
String msg = JSONObject.toJSONString(message);
log.info("send message : "+msg);
amqpTemplate.convertAndSend(MQConstants.QUEUE,msg);
}
}

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

6.4 MQReceiver.java

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

@Service
@Slf4j
public class MQReceiver {
 
    @Autowired
    private TicketService ticketService;
    @Autowired
    private ResultService resultService;
 
    @RabbitListener(queues = MQConstants.QUEUE)
    public void receive(String message){
        log.info("receive msg : "+message);
        JSONObject jsonObject = JSONObject.parseObject(message);
        System.out.println(jsonObject);
        Message msg = JSONObject.toJavaObject(jsonObject, Message.class);
        Integer ticketId = msg.getTicketId();
        Integer userId = msg.getUserId();
        // 减库存
        Ticket ticket = ticketService.reduceCount(ticketId);
        if (ticket.getCount() <= 0){
            return;
        }
        // 判断是否已经抢过
        Result oneByUserId = resultService.findOneByUserId(userId);
        if (oneByUserId != null){
            return;
        }
        resultService.add(new Result(ticketId,userId));
    }
}

    @Autowired
private TicketService ticketService;
@Autowired
private ResultService resultService;

@RabbitListener(queues = MQConstants.QUEUE)
public void receive(String message){
log.info("receive msg : "+message);
JSONObject jsonObject = JSONObject.parseObject(message);
System.out.println(jsonObject);
Message msg = JSONObject.toJavaObject(jsonObject, Message.class);
Integer ticketId = msg.getTicketId();
Integer userId = msg.getUserId();
// 减库存
Ticket ticket = ticketService.reduceCount(ticketId);
if (ticket.getCount() <= 0){
return;
}
// 判断是否已经抢过
Result oneByUserId = resultService.findOneByUserId(userId);
if (oneByUserId != null){
return;
}
resultService.add(new Result(ticketId,userId));
}
}

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

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

运行效果

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

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

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

压测报告:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

Server Software:        
Server Hostname:        127.0.0.1
Server Port:            10000
 
Document Path:          /ticket/reduce/1/10
Document Length:        13 bytes
 
Concurrency Level:      1000
Time taken for tests:   423.101 seconds
Complete requests:      9999
Failed requests:        0
Total transferred:      1459854 bytes
HTML transferred:       129987 bytes
Requests per second:    23.63 [#/sec] (mean)
Time per request:       42314.334 [ms] (mean)
Time per request:       42.314 [ms] (mean, across all concurrent requests)
Transfer rate:          3.37 [Kbytes/sec] received
 
Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    2   6.8      0      29
Processing:   217 40197 7390.7  41984   58488
Waiting:      217 40197 7390.8  41984   58488
Total:        246 40199 7384.8  41985   58488
 
Percentage of the requests served within a certain time (ms)
  50%  41984
  66%  42670
  75%  42744
  80%  42758
  90%  42801
  95%  42828
  98%  42850
  99%  42868
 100%  58488 (longest request)

Document Path:          /ticket/reduce/1/10
Document Length:        13 bytes

Concurrency Level:      1000
Time taken for tests:   423.101 seconds
Complete requests:      9999
Failed requests:        0
Total transferred:      1459854 bytes
HTML transferred:       129987 bytes
Requests per second:    23.63 [#/sec] (mean)
Time per request:       42314.334 [ms] (mean)
Time per request:       42.314 [ms] (mean, across all concurrent requests)
Transfer rate:          3.37 [Kbytes/sec] received

Connection Times (ms)
min  mean[+/-sd] median   max
Connect:        0    2   6.8      0      29
Processing:   217 40197 7390.7  41984   58488
Waiting:      217 40197 7390.8  41984   58488
Total:        246 40199 7384.8  41985   58488

Percentage of the requests served within a certain time (ms)
50%  41984
66%  42670
75%  42744
80%  42758
90%  42801
95%  42828
98%  42850
99%  42868
100%  58488 (longest request)

注意

  1. 本项目没有考虑线程安全的问题，事实上线程是不安全的，线程安全问题后面会说。
  2. 本项目只是为了mq的削峰和异步处理，最直观的就是数据库可以称住高并发,一般来讲，数据库连接这块是称不住的。
  3. mq在分布式下的问题后面会说。
赞 (0)
超, 哥超, 哥管理员
0 0
上一篇 2021年1月26日 pm3:17
下一篇 2021年1月26日 pm3:22

相关推荐

发表评论

您的电子邮箱地址不会被公开。 必填项已用*标注

记住昵称、邮箱和网址，下次评论免输入
关注我们