Springboot与RabbitMQ上手学习之Fanout模式(二)

前言

目的主要是学习RabbitMQ其中一种Fanout交换机，大概会简单介绍学习为主：毕竟还是要来演示Springboot整合RabbitMQ注解的方式来使用

一.Fanout交换机模式

1.旁白

Fanout是Fanout(广播路由): 群发绑定的所有队列，适用于消息广播，是一对多的关系。
简单的来说不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到该类型交换机的消息都会被广播到与该交换机绑定的所有队列上。
但是这种模式收看电视直播或电台直播是一样的，必须要先在消费中监听队列，就是我们要在节目开始前打开电视机，调到对应的频道。否则就会错过。

2.图说

红色：Producer代表着生产者：也就是发消息的一端,
绿色：Consumer代表着消费生产者的消息
蓝色：直接通过交换机去绑定队列而不需要routingKey就可以用消费端去消费生产者的消息

image.png

二.Springboot整合Rabbimq实现fanout

准备创建工程项目，目录结构如下：

image.png
2.1 统一配置pom.xml依赖
  1. 父工程
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.4.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.mi</groupId>
    <artifactId>springboot-rabbitmq-day1</artifactId>
    <packaging>pom</packaging>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>springboot-send-rabbitmq</module>
        <module>springboot-recive-rabbitmq</module>
    </modules>
  1. 发送工程和接受工程一样
    <parent>
        <artifactId>springboot-rabbitmq-day1</artifactId>
        <groupId>com.mi</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.mi</groupId>
    <artifactId>springboot-send-rabbitmq</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
            <version>2.1.2.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>2.1.1</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
    </dependencies>
2.2 统一配置 application.properties
#发送端8082，接受端8081
server.port=8082
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/food?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=GMT%2B8
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.rabbitmq.address=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

#消息确认机制
spring.rabbitmq.listener.direct.acknowledge-mode=auto
2.3 统一配置 application启动器

发送端和接受端基本一样

@SpringBootApplication
@MapperScan(value = "com.xxx.xxx",annotationClass = Mapper.class)
@ComponentScan("com.xxx.xxx")
public class ReciveApplication {
    public static void main(String args[]) {
        SpringApplication.run(ReciveApplication.class, args);
    }
}
3 Send发送端工程
3.1 config包

1）配置连接Rabbit连接
两种配置：以防漏掉些什么

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setPassword("guest");
        connectionFactory.setUsername("guest");
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setPublisherReturns(true);
        connectionFactory.createConnection();
        return connectionFactory;
    }

第二种:在之前application.properties里面配置连接地址和端口，然后RabbitListenerContainerFactory 去自动去连接配置里面地址和端口

   @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
  1. 声明fanout交换机，队列，路由键

    @Bean
    public Exchange fanoutExchange(){
        return new FanoutExchange("fanout.exchange.test");
    }

    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue.test1");
    }

    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue.test2");
    }

    @Bean
    public Queue fanoutQueue3(){
        return new Queue("fanout.queue.test3");
    }


    @Bean
    public Binding fanoutBinding1(){
        return new Binding("fanout.queue.test1",
                Binding.DestinationType.QUEUE,
                "fanout.exchange.test",
                "",null);
    }


    @Bean
    public Binding fanoutBinding2(){
        return new Binding("fanout.queue.test2",
                Binding.DestinationType.QUEUE,
                "fanout.exchange.test",
                "",null);
    }


    @Bean
    public Binding fanoutBinding3(){
        return new Binding("fanout.queue.test3",
                Binding.DestinationType.QUEUE,
                "fanout.exchange.test",
                "",null);
    }
  1. 配置RabbitTemplate回调方法和发送方法确认方法
    @Bean
    @Qualifier("rabbitTemplate")
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //开启mandatory模式（开启失败回调）
        rabbitTemplate.setMandatory(true);
        //添加失败回调方法
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("message:{}, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}",
                    message, replyCode, replyText, exchange, routingKey);
        });
        // 添加发送方确认模式方法
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) ->
                log.info("correlationData:{}, ack:{}, cause:{}",
                        correlationData.getId(), ack, cause));
        return rabbitTemplate;
    }

4):整个config

@Component
@Slf4j
public class RabbitListenerConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setPassword("guest");
        connectionFactory.setUsername("guest");
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setPublisherReturns(true);
        connectionFactory.createConnection();
        return connectionFactory;
    }

        @Bean
        public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            return factory;
        }
    @Bean
    @Qualifier("rabbitTemplate")
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //开启mandatory模式（开启失败回调）
        rabbitTemplate.setMandatory(true);
        //添加失败回调方法
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("message:{}, replyCode:{}, replyText:{}, exchange:{}, routingKey:{}",
                    message, replyCode, replyText, exchange, routingKey);
        });
        // 添加发送方确认模式方法
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) ->
                log.info("correlationData:{}, ack:{}, cause:{}",
                        correlationData.getId(), ack, cause));
        return rabbitTemplate;
    }


 
    /**声明 fanout   一对多队列**/

    @Bean
    public Exchange fanoutExchange(){
        return new FanoutExchange("fanout.exchange.test");
    }

    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue.test1");
    }

    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue.test2");
    }

    @Bean
    public Queue fanoutQueue3(){
        return new Queue("fanout.queue.test3");
    }


    @Bean
    public Binding fanoutBinding1(){
        return new Binding("fanout.queue.test1",
                Binding.DestinationType.QUEUE,
                "fanout.exchange.test",
                "",null);
    }


    @Bean
    public Binding fanoutBinding2(){
        return new Binding("fanout.queue.test2",
                Binding.DestinationType.QUEUE,
                "fanout.exchange.test",
                "",null);
    }


    @Bean
    public Binding fanoutBinding3(){
        return new Binding("fanout.queue.test3",
                Binding.DestinationType.QUEUE,
                "fanout.exchange.test",
                "",null);
    }
