02.交换机RabbitMQ交换机

1.Exchange(交换机)的作用

在RabbitMQ中，生产者发送消息不会直接将消息投递到队列中，而是先将消息投递到交换机中，在由交换机转发到具体的队列，

队列再将消息以推送或者拉取方式给消费者进行消费

创建消息              路由键         pull/push

生产者------------>交换机------------>队列------------>消费者

2. Exchange(交换机)的类型

2.1.直连交换机：Direct Exchange

直连交换机是一种带路由功能的交换机，一个队列会和一个交换机绑定，除此之外再绑定一个routing_key，当消息被发送的时候，需要指定一个binding_key，这个消息被送达交换机的时候，就会被这个交换机送到指定的队列里面去。同样的一个binding_key也是支持应用到多个队列中的。

这样当一个交换机绑定多个队列，就会被送到对应的队列去处理。

注1：什么是路由键

每个消息都有一个称为路由键(routing key)的属性，它其实就是一个简单的字符串

注2：直连交换机适用场景

有优先级的任务，根据任务的优先级把消息发送到对应的队列，这样可以指派更多的资源去处理高优先级的队列。

2.2.主题交换机：Topic Exchange

直连交换机的缺点！

直连交换机的routing_key方案非常简单，如果我们希望一条消息发送给多个队列，那么这个交换机需要绑定上非常多的routing_key，

假设每个交换机上都绑定一堆的routing_key连接到各个队列上。那么消息的管理就会异常地困难。

所以RabbitMQ提供了一种主题交换机，发送到主题交换机上的消息需要携带指定规则的routing_key，

主题交换机会根据这个规则将数据发送到对应的(多个)队列上。

主题交换机的routing_key需要有一定的规则，交换机和队列的binding_key需要采用*.#.*…的格式，每个部分用.分开，其中

*表示一个单词

#表示任意数量（零个或多个）单词。

示例：
  队列Q1绑定键为 *.TT.*
  队列Q2绑定键为TT.#
  如果一条消息携带的路由键为 A.TT.B，那么队列Q1将会收到 
  如果一条消息携带的路由键为TT.AA.BB，那么队列Q2将会收到

2.3.扇形交换机：Fanout Exchange

扇形交换机是最基本的交换机类型，它所能做的事情非常简单———广播消息。

扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列。因为广播不需要“思考”，

所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。

这个交换机没有路由键概念，就算你绑了路由键也是无视的。

2.4.首部交换机：Headers exchange

2.5.默认交换机

实际上是一个由RabbitMQ预先声明好的名字为空字符串的直连交换机（direct exchange）。它有一个特殊的属性使得它对于

简单应用特别有用处：那就是每个新建队列（queue）都会自动绑定到默认交换机上，绑定的路由键（routing key）名称与队列名称相同。

如：当你声明了一个名为”hello”的队列，RabbitMQ会自动将其绑定到默认交换机上，绑定（binding）的路由键名称也是为”hello”。

因此，当携带着名为”hello”的路由键的消息被发送到默认交换机的时候，此消息会被默认交换机路由至名为”hello”的队列中

类似amq.*的名称的交换机：这些是RabbitMQ默认创建的交换机。这些队列名称被预留做RabbitMQ内部使用，不能被应用使用，否则抛出403 (ACCESS_REFUSED)错误

2.6.Dead Letter Exchange（死信交换机）

在默认情况，如果消息在投递到交换机时，交换机发现此消息没有匹配的队列，则这个消息将被悄悄丢弃。

为了解决这个问题，RabbitMQ中有一种交换机叫死信交换机。当消费者不能处理接收到的消息时，将这个消息重新发布到另外一个队列中，等待重试或者人工干预。这个过程中的exchange和queue就是所谓的”Dead Letter Exchange 和 Queue

3. 交换机的属性

除交换机类型外，在声明交换机时还可以附带许多其他的属性，其中最重要的几个分别是：

Name：交换机名称

Durability：是否持久化。如果持久性，则RabbitMQ重启后，交换机还存在

Auto-delete：当所有与之绑定的消息队列都完成了对此交换机的使用后，删掉它

Arguments：扩展参数

4. 综合案例：交换机的使用

