[大数据]RabbitMQ系列--Fanout Exchange

关于RabbitMQ的介绍以及工作流程，还有Direct Exchange、Topic Exchange的相关使用大家可以看下前两篇文章，本篇主要介绍Fanout Exchange，下面直接进入正题。

Fanout Exchange

这个扇形交换机的话，相对前两种而言是更加简单的，因为它没有路由键的这个概念，如果你去绑定了路由键的话，也是无效的哈，交换机接收到生产者发送的消息后会直接投递给绑定的队列中。

编写RabbitMQ示例

• 生产者项目创建扇形交换机

package com.chentawen.rabbitmqprovider.config;

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

/**
 * @Author: CTW
 * @Date: create in 2021/8/4 21:38
 */
@Configuration
public class FanoutExchangeConfig {

    /**
     * 声明扇形交换机
     *
     * @return
     */
    @Bean
    FanoutExchange MyFanoutExchange() {
        return new FanoutExchange("MyFanoutExchange", true, false);
    }

    /**
     * 声明队列A
     *
     * @return
     */
    @Bean
    Queue MyFanoutQueueA() {
        return new Queue("MyFanoutQueueA", true);
    }

    /**
     * 声明队列B
     *
     * @return
     */
    @Bean
    Queue MyFanoutQueueB() {
        return new Queue("MyFanoutQueueB", true);
    }

    /**
     * 将交换机和队列进行绑定
     *
     * @return
     */
    @Bean
    Binding bindingFanoutA() {
        return BindingBuilder.bind(MyFanoutQueueA()).to(MyFanoutExchange());
    }

    /**
     * 将交换机和队列进行绑定
     *
     * @return
     */
    @Bean
    Binding bindingFanoutB() {
        return BindingBuilder.bind(MyFanoutQueueB()).to(MyFanoutExchange());
    }
}

• 生产者项目创建消息发送接口

/**
 * 发送消息至扇形交换机
 *
 * @return
 */
@GetMapping("sendMessageFanoutExchange")
public String sendMessageFanoutExchange() {
    String messageId = String.valueOf(UUID.randomUUID());
    String messageData = "Hello World! sendMessageFanoutExchange";
    String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy年MM月dd日 HH:mm:ss"));
    Map<String, Object> map = new HashMap<>(16);
    map.put("messageId", messageId);
    map.put("messageData", messageData);
    map.put("createTime", createTime);
    /**
     * exchange 交换机名称
     * routingKey 路由key
     * map 发送的消息内容
     */
    rabbitTemplate.convertAndSend("MyFanoutExchange", null, map);
    return "消息发送成功！";
}

• 消费者项目创建消息接收监听类
  监听队列A

package com.chentawen.rabbitmqconsumer.listener;//package com.chentawen.springbootall.config.rabbitlistener;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @Author: CTW
 * @Date: create in 2021/8/2 21:25
 */
@Component
@RabbitListener(queues = "MyFanoutQueueA")
public class FanoutReceiver {

    @RabbitHandler
    public void process(Map MessageData) {
        System.out.println("rabbitmq-consumer1接收到消息  : " + MessageData.toString());
    }

}

监听队列B

package com.chentawen.rabbitmqconsumer.listener;//package com.chentawen.springbootall.config.rabbitlistener;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @Author: CTW
 * @Date: create in 2021/8/2 21:25
 */
@Component
@RabbitListener(queues = "MyFanoutQueueB")
public class FanoutReceiver2 {

    @RabbitHandler
    public void process(Map MessageData) {
        System.out.println("rabbitmq-consumer2接收到消息  : " + MessageData.toString());
    }

}

• 启动项目，使用postman访问接口发送消息，观察消费者项目控制台

可以看到队列A、队列B都收到了消息，只要与交换机绑定了的队列就可以收到

至此，三个主流的交换机已经介绍完毕了，后续主要介绍生产者在推送消息后可能会发生的几种情况，以及如何去处理、消费者的消息确认(手动/自动)等

以上就是本期内容，后续内容持续更新