DirectExchange直连交换机

客官°小女子只卖身不卖艺 2023-10-15 10:19 69阅读 0赞

**目录**

一、简介

二、使用步骤

三、demo

父pom文件

pom文件

配置文件

config

过期时间

消费者

生产者

测试

四、多个消费者获取生产者发送的所有消息

--------------------

## 一、简介 ##

**直连**型交换机，根据消息携带的路由键将消息投递给对应队列。

大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 routing key 。

然后当一个消息携带着路由值为abc，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值abc去寻找绑定值也是X的队列。

## 二、使用步骤 ##

*
     * 步骤：
     * 1、交换机注入bean
     * 2、队列注入bean
     * 3、队列与交换机绑定
     * 注意:虽然交换机、队列、绑定关系都已注入bean中,但是rabbitMQ中还是没有创建的交换机和队列。
     *     那是因为消费者还未创建,当消费者创建并与队列绑定后就能在rabbitMQ服务中看到创建的交换机和队列;
     *     那么,若想在rabbitMQ服务中查看到创建的交换机和队列,还需要第4步,创建消费者
     * 4、创建消费者consumer

## 三、demo ##

![9da72b2349904c4e85a41c79302a7648.png][]

### 父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.1</version>
    
            <relativePath/> 
        </parent>
        <groupId>com.chensir</groupId>
        <artifactId>spring-boot-rabbitmq</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <name>spring-boot-rabbitmq</name>
    
        <properties>
            <java.version>8</java.version>
            <hutool.version>5.8.3</hutool.version>
            <lombok.version>1.18.24</lombok.version>
        </properties>
    
        <description>spring-boot-rabbitmq</description>
    
        <packaging>pom</packaging>
    
        <modules>
            <module>direct-exchange</module>
            <module>fanout-exchange</module>
            <module>topic-exchange</module>
            <module>game-exchange</module>
            <module>dead-letter-queue</module>
            <module>delay-queue</module>
            <module>delay-queue2</module>
        </modules>
    
        <dependencyManagement>
            <dependencies>
                <dependency>
                    <groupId>cn.hutool</groupId>
                    <artifactId>hutool-all</artifactId>
                    <version>${hutool.version}</version>
                </dependency>
    
                <dependency>
                    <groupId>org.projectlombok</groupId>
                    <artifactId>lombok</artifactId>
                    <version>${lombok.version}</version>
                </dependency>
            </dependencies>
        </dependencyManagement>
    
    
    </project>

### 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>com.chensir</groupId>
            <artifactId>spring-boot-rabbitmq</artifactId>
            <version>0.0.1-SNAPSHOT</version>
            <relativePath>../pom.xml</relativePath>
        </parent>
    
        <artifactId>direct-exchange</artifactId>
    
        <dependencies>
            <dependency>
                <groupId>cn.hutool</groupId>
                <artifactId>hutool-all</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </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>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.amqp</groupId>
                <artifactId>spring-rabbit-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-maven-plugin</artifactId>
                </plugin>
            </plugins>
        </build>
    
    </project>

### 配置文件 ###

server.port=8081
    
    logging.level.com.chensir=debug
    #host
    spring.rabbitmq.host=121.40.100.66
    #spring.rabbitmq.host=121.199.39.162
    
    #默认5672
    spring.rabbitmq.port=5672
    #用户名
    spring.rabbitmq.username=guest
    #密码
    spring.rabbitmq.password=guest
    #连接到代理时用的虚拟主机
    spring.rabbitmq.virtual-host=/
    #每个消费者每次可最大处理的nack消息数量 默认是250个 可在服务界面看到;意思是每批投递的数量
    spring.rabbitmq.listener.simple.prefetch=1
    #表示消息确认方式，其有三种配置方式，分别是none、manual(手动)和auto(自动)；默认auto
    spring.rabbitmq.listener.simple.acknowledge-mode=auto
    #是否开启自动重试 默认为false 不开启
    spring.rabbitmq.listener.simple.retry.enabled=true
    #最大重试次数
    spring.rabbitmq.listener.simple.retry.max-attempts=5
    #最大重试时间间隔
    spring.rabbitmq.listener.simple.retry.max-interval=20000ms
    #重试时间间隔
    spring.rabbitmq.listener.simple.retry.initial-interval=2000ms
    # 最大重试间隔*乘数
    #应用于上一重试间隔的乘数 第一次(重试时间间隔)2s 4s 8s 16s 32s 此处32s>20s 以后都以20s为间隔 总的次数为最大重试次数
    spring.rabbitmq.listener.simple.retry.multiplier=2
    #决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
    spring.rabbitmq.listener.simple.default-requeue-rejected=false

