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.扇形交换机

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

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.扇形交换机

热门文章

最新文章

相关课程

相关电子书

相关实验场景