3.2 dto包
@Getter
@Setter
@ToString
public class OrderMessageDTO {
    private Integer orderId;
    private BigDecimal price;
    private Integer productId;
}
3.3 service包
public interface FanoutService {

    public void sendMessage();
}

@Slf4j
@Service
public class FanoutServiceImpl implements FanoutService {

    @Autowired
    private RabbitTemplate rabbitTemplate;


    ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void sendMessage() {
        log.info("==========发送Fanout类型消息=======");
        try {
            // 第一种方式
            OrderMessageDTO orderMessageDTO = new OrderMessageDTO();
            orderMessageDTO.setOrderId(1);
            orderMessageDTO.setPrice(new BigDecimal("20"));
            orderMessageDTO.setProductId(100);
            String messageToSend = objectMapper.writeValueAsString(orderMessageDTO);
            // 发送端确认是否确认消费
            CorrelationData correlationData = new CorrelationData();
            // 唯一ID
            correlationData.setId(orderMessageDTO.getOrderId().toString());
            // 发送
                        rabbitTemplate.convertAndSend("fanout.exchange.test","fanout.queue.test",messageToSend,correlationData);
            log.info("发送成功");
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
    }
}
3.4 controller包
@RestController
@Slf4j
@RequestMapping("/api")
public class SendController {

    @Autowired
    private FanoutService fanoutService;
    @GetMapping
    public void sendOrder(){
        for (int i = 0; i < 9000; i++) {
            fanoutService.sendMessage();
        }

    }
}

4. Receive 消费端工程

由于多个Receive基本都一样，在此就只写一个消费端，另外两个自行去复制就行
#######4.1config包
同上，选择一个连接RabbitMQ工厂

@Component
@Slf4j
public class RabbitListenerConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost("localhost");
        connectionFactory.setPort(5672);
        connectionFactory.setPassword("guest");
        connectionFactory.setUsername("guest");
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setPublisherReturns(true);
        connectionFactory.createConnection();
        return connectionFactory;
    }

    @Bean
    public RabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
}

#######4.2 service包

public interface FanoutReciveService {

    public void FanoutRecive(Message message);
}

@Service
@Slf4j
public class FanoutReciveServiceImpl implements FanoutReciveService{


    @RabbitListener(
            containerFactory = "rabbitListenerContainerFactory",
            bindings = {
                    @QueueBinding(
                            value = @Queue(name = "fanout.queue.test"),
                            exchange = @Exchange(name = "fanout.exchange.test",
                                    type = ExchangeTypes.DIRECT),
                            key = ""
                    )
            }
    )
    @Override
    public void FanoutRecive(@Payload Message message) {
        log.info("========Fanout接受消息===========");
        String messageBody = new String(message.getBody());
        log.info(" body = {} " ,messageBody);
    }
}

5. 启动发送端和接收端工程

1)访问：发送端地址http://localhost:8082/api
2)发送端：

correlationData：1 确认返回的标认ID，
ack：true 确认发送端的发出被消息返回的确认
cause:暂时未知

2021-05-01 15:43:29.840  INFO 17260 --- [nio-8082-exec-4] c.m.send.service.impl.FanoutServiceImpl  : ==========发送Fanout类型消息=======
2021-05-01 15:43:29.849  INFO 17260 --- [nio-8082-exec-4] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: 42.194.178.240:5672
2021-05-01 15:43:29.908  INFO 17260 --- [nio-8082-exec-4] o.s.a.r.c.CachingConnectionFactory       : Created new connection: connectionFactory#4fbb001b:1/SimpleConnection@310a0150 [delegate=amqp://guest@42.194.178.240:5672/, localPort= 9323]
2021-05-01 15:43:30.114  INFO 17260 --- [nio-8082-exec-4] c.m.send.service.impl.FanoutServiceImpl  : 发送成功
2021-05-01 15:43:30.138  INFO 17260 --- [nectionFactory3] com.mi.send.config.RabbitListenerConfig  : correlationData:1, ack:true, cause:null

3):接收端一
消费发送端的Message

2021-05-01 01:43:43.984  INFO 13516 --- [ntContainer#0-1] c.m.r.s.Impl.DirectReciveServiceImpl     : ========direct接受消息===========
2021-05-01 01:43:43.984  INFO 13516 --- [ntContainer#0-1] c.m.r.s.Impl.DirectReciveServiceImpl     :  body = {"orderId":1,"price":20,"productId":100}

4):接收端二

2021-05-01 15:43:30.179  INFO 16440 --- [ntContainer#1-2] c.m.r.service.Impl.FanoutServiceImpl1    : ========fanout接受消息===========
2021-05-01 15:43:30.183  INFO 16440 --- [ntContainer#1-2] c.m.r.service.Impl.FanoutServiceImpl1    :  body = {"orderId":1,"price":20,"productId":100}

5):接收端三

2021-05-01 15:43:30.179  INFO 16440 --- [ntContainer#1-2] c.m.r.service.Impl.FanoutServiceImpl1    : ========fanout接受消息===========
2021-05-01 15:43:30.183  INFO 16440 --- [ntContainer#1-2] c.m.r.service.Impl.FanoutServiceImpl1    :  body = {"orderId":1,"price":20,"productId":100}

6.结语

Springboot与RabbitMQ上手学习之Fanout模式就到此为止

©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容