### config ###

在config中去注入交换机、队列、以及配置队列与交换机的绑定；

一个交换机可以有多个队列与之绑定；

一个队列可以多个消费者绑定消费;

下面为config注入bean的方式;还有种使用注解的方式(使用注解则不需要config配置,但是需要在每个消费者上使用注解配置,这样的缺点有不方便管理)

> config注入

package com.chensir.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    
    /**
     * 步骤：
     * 1、交换机注入bean
     * 2、队列注入bean
     * 3、队列与交换机绑定
     * 注意:虽然交换机、队列、绑定关系都已注入bean中,但是rabbitMQ中还是没有创建的交换机和队列。
     * 那是因为消费者还未创建,当消费者创建并与队列绑定后就能在rabbitMQ服务中看到创建的交换机和队列;
     * 那么,若想在rabbitMQ服务中查看到创建的交换机和队列,还需要第4步,创建消费者
     * 4、创建消费者consumer
     */
    @Configuration
    public class RabbitConfig {
    
        //交换机
        public static final String EXCHANGE_KEY = "DirectExchange-01";
    
        //队列
        public static final String QUEUE_KEY_01 = "DirectQueue-01";
        public static final String QUEUE_KEY_02 = "DirectQueue-02";
        public static final String QUEUE_KEY_03 = "DirectQueue-03";
    
        //routing
        public static final String ROUTING_KEY_01 = "Direct-RoutingKey-01";
        public static final String ROUTING_KEY_02 = "Direct-RoutingKey-02";
        public static final String ROUTING_KEY_03 = "Direct-RoutingKey-03";
    
    
    
        //解决对象类型乱码  默认的对象类型时base64
    //    @Bean
    //    public Jackson2JsonMessageConverter messageConverter() {
    //        return new Jackson2JsonMessageConverter();
    //    }
    
        /**
         * 直连交换机 一个交换机可绑定多个队列
         */
        @Bean
        public DirectExchange directExchange() {
            // name:交换机名称 durable:是否持久化 autoDelete:是否自动删除(没人用时是否自动删除)
            DirectExchange directExchange = new DirectExchange(EXCHANGE_KEY, true, false);
            return directExchange;
        }
    
        /**
         * 队列 一个队列可多个消费者使用
         *
         * @return
         */
        @Bean
        public Queue directQueue() {
            //autoDelete() 队列自动删除,当消费者不存在,队列自动删除;注意,这个并不常用!容易丢失消息(队列本来就是存放消息的,自动删除就没意义了)
    //        Queue queue = QueueBuilder.durable("DirectQueue-01").autoDelete().build();
            //durable 持久化;持久化到server硬盘中; 使用这个后每次本地服务(项目)重启 rabbitMQ服务上会自动创建此队列
            Queue queue = QueueBuilder.durable(QUEUE_KEY_01).build();
            return queue;
        }
    
        /**
         * 队列2
         * @return
         */
        @Bean
        public Queue directQueue2() {
            Queue queue = QueueBuilder.durable(QUEUE_KEY_02).build();
            return queue;
        }
    
        /**
         * 队列 测试接收user对象
         * @return
         */
        @Bean
        public Queue directQueueUser() {
            Queue queue = QueueBuilder.durable(QUEUE_KEY_03).build();
            return queue;
        }
    
        /**
         * 绑定队列与交换机
         *
         * @return
         */
        @Bean
        public Binding binding() {
            //绑定directQueue这个队列给 directExchange 这个交换机
            Binding binding = BindingBuilder.bind(directQueue()).to(directExchange()).with(ROUTING_KEY_01);
            return binding;
        }
    
        /**
         * 也与交换机DirectExchange-01 绑定;一个交换机可以绑定多个队列
         *
         * @return
         */
        @Bean
        public Binding binding2() {
            Binding binding = BindingBuilder.bind(directQueue2()).to(directExchange()).with(ROUTING_KEY_02);
            return binding;
        }
    
        /**
         * 也与交换机DirectExchange-01 绑定;一个交换机可以绑定多个队列
         * 测试user对象
         *
         * @return
         */
        @Bean
        public Binding bindingQueueUser() {
            Binding binding = BindingBuilder.bind(directQueueUser()).to(directExchange()).with(ROUTING_KEY_03);
            return binding;
        }
    }