4.1.新建两个模块

rabbitmq-provider和rabbitmq-consumer

rabbitmq-provider的pom

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.2</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.zjzaki</groupId>
    <artifactId>rabbitmq-provider</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rabbitmq-provider</name>
    <description>rabbitmq-provider</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </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-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

application.yml

server:
  port: 8081
spring:
  rabbitmq:
    host: localhost
    password: 123456
    port: 5672
    username: zjzaki
    virtual-host: my_vhost

rabbitmq-consumer的pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.2</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.zjzaki</groupId>
    <artifactId>rabbitmq-consumer</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rabbitmq-consumer</name>
    <description>rabbitmq-consumer</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </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-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

application.yml

server:
  port: 8082
spring:
  rabbitmq:
    host: localhost
    password: 123456
    port: 5672
    username: zjzaki
    virtual-host: my_vhost

4.2.在rabbitmq-provider中新建config包

package com.zjzaki.rabbitmqprovider.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @Author zjzaki
 * @Package com.zjzaki.rabbitmqprovider.config
 * @Date 2023-09-05 21:50:58
 *
 * 1.Spring 要管理交换机以及交换机对应的队列,还有绑定关系
 * 2.生产者微服务要向交换机投递消息
 * 3.rabbitmq的控制台是否能够接收到消息
 * 4.消费者微服务能够消费消息
 */
@Configuration
public class RabbitmqDirectConfig {
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("zjzaki-direct-exchange");
    }
    @Bean
    public Queue directQueue(){
        return new Queue("zjzaki-direct-queue");
    }
    @Bean
    public Binding directBinding(){
        return BindingBuilder
                .bind(directQueue())
                .to(directExchange())
                .with("orange");
    }
}

4.3.rabbitmq-provider新建controller包

SendMessageController.java

package com.zjzaki.rabbitmqprovider.controller;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
/**
 * @Author zjzaki
 * @Package com.zjzaki.rabbitmqprovider.controller
 * @Date 2023-09-05 22:23:29
 */
@RestController
public class SendMessageController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @RequestMapping("/direct")
    public String directMsg(String rk) {
        Map map = new HashMap<>();
        map.put("msg", "此消息为[直连交换机]路由过来的");
        map.put("time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        rabbitTemplate.convertAndSend("zjzaki-direct-exchange", rk, map);
        return "ok";
    }
}

4.4.访问 http://localhost:8081/direct

访问 http://localhost:8081/direct?rk=orange

4.5.rabbitmq-consumer新建config包

新建 DirectReceiver.java

package com.zjzaki.rabbitmqconsumer.config;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component
@RabbitListener(queues = {"zjzaki-direct-queue"})
public class DirectReceiver {
    // @RabbitListener(queues = {"direct-queue"})
    @RabbitHandler
    public void handler(Map msg) {
        System.out.println(msg);
    }
}

重启项目

4.6.主题交换机

1.新建RabbitTopicConfig.java

package com.zjzaki.rabbitmqprovider.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @Author zjzaki
 * @Package com.zjzaki.rabbitmqprovider.config
 * @Date 2023-09-05 22:46:43
 */
@Configuration
public class RabbitTopicConfig {
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("zjzaki-topic-exchange");
    }
    @Bean
    public Queue directQueue1(){
        return new Queue("zjzaki-direct-queue1");
    }
    @Bean
    public Queue directQueue2(){
        return new Queue("zjzaki-direct-queue2");
    }
    @Bean
    public Queue directQueue3(){
        return new Queue("zjzaki-direct-queue3");
    }
    @Bean
    public Binding directBinding1(){
        return BindingBuilder
                .bind(directQueue1())
                .to(topicExchange())
                .with("zjzaki.xx");
    }
    @Bean
    public Binding directBinding2(){
        return BindingBuilder
                .bind(directQueue2())
                .to(topicExchange())
                .with("zjzaki.yy");
    }
    @Bean
    public Binding directBinding3(){
        return BindingBuilder
                .bind(directQueue3())
                .to(topicExchange())
                .with("zjzaki.*");
    }
}

2.controller中添加方法

