RabbitMQ常见的交换机类型

news2024/9/20 18:33:04

`RabbitMQ安装`

pom.xml里导入相关的依赖：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

application.properties配置文件

spring.rabbitmq.host=192.168.152.155
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
#开启发送端确认 生产者Publisher 到服务器Broker
spring.rabbitmq.publisher-confirms=true
#开启发送端消息抵达队列的确认
spring.rabbitmq.publisher-returns=true
#只要抵达队列，以异步发送优先回调我们这个returnConfirm
spring.rabbitmq.template.mandatory=true

`Direct exchange（直通交换机）`

消息中的路由键（routing key）如果和 Binding中的binding key 一致，交换器就将消息发到对应的队列中。路由键与队列名完全匹配，如果一个队列绑定到交换机要求路由键为“dog”，则只转发routing key 标记为“dog”的消息，不会转发 “dog.puppy”，也不会转发“dog.guard” 等等。它是 完全匹配、单播的模式 。

接着我们先使用下 direct exchange(直连型交换机)，创建 DirectRabbitConfig.java：

package com.beijing.gulimall.product.rabbitmq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Slf4j
@Configuration
public class DirectRabbitConfig {

    //创建队列
    @Bean
    public Queue TestDirectQueue() {
        //public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete)
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
        // exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
        log.info("Queue[{}]创建成功", "TestDirectQueue");
        return new Queue("TestDirectQueue", true, false, false);
    }

    //创建交换机
    @Bean
    DirectExchange TestDirectExchange() {
        log.info("Exchange[{}]创建成功", "TestDirectExchange");
        return new DirectExchange("TestDirectExchange", true, false);
    }

    //创建绑定关系
    @Bean
    Binding TestBindingDirect() {
        //	public Binding(String destination, DestinationType destinationType, String exchange, String routingKey,
        //			Map<String, Object> arguments) {
        log.info("Binding[{}]创建成功", "TestBindingDirect");
        return new Binding("TestDirectQueue", Binding.DestinationType.QUEUE, "TestDirectExchange", "direct.test", null);
    }

}

然后写个简单的接口进行消息推送 SendMessageController.java：

package com.beijing.gulimall.product.rabbitmq;

import com.alibaba.fastjson.JSONObject;
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.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@RestController
public class SendMessageController {

    @Autowired
    RabbitTemplate rabbitTemplate;


    @RequestMapping("/hello")
    public void testRabbitMQ() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);

        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "direct.test", JSONObject.toJSONString(map));
    }
}

Fanout Exchange （扇出交换机）

每个发到 fanout 类型交换器的消息都会分到所有绑定的队列上去。fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。fanout 类型转发消息是最快的。

创建FanoutRabbitConfig.class

package com.beijing.gulimall.product.rabbitmq;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutRabbitConfig {
    
    @Bean
    public FanoutExchange testFanoutExchange(){
        return new FanoutExchange("TestFanoutExchange",true,false);
    }

    @Bean
    public Queue testFanoutQueueOne(){
        return new Queue("TestFanoutQueueOne",true,false,false);
    }

    @Bean
    public Queue testFanoutQueueTwo(){
        return new Queue("TestFanoutQueueTwo",true,false,false);
    }
    @Bean
    public Binding createFanoutBindingOne(){
        return new Binding("TestFanoutQueueOne", Binding.DestinationType.QUEUE,"TestFanoutExchange","fanout.one231.test",null);
    }

    @Bean
    public Binding createFanoutBindingTwo(){
        return new Binding("TestFanoutQueueTwo", Binding.DestinationType.QUEUE,"TestFanoutExchange","fanout123.one.#",null);
    }
}

发送消息

    @RequestMapping("/hello3")
    public void testRabbitMQ3() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        //topic.one.test2
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestFanoutExchange","",JSONObject.toJSONString(map));
    }

结果：

主题交换机（Topic Exchange）

topic 交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开 。它同样也

会识别两个通配符：符号“ #” 和符号 “* ”。 # 匹配 0 个或多个单词， * 匹配一个单词。

创建 TopicRabbitConfig.class

package com.beijing.gulimall.product.rabbitmq;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TopicRabbitConfig {


    @Bean
    public TopicExchange testTopicExchange(){
        return new TopicExchange("TestTopicExchange");
    }

    @Bean
    public Queue testTopicQueueOne(){
        return new Queue("TestTopicQueueOne",true,false,false);
    }

    @Bean
    public Queue testTopicQueueTwo(){
        return new Queue("TestTopicQueueTwo",true,false,false);
    }

    @Bean
    public Binding createBindingOne(){
        return new Binding("TestTopicQueueOne", Binding.DestinationType.QUEUE,"TestTopicExchange","topic.one.test",null);
    }


    @Bean
    public Binding createBindingTwo(){
        return new Binding("TestTopicQueueTwo", Binding.DestinationType.QUEUE,"TestTopicExchange","topic.one.#",null);
    }

}

发送消息

    @RequestMapping("/hello2")
    public void testRabbitMQ2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        //topic.one.test2
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestTopicExchange", "topic.one.test", JSONObject.toJSONString(map));
    }

结果：