#### 过期时间 ####

使用ttl对‘队列’进行设置过期时间；若10s内消息未被消费则自动删除；(队列中消息均会被删除)不建议使用![9b692647ce6747d09bfaafc62f0dcc70.png][]

在生产者中对消息设置过期时间

![4cea11a809a344d39135e5a0e1f4f2c1.png][]

**注意：一般过期时间会结合者死信队列使用,当消息过期时将消息打到死信队列中**

> 使用注解

使用注解需要在消费者上使用注解并把交换机、队列、key值配置上去即可;

使用了注解方式就不需要在config中配置了。

@RabbitHandler
        @RabbitListener(bindings = @QueueBinding(
                value = @Queue(value = "DirectQueue-01",durable = "true",autoDelete = "false"),
                exchange = @Exchange(value = "DirectExchange-01",type = ExchangeTypes.DIRECT),
                key = "Direct-RoutingKey-01"
        ))
    //    @RabbitListener(queues = {"DirectQueue-01","DirectQueue-02"})
        public void process2(Message message) {
            System.out.println(message);
            String test = new String(message.getBody());
            System.out.println("消费2消费了"+test);
        }

### 消费者 ###

package com.chensir.consumer;
    
    import cn.hutool.core.codec.Base32;
    import cn.hutool.json.JSON;
    import cn.hutool.json.JSONObject;
    import cn.hutool.json.JSONUtil;
    import com.chensir.config.RabbitConfig;
    import com.chensir.model.User;
    
    
    import lombok.extern.slf4j.Slf4j;
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.ExchangeTypes;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    import java.io.UnsupportedEncodingException;
    import java.nio.charset.StandardCharsets;
    import java.util.Base64;
    
    /**
     * 消费者
     */
    
    @Component
    @Slf4j
    public class DirectConsumer {
    
    //   注解示例
    //    /**
    //     * 使用注解方式 使消费者与交换机、队列绑定;使用了注解就无需在config中进行配置
    //     * @param message
    //     */
    //    @RabbitHandler
    //    @RabbitListener(bindings = @QueueBinding(
    //            value = @Queue(value = "DirectQueue-01",durable = "true",autoDelete = "false"),
    //            exchange = @Exchange(value = "DirectExchange-01",type = ExchangeTypes.DIRECT),
    //            key = "Direct-RoutingKey-01"
    //    ))
        @RabbitListener(queues = {"DirectQueue-01","DirectQueue-02"})
    //    public void process2(Message message) {
    //        System.out.println(message);
    //        String test = new String(message.getBody());
    //        System.out.println("消费2消费了"+test);
        }
    
    
        /**
         * 消费者process 消费DirectQueue-01队列;可多个消费之消费同一个队列
         *
         * @param message
         */
        @RabbitHandler
        @RabbitListener(queues = RabbitConfig.QUEUE_KEY_01)
    //    @RabbitListener(queues = {"DirectQueue-01","DirectQueue-02"})
        public void process(String message) {
            System.out.println("消费1消费了" + message);
        }
    
        /**
         * 消费者process2 也消费DirectQueue-01队列;与消费者process验证多个消费者消费同一个队列
         * <p>
         * 通过验证发现,多个消费者消费同一个队列 会均摊队列中的消息;比如队列中有编号为1-10 10个队列
         * 那么,消费者1 可能消费编号为:1、3、5、7、9的消息, 消费者2 可能消费编号为:2、4、6、8、10
         *
         * @param message
         */
        @RabbitHandler
        @RabbitListener(queues = RabbitConfig.QUEUE_KEY_01)
        public void process2(Message message) {
            System.out.println(message);
            String test = new String(message.getBody());
            System.out.println("消费2消费了" + test);
        }
    
        /**
         * 消费者3去消费队列DirectQueue-02队列
         *
         * @param message
         */
        @RabbitHandler
        @RabbitListener(queues = RabbitConfig.QUEUE_KEY_02)
        public void process3(Message message) {
            System.out.println(message);
            String test = new String(message.getBody());
            System.out.println("消费3消费了" + test);
        }
    
        /**
         * 消费user类型 方式一
         *
         * @param message
         */
        @RabbitHandler
        @RabbitListener(queues = RabbitConfig.QUEUE_KEY_03)
        public void consumerUser(Message message) {
            String test = new String(message.getBody());
            System.out.println(test);
    
        }
    
        /**
         * 消费user类型 方式二
         *
         * @param message
         */
        @RabbitHandler
        @RabbitListener(queues = RabbitConfig.QUEUE_KEY_03)
        public void consumerUser2(String message) {
    
            User user = JSONUtil.toBean(message, User.class);
            String name = user.getName();
            System.out.println(user);
            System.out.println(name);
        }
    
    }