package com.zjzaki.rabbitmqprovider.controller;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
/**
 * @Author zjzaki
 * @Package com.zjzaki.rabbitmqprovider.controller
 * @Date 2023-09-05 22:23:29
 */
@RestController
public class SendMessageController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @RequestMapping("/direct")
    public String directMsg(String rk) {
        Map map = new HashMap<>();
        map.put("msg", "此消息为[直连交换机]路由过来的");
        map.put("time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        rabbitTemplate.convertAndSend("zjzaki-direct-exchange", rk, map);
        return "ok";
    }
    @RequestMapping("/topic")
    public String topic(String rk) {
        Map map = new HashMap<>();
        map.put("msg", "此消息为[主题交换机]路由过来的");
        map.put("time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        rabbitTemplate.convertAndSend("zjzaki-topic-exchange1", rk, map);
        return "ok";
    }
}

3.浏览器访问 http://localhost:8081/topic?rk=zjzaki.xx

4.浏览器访问 http://localhost:8081/topic?rk=zjzaki.yy

5.浏览器访问 http://localhost:8081/topic?rk=zjzaki.zz

4.7.扇形交换机

添加RabbitmqFanoutConfig.java

package com.zjzaki.rabbitmqprovider.config;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @Author zjzaki
 * @Package com.zjzaki.rabbitmqprovider.config
 * @Date 2023-09-05 23:03:56
 */
@Configuration
public class RabbitmqFanoutConfig {
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("zjzaki-fanout-exchange");
    }
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("zjzaki-fanout-queue1");
    }
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("zjzaki-fanout-queue2");
    }
    @Bean
    public Queue fanoutQueue3(){
        return new Queue("zjzaki-fanout-queue3");
    }
    @Bean
    public Binding fanoutBinding1(){
        return BindingBuilder
                .bind(fanoutQueue1())
                .to(fanoutExchange());
    }
    @Bean
    public Binding fanoutBinding2(){
        return BindingBuilder
                .bind(fanoutQueue2())
                .to(fanoutExchange());
    }
    @Bean
    public Binding fanoutBinding3(){
        return BindingBuilder
                .bind(fanoutQueue3())
                .to(fanoutExchange());
    }
}

SendMessageController添加内容

package com.zjzaki.rabbitmqprovider.controller;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
/**
 * @Author zjzaki
 * @Package com.zjzaki.rabbitmqprovider.controller
 * @Date 2023-09-05 22:23:29
 */
@RestController
public class SendMessageController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @RequestMapping("/direct")
    public String directMsg(String rk) {
        Map map = new HashMap<>();
        map.put("msg", "此消息为[直连交换机]路由过来的");
        map.put("time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        rabbitTemplate.convertAndSend("zjzaki-direct-exchange", rk, map);
        return "ok";
    }
    @RequestMapping("/topic")
    public String topic(String rk) {
        Map map = new HashMap<>();
        map.put("msg", "此消息为[主题交换机]路由过来的");
        map.put("time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        rabbitTemplate.convertAndSend("zjzaki-topic-exchange", rk, map);
        return "ok";
    }
    @RequestMapping("/fanout")
    public String fanoutMsg(String rk) {
        Map map = new HashMap<>();
        map.put("msg", "此消息为[扇形交换机]路由过来的");
        map.put("time", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        rabbitTemplate.convertAndSend("zjzaki-fanout-exchange", rk, map);
        return "ok";
    }
}

访问: http://localhost:8081/fanout

02.交换机RabbitMQ交换机

02.交换机RabbitMQ交换机

1.Exchange(交换机)的作用

2. Exchange(交换机)的类型

2.1.直连交换机：Direct Exchange

2.2.主题交换机：Topic Exchange

2.3.扇形交换机：Fanout Exchange

2.4.首部交换机：Headers exchange

2.5.默认交换机

2.6.Dead Letter Exchange（死信交换机）

3. 交换机的属性

4. 综合案例：交换机的使用

4.1.新建两个模块

4.2.在rabbitmq-provider中新建config包

4.3.rabbitmq-provider新建controller包

4.4.访问 http://localhost:8081/direct

4.5.rabbitmq-consumer新建config包

4.6.主题交换机

4.7.扇形交换机

热门文章

最新文章

相关课程

相关电子书

相关实验场景