RabbitMQ02-RebbitMQ简介及交换器

一. AMQP协议

什么是AMQP协议

AMQP(Advanced Message Queuing Protocol,高级消息队列协议):它是进程之间传递异步消息的网络协议

AMQP工作过程

发布者通过发布消息，通过交换机，交换机根据路由规则将收到的消息分发交换机绑定的下消息队列，最后AMQP代理将消息推送给订阅了此队列的消费者
或消费者按照需求自行获取。

二. RabbitMQ简介

RabbitMQ是通过Erlang语言基于AMQP协议编写的消息中间件，它在分布式系统中可以解应用耦合、流量削峰、异步消息等问题。它有两个特性
队列排队和异步

1. 应用解耦：多个个应用程序之间可通过RabbitMQ作为媒介，两个应用不再粘连，实现解耦；
2. 异步消息：多个应用可通过RabbitMQ进行消息传递；
3. 流量削峰：在高并发情况下，可以通过RabbitMQ的队列特性实现流量削峰；
4. 应用场景：
   1. 应用到队列特性的应用场景： 排序算法、秒杀活动。
   2. 应用到异步特性的应用场景： 消息分发、异步处理、数据同步、处理耗时任务。

三.springBoot整合RabbitMQ

生产者端发送消息

pom文件

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

yml文件

spring:
  application:
    name: producer
  rabbitmq:
    host: xxx
    username: admin
    password: admin
1
2
3
4
5
6
7

配置类,需要返回一个Queue,org.springframework.amqp.core.Queue下的Queue对象

@Configuration
public class RabbitMqConfig {

    @Bean
    protected Queue queue(){
        return new Queue("myQueue");
    }
}
1
2
3
4
5
6
7
8

使用RabbitMQ发送消息，注入AmqpTemplate,调用convertAndSend()方法

class ProducerApplicationTests {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    void send() {
        for (int i = 0; i < 10; i++) {
            amqpTemplate.convertAndSend("myQueue","这是发送的消息");
            System.out.println("发送成功！");
        }

    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

消费端接收消息

配置同生产端，不需要配置RabbitMqConfig，接收消息时只需要使用注解RabbitMqConfig，queues属性绑定相应的队列即可。

@Component
public class ReceiveService {

    @RabbitListener(queues = "myQueue")
    public void test01(String msg){
        System.out.println("接收到消息1" + msg);
    }

    @RabbitListener(queues = "myQueue")
    public void test02(String msg){
        System.out.println("接收到消息2" + msg);
    }