### 生产者 ###

package com.chensir.provider;
    
    import cn.hutool.json.JSONUtil;
    import com.chensir.config.RabbitConfig;
    import com.chensir.model.User;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * 生产者
     */
    @Component
    @Slf4j
    public class DirectProvider {
    
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        /**
         * 生产者-简单的发送消息 示例
         */
    //    public void send0() {
    //
    //        // 往DirectExchange-01交换机中发送消息,交换机的密钥为Direct-RoutingKey-01
            rabbitTemplate.send("DirectExchange-01", "Direct-RoutingKey-01", "hello,rabbitMQ");
    //        rabbitTemplate.convertAndSend("DirectExchange-01", "Direct-RoutingKey-01", "hello,rabbitMQ");
    //    }
    
        /**
         * 生产者1,往routing1中(也就是队列1)投递消息
         */
        public void send1() {
            for (int i = 1; i < 11; i++) {
    
                rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_KEY, RabbitConfig.ROUTING_KEY_01, "生产者1-" + i);
            }
    
        }
    
        /**
         * 生产者2,往routing2中(也就是队列2)投递消息
         */
        public void send2() {
            for (int i = 1; i < 11; i++) {
    
                rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_KEY, RabbitConfig.ROUTING_KEY_02, "生产者2-" + i);
            }
        }
    
        /**
         * 生产者 发送一个对象
         */
        public void sendUser() {
            User user = User.builder()
                    .id(1)
                    .name("陈sir")
                    .sex("男")
                    .build();
            String s = JSONUtil.toJsonStr(user);
    
            Map<String,String> stringMap =new HashMap<>();
            stringMap.put("key","ss");
    
            rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_KEY, RabbitConfig.ROUTING_KEY_03, s);
        }
    
    }

### 测试 ###

![5542b294af0e4333bdfa3eddd476f67d.png][]

![8ea350b0da0a4e799b2d7c1649b079af.png][]

## 四、多个消费者获取生产者发送的所有消息 ##

若使用多个消费者去和一个队列绑定,当生产者发送2个消息,则消费者会均摊; 比如一个生产者发送2个消息,而有两个消费者都绑定了同一个队列,此时消费者1和消费者2 会一人消费队列中的一个消息;不会说队列中的2个消息，消费者1和消费者2都能拿到并消费;

若想让多个消费者都能消费到消息，可以起两个队列,队列名不同,但此时交换机还是一个，队列起2个routingkey可以使用一个；

或者使用广播交换机

[9da72b2349904c4e85a41c79302a7648.png]: https://img-blog.csdnimg.cn/9da72b2349904c4e85a41c79302a7648.png
[9b692647ce6747d09bfaafc62f0dcc70.png]: https://img-blog.csdnimg.cn/9b692647ce6747d09bfaafc62f0dcc70.png
[4cea11a809a344d39135e5a0e1f4f2c1.png]: https://img-blog.csdnimg.cn/4cea11a809a344d39135e5a0e1f4f2c1.png
[5542b294af0e4333bdfa3eddd476f67d.png]: https://img-blog.csdnimg.cn/5542b294af0e4333bdfa3eddd476f67d.png
[8ea350b0da0a4e799b2d7c1649b079af.png]: https://img-blog.csdnimg.cn/8ea350b0da0a4e799b2d7c1649b079af.png