软件开发架构师

【SpringBoot系列5】SpringBoot整合RabbitMQ

java 71 2019-02-25 15:18

前言:

因为项目需要用到RabbitMQ,前几天就看了看RabbitMQ的知识,记录下SpringBoot整合RabbitMQ的过程。

给出两个网址:

RabbitMQ官方教程:http://www.rabbitmq.com/getstarted.html

SpringBoot整个RabbitMQ教程:https://docs.spring.io/spring-boot/docs/2.0.1.RELEASE/reference/htmlsingle/#boot-features-amqp

正文:

1:最简单的HelloWorld

首先加上SpringBoot的依赖

1 <!-- rabbitmq -->
2 <dependency>
3     <groupId>org.springframework.boot</groupId>
4     <artifactId>spring-boot-starter-amqp</artifactId>
5 </dependency>

其实我们要申明一个队列,这里队列存放生产者生产的消息,然后消费者来这里面取出来

 1 @Configuration
 2 public class RabbitBeanConf {
 3     
 4     public static final String QUEUE = "queue";
 5     
 6     @Bean
 7     public Queue queue() {
 8         return new Queue(QUEUE, true);
 9     }
10 }

生产者:

 1 @Service
 2 public class RabbitSender {
 3 
 4     @Autowired
 5     AmqpTemplate amqpTemplate;
 6     
 7     public void send(String msg) {
 8         amqpTemplate.convertAndSend(RabbitBeanConf.QUEUE, msg);
 9         System.out.println("生产者生产了一个消息: " + msg + "  " + new Date().getTime());
10     }
11 }

消费者:

1 @Service
2 public class RabbitReceiver {
3 
4     @RabbitListener(queues = RabbitBeanConf.QUEUE)
5     public void receive(String msg) {
6         
7         System.out.println("消费者收到了一个消息: " + msg + "  " + new Date().getTime());
8     }
9 }

 

2:广播模式(Fanout模式)

这个模式不同于上面的模式,这个模式只要一个生产者生产出一个消息,那么所有消费者全部可以接收到。

最简单一个例子就是你微信群发消息的时候,你发了一条消息,但你的好友全部都能收到这条消息

RabbitMQ配置:

 1 @Configuration
 2 public class FanoutRabbitMQConfig {
 3 
 4     //默认持久化durable为true
 5     //根据方法名来进行绑定的 firstFanoutQueue()
 6     @Bean
 7     public Queue firstFanoutQueue() {
 8         return new Queue("firstFanoutQueue");
 9     }
10     
11     @Bean
12     public Queue secondFanoutQueue() {
13         return new Queue("secondFanoutQueue");
14     }
15     
16     //默认持久化durbale为true
17     //exchange交换机
18     @Bean
19     public FanoutExchange fanoutExchange() {
20         return new FanoutExchange("fanoutExchange");
21     }
22     
23     /**
24      * 以下两种方法都可以成功
25      * 
26      * @return
27      * 2018年6月11日
28      */
29     @Bean
30     public Binding bindingFirst() {
31         return BindingBuilder.bind(firstFanoutQueue()).to(fanoutExchange());
32     }
33     
34     @Bean
35     public Binding bindingSecond(FanoutExchange fanoutExchange, Queue secondFanoutQueue) {
36         return BindingBuilder.bind(secondFanoutQueue).to(fanoutExchange);
37     }
38 }

这里我解释一下,官方引入了一个叫做 “exchange”交换机的东西。

我是这样理解的:这个交换机主要是为了解耦生产了和消费者直接的强联系。生产者不直接往队列里面送东西了,而且往exchange里面送东西,然后exchange绑定想要送消息进去的队列,然后消费者监听即可。主要解耦。

生产者:

 1 @Service
 2 public class FanoutSender {
 3     
 4     @Autowired
 5     AmqpTemplate amqpTemplate;
 6     
 7     public void send(String msg) {
 8         
 9         amqpTemplate.convertAndSend("fanoutExchange", "", msg);
10         System.out.println("生产者生产了一个消息: " + msg + "  " + new Date().getTime());
11     }
12 }

消费者:

1 @Service
2 public class FanoutRecevi {
3 
4     @RabbitListener(queues = {"firstFanoutQueue", "secondFanoutQueue"})
5     public void receive(String msg) {
6         
7         System.out.println("消费者收到了一个消息: " + msg + "  " + new Date().getTime());
8     }
9 }

 

3:Topic模式

感觉和直接模式(direct模式)有点相像。只不过这里用了类似正则的东西,一个exchange匹配符合规则的队列。

topic 和 direct 类似, 只是匹配上支持了"模式", 在"点分"的 routing_key 形式中, 可以使用两个通配符:

  • *表示一个词.
  • #表示零个或多个词.

 

4:Header模式

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

 

文章评论