    @RabbitListener(queues = "myQueue")
    public void test03(String msg){
        System.out.println("接收到消息3" + msg);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

四.交换器（四种)

Direct Exchange:直连交换器

它是RabbitMQ的默认交换器，给指定队列发消息，绑定该消息队列的消费者一次获取消息

实战：

/** 生产者发送消息，发送10个消息*/
@SpringBootTest
class ProducerApplicationTests {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @Test
    void send() {
        for (int i = 0; i < 10; i++) {
            amqpTemplate.convertAndSend("myQueue","这是发送的消息");
            System.out.println("发送成功！");
        }

    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

/** 接收消息*/
@Component
public class ReceiveService {

    @RabbitListener(queues = "myQueue")
    public void test01(String msg){
        System.out.println("接收到消息1" + msg);
    }

    @RabbitListener(queues = "myQueue")
    public void test02(String msg){
        System.out.println("接收到消息2" + msg);
    }

    @RabbitListener(queues = "myQueue")
    public void test03(String msg){
        System.out.println("接收到消息3" + msg);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

结果：可以看到1、2、3依次接收消息

接收到消息1这是发送的消息
接收到消息2这是发送的消息
接收到消息3这是发送的消息
接收到消息2这是发送的消息
接收到消息3这是发送的消息
接收到消息1这是发送的消息
接收到消息3这是发送的消息
接收到消息1这是发送的消息
接收到消息2这是发送的消息
接收到消息1这是发送的消息
1
2
3
4
5
6
7
8
9
10

Fanout Exchange:扇形交换器

绑定该交换器的所有队列都可以接收到消息，扇形交换机将消息广播到所有与之绑定的队列。无论消息的路由键是什么，扇形交换机都会将消息发送到所有绑定的队列中。这种类型的交换机常用于实现发布-订阅模式，将消息广播给多个消费者。

实战

/** 绑定*/
/** Fanout Exchange*/
@Bean
public Queue FanoutExchangeQueue1(){
        return new Queue("fanoutExchangeQueue1");
        }
@Bean
public Queue FanoutExchangeQueue2(){
        return new Queue("fanoutExchangeQueue2");
        }
@Bean
public FanoutExchange fanoutExchange(){
        return new FanoutExchange("amq.fanout");
        }
@Bean
public Binding  FanoutExchangeBinding1(Queue FanoutExchangeQueue1,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(FanoutExchangeQueue1).to(fanoutExchange);
        }
@Bean
public Binding  FanoutExchangeBinding2(Queue FanoutExchangeQueue2,FanoutExchange fanoutExchange){
        return BindingBuilder.bind(FanoutExchangeQueue2).to(fanoutExchange);
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

/** 生产者发送消息*/
    @Test
    void sendByFanoutExchange() {
        amqpTemplate.convertAndSend("amq.fanout","key","这是发送到的消息");
        System.out.println("发送成功！");
    }
1
2
3
4
5
6

    /** 消费者 Direct Exchange*/
    @RabbitListener(queues = "fanoutExchangeQueue1")
    public void test04(String msg){
        System.out.println("接收到消息4" + msg);
    }
    @RabbitListener(queues = "fanoutExchangeQueue2")
    public void test05(String msg){
        System.out.println("接收到消息5" + msg);
    }
1
2
3
4
5
6
7
8
9

结果：每一个绑定到Fanout Exchange上的队列都可以接收到消息

接收到消息4这是发送到的消息
接收到消息5这是发送到的消息
1
2

Topic Exchange:主题交换器

允许在路由键中设置匹配规则：'*‘代表一个字母两个’.'之间的内容；‘#’代表0或多个字符；

实战

    /** 绑定*/
    @Bean
    public Queue topicExchangeQueue1(){
            return new Queue("topicExchangeQueue1");
            }
    @Bean
    public Queue topicExchangeQueue2(){
            return new Queue("topicExchangeQueue2");
            }
    @Bean
    public TopicExchange topicExchange(){
            return new TopicExchange("amq.topic");
            }
    @Bean
    public Binding TopicExchangeToQueue1(Queue topicExchangeQueue1,TopicExchange topicExchange){
            return BindingBuilder.bind(topicExchangeQueue1).to(topicExchange).with("com.shaoby.*");
            }
    @Bean
    public Binding TopicExchangeToQueue2(Queue topicExchangeQueue2,TopicExchange topicExchange){
            return BindingBuilder.bind(topicExchangeQueue2).to(topicExchange).with("com.shaoby.test.#");
    
            }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

    /**生产者发送消息*/
    /** key为com.shaoby.test*/
    @Test
    void sendByTopicExchange() {
            amqpTemplate.convertAndSend("amq.topic","com.shaoby.test","这是发送到的消息");
            System.out.println("发送成功！");
    }
    /** key为com.shaoby.test.a*/
    @Test
    void sendByTopicExchange() {
            amqpTemplate.convertAndSend("amq.topic","com.shaoby.test.a.b","这是发送到的消息");
            System.out.println("发送成功！");
            }
1
2
3
4
5
6
7
8
9
10
11
12
13

    /**消费者接收消息*/
    /**Topic Exchange*/
    @RabbitListener(queues = "topicExchangeQueue1")
    public void test06(String msg){
        System.out.println("接收到消息6" + msg);
    }

    @RabbitListener(queues = "topicExchangeQueue2")
    public void test07(String msg){
        System.out.println("接收到消息7" + msg);
    }
1
2
3
4
5
6
7
8
9
10
11

结果：

路由key为com.shaoby.test都能接收到消息，com.shaoby.test.a.b只有topicExchangeQueue2能接收到消息

Header Exchange:首部交换器

绑定：

/** Header Exchange*/
@Bean
public Queue headerExchangeQueue1(){
        return new Queue("headerExchangeQueue1");
        }

@Bean
public Queue headerExchangeQueue2(){
        return new Queue("headerExchangeQueue2");
        }
@Bean
public HeadersExchange headersExchange(){
        return new HeadersExchange("amp.header");
        }
@Bean
public Binding headExchangeToQueue1(Queue headerExchangeQueue1,HeadersExchange headersExchange){
        HashMap<String, Object> map = new HashMap<>();
        map.put("type","OK");
        map.put("status","200");
        return BindingBuilder.bind(headerExchangeQueue1).to(headersExchange).whereAll(map).match();
        }
@Bean
public Binding headExchangeToQueue2(Queue headerExchangeQueue2,HeadersExchange headersExchange){
        HashMap<String, Object> map = new HashMap<>();
        map.put("type","error");
        map.put("status","500");
        return BindingBuilder.bind(headerExchangeQueue2).to(headersExchange).whereAll(map).match();
        }

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

/** 生产者发送消息*/
@Test
    void sendByHeadExchange() {
            Map<String, Object> headers = new HashMap<>();
        headers.put("type","OK");
        headers.put("status","200");
        String message = "这是发送到的消息";
        MessageProperties messageProperties = new MessageProperties();
        headers.forEach(messageProperties::setHeader);
        Message msg = new Message(message.getBytes(), messageProperties);
        amqpTemplate.convertAndSend("amp.header",null, msg);
        System.out.println("发送成功！");

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

    @RabbitListener(queues = "headerExchangeQueue1")
    public void test08(Message msg){
        System.out.println("接收到消息8:" + msg.toString());

    }
    @RabbitListener(queues = "headerExchangeQueue2")
    public void test09(Message msg){
        System.out.println("接收到消息9:" + msg.toString());
    }
1
2
3
4
5
6
7
8
9

结果：只有匹配上header才能收到消息

接收到消息8:(Body:'[B@a7b38a8(byte[24])' MessageProperties [headers={type=OK, status=200}, contentType=application/octet-stream, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=amp.header, receivedRoutingKey=, deliveryTag=2, consumerTag=amq.ctag-1WTdKW4n_rAEdJUosQD7bg, consumerQueue=headerExchangeQueue1])

1
2