SpringBoot整合RabbitMQ，实现单机抢票系统

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。MQ是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取队列中的消息。

消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC的调用等等。

RabbitMQ介绍

RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

四种交换机(Exchange)

交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct, topic, Headers and Fanout

1. Direct Exchange

direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routingkey, 消息的routingkey 匹配时, 才会被交换器投送到绑定的队列中去.是RabbitMQ默认的交换机模式，也是最简单的模式，根据key全文匹配去寻找队列。

配置：设置一个路由键

 public  static  final String QUEUE="queue";

   /**

    * direct 交换机模式

    */

   @Bean

   public Queue queue(){

       return new Queue(QUEUE,true);

   }

发送服务：

@Service

@Slf4j

public class MQSender {



   @Autowired

   AmqpTemplate amqpTemplate;



   public void send(Object message){

       String msg = (String) message;

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

       amqpTemplate.convertAndSend(MQConfig.QUEUE,msg);

   }

}

接收服务：

@Service

@Slf4j

public class MQReceiver {



   //监听的queue

   @RabbitListener(queues = MQConfig.QUEUE)

   public void receive(String msg){

       log.info("receive msg "+msg);

   }

}

测试：

 @Autowired

   private MQSender sender;



   sender.send("hello direct Exchange");

2. Topic Exchange

按规则转发消息（最灵活）转发消息主要是根据通配符。在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。

路由键必须是一串字符，用句号（.）隔开，

路由模式必须包含一个星号（*），主要用于匹配路由键指定位置的一个单词，井号（#）就表示相当于一个或者多个单词

配置类：

 public  static  final String TOPIC_QUEUE1="topic.queue1";

   public  static  final String TOPIC_QUEUE2="topic.queue2";

   public  static  final String ROUTING_KEY1="topic.key1";

   public  static  final String ROUTING_KEY2="topic.#";

   /**

    * Topic 交换机模式  可以用通配符

    */

   @Bean

   public Queue topicQueue1(){

       return new Queue(TOPIC_QUEUE1,true);

   }

   @Bean

   public Queue topicQueue2(){

       return new Queue(TOPIC_QUEUE2,true);

   }



   @Bean

   public TopicExchange topicExchange(){

       return new TopicExchange(TOPIC_EXCHANGE);

   }

   @Bean

   public Binding topicBinding1(){

       return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with(ROUTING_KEY1);

   }

   @Bean

   public Binding topicBinding2(){

       return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with(ROUTING_KEY2);

   }

发送类：

 public void sendTopic(Object message){

       String msg = (String) message;

       log.info("send topic message"+msg);

       amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key1",msg+"1");

       amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE,"topic.key2",msg+"2");

   }

接收类：

@RabbitListener(queues = MQConfig.TOPIC_QUEUE1)

   public void receiveTopic1(String msg){

       log.info("receive topic1 msg "+msg);

   }

测试：

@Autowired

   private MQSender sender;



   sender.sendTopic("hello topic Exchange");

3. Headers Exchange

设置header attribute参数类型的交换机，相较于 direct 和 topic 固定地使用 routing_key , headers 则是一个自定义匹配规则的类型. 在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列.

  public static final String HEADER_EXCHANGE="headerExchange";

   /**

    * Header 交换机模式

    */

   @Bean

   public HeadersExchange headersExchange(){

       return new HeadersExchange(HEADER_EXCHANGE);

   }

   @Bean

   public Queue headerQueue(){

       return new Queue(HEADER_QUEUE2,true);

   }

   // 绑定需要指定header，如果不匹配 则不能使用

   @Bean

   public Binding headerBinding(){

       Map<String,Object> map = new HashMap();

       map.put("header1","value1");

       map.put("header2","value2");

       return BindingBuilder.bind(headerQueue()).to(headersExchange()).whereAll(map).match();

   }

 public void sendHeader(Object massage){

       String msg = (String) massage;

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

       MessageProperties properties = new MessageProperties();

       properties.setHeader("header1","value1");

       properties.setHeader("header2","value2");

       Message obj = new Message(msg.getBytes(),properties);

       amqpTemplate.convertAndSend(MQConfig.HEADER_EXCHANGE,"",obj);

   }

用MessageProperties来添加Header信息，然后与接收者的header比对。我都设置的是"header1","value1"；"header2","value2"

 //监听 header模式的queue

   @RabbitListener(queues = MQConfig.HEADER_QUEUE2)

   //因为发送的是 byte 类型，所以接受也是该数据类型

   public void receiveHeader(byte[] message){

       log.info("header queue message"+new String(message));

   }

测试：

 @Autowired

   private MQSender sender;



   sender.sendHeader("hello header Exchange");

4. Fanout Exchange

转发消息到所有绑定队列，消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了routing_key会被忽略。

 public static final String FANOUT_EXCHANGE="fanoutExchange";

   /**

    * Fanout 交换机模式（广播模式）,不用绑定key

    */

   @Bean

   public FanoutExchange fanoutExchange(){

       return new FanoutExchange(FANOUT_EXCHANGE);

   }

   @Bean

   public Binding fanoutBinding1(){

       return BindingBuilder.bind(topicQueue1()).to(fanoutExchange());

   }

   @Bean

   public Binding fanoutBinding2(){

       return BindingBuilder.bind(topicQueue2()).to(fanoutExchange());

   }

 public void sendFanout(Object massage){

       String msg = (String) massage;

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

       amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE,"",msg);

   }

测试：

 @Autowired

   private MQSender sender;



   sender.sendFanout("hello fanout Exchange");

补充

这个示例是基于springboot的示例。

pom依赖

<!--rabbbitMQ相关依赖-->

<dependency>

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

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

</dependency>

配置文件

 rabbitmq:

   host: 127.0.0.1

   port: 5672

   username: guest

   password: guest

   # 这个账号密码只能连接本地的mq，远程的话需要配置

   virtual-host: /

   listener:

     simple:

       concurrency: 10

       max-concurrency: 10

       prefetch: 1 # 从队列每次取一个

       auto-startup: true

       default-requeue-rejected: true # 失败后重试

实现单机抢票系统

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

1. 配置部分

1.1 pom.xml

  <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>

1.2 application.properties

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

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

1.3 数据表

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

);

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

├── 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. 启动类

@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类，实现查询和抢票功能。

@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 "抢票成功!";

   }

}

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

4. Service

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

4.1 ResultServiceImpl.java

@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);

   }

}

4.2 TicketServiceImpl.java

@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);

   }

}

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

5. Dao

5.1 ResultRepository.java

@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);

}

5.2 TicketRepository.java


@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进行编解码，得到我们想要的数据。

@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;

   }

}

6.2 MQConstants.java

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

public class MQConstants {

   public static final String QUEUE= "qiangpiao";

}

6.3 MQSender.java

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

@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);

   }

}

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

6.4 MQReceiver.java

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

@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));

   }

}

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

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

运行效果

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

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

压测报告:

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)