02.交换机RabbitMQ交换机

简介: 02.交换机RabbitMQ交换机

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

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
2月前
|
消息中间件 存储 缓存
RabbitMQ中的交换机
RabbitMQ中的交换机
34 2
|
2月前
|
消息中间件 Java Spring
SpringBoot实现RabbitMQ的广播交换机(SpringAMQP 实现Fanout广播交换机)
SpringBoot实现RabbitMQ的广播交换机(SpringAMQP 实现Fanout广播交换机)
31 2
|
2月前
|
消息中间件 Java
SpringBoot实现RabbitMQ的通配符交换机(SpringAMQP 实现Topic交换机)
SpringBoot实现RabbitMQ的通配符交换机(SpringAMQP 实现Topic交换机)
22 1
|
2月前
|
消息中间件 Java Spring
SpringBoot实现RabbitMQ的定向交换机(SpringAMQP 实现Direct定向交换机)
SpringBoot实现RabbitMQ的定向交换机(SpringAMQP 实现Direct定向交换机)
30 1
|
2月前
|
消息中间件 微服务
RabbitMQ入门指南(十):延迟消息-死信交换机
RabbitMQ是一个高效、可靠的开源消息队列系统,广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了死信交换机、死信交换机实现延迟消息等内容。
93 0
|
2月前
|
消息中间件 Java API
RabbitMQ入门指南(五):Java声明队列、交换机以及绑定
RabbitMQ是一个高效、可靠的开源消息队列系统,广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了Java声明队列、交换机以及绑定队列和交换机等内容。
112 0
|
2月前
|
消息中间件 微服务
RabbitMQ入门指南(四):交换机与案例解析
RabbitMQ是一个高效、可靠的开源消息队列系统,广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了交换机在RabbitMQ中的作用与类型、交换机案例(Fanout交换机、Direct交换机、Topic交换机)等内容。
180 0
|
2月前
|
消息中间件 Java Maven
springboot 使用注解的方式创建rabbitmq的交换机、路由key、以及监听队列的名称
springboot 使用注解的方式创建rabbitmq的交换机、路由key、以及监听队列的名称
|
2月前
|
消息中间件 算法 微服务
升华 RabbitMQ:解锁一致性哈希交换机的奥秘【RabbitMQ 十】
升华 RabbitMQ:解锁一致性哈希交换机的奥秘【RabbitMQ 十】
60 0
|
13天前
|
消息中间件 测试技术 RocketMQ
消息队列 MQ产品使用合集之在异步发送消息函数sendMessage()中出现了错误,错误代码为-3,该如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。