[Java知识库]RabbitMQ-SpringBoot案例 -fanout模式

1、创建生产者工程

1、在pom.xml中引入依赖

<dependencies>
        <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>

2、配置application.yml

server:
  port: 8080

spring:
  # RabbitMQ配置
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: admin
    virtual-host: /

3、RabbitTemplate 配置类

package com.sky.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.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author 尹稳健~
 * @version 1.0
 */
@Configuration
public class RabbitMQConfig {

    // 声明注册交换机
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanout_order_exchange");
    }

    // 声明队列
    @Bean
    public Queue smsQueue(){
        return new Queue("sms.fanout.queue",true);
    }

    @Bean
    public Queue duanxinQueue(){
        return new Queue("duanxin.fanout.queue",true);
    }

    @Bean
    public Queue emailQueue(){
        return new Queue("email.fanout.queue",true);
    }

    // 交换机和队列绑定
    @Bean
    public Binding smsBinding(){
        return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
    }

    @Bean
    public Binding duanxinBinding(){
        return BindingBuilder.bind(duanxinQueue()).to(fanoutExchange());
    }

    @Bean
    public Binding emailBinding(){
        return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
    }

}

4、发送订单

package com.sky.service.impl;

import com.sky.service.OrderService;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.UUID;
@Service
public class OrderServiceImpl implements OrderService {	
	@Autowired
    private RabbitTemplate rabbitTemplate;

    @Override
    public void makeOrder(String userId, String productId, int num) {
        // 1:根据商品id查询库存是否充足

        //2:保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生成成功"+orderId);
        //3：通过MQ来完成消息的分发
        // 参数1 ： 交换机 参数2：路由key / 队列名称  参数3 ：消息内容
        String exchangeName = "fanout_order_exchange";
        String routeKey = "";

        rabbitTemplate.convertAndSend(exchangeName,routeKey,orderId);
    }	
}

2、创建消费者工程

1、导入pom依赖

<dependencies>
        <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>

2、配置application.yml

server:
  port: 8081

spring:
  # RabbitMQ配置
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: admin
    virtual-host: /

3、编写接收消息service

消费者短信服务：

package com.sky.service.fanout;

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

/**
 * @author 尹稳健~
 * @version 1.0
 */
@Service
@RabbitListener(queues = {"duanxin.fanout.queue"})
public class FanoutDuanxinConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("duanxin fanout ---->"+message);
    }
}

消费者邮件服务：

package com.sky.service.fanout;

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

/**
 * @author 尹稳健~
 * @version 1.0
 */
@Service
@RabbitListener(queues = {"email.fanout.queue"})
public class FanoutEmailConsumer {

    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("duanxin fanout ---->"+message);
    }

}

消费者短信息服务:

package com.sky.service.fanout;

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

/**
 * @author 尹稳健~
 * @version 1.0
 */
@Service
@RabbitListener(queues = {"sms.fanout.queue"})
public class FanoutSMSConsumer {

    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("sms fanout ---->"+message);
    }
}

3、测试

1、先开启生产者通过exchange（交换机）发送消息

还有一条没截到。。。

2、开启消费者

测试通过！

小结

看视频时看到老师说到一个面试题：

问到交换机的创建和队列声明还有绑定应该在消费者中还是生成者中？

• 老师给出的答案是消费者中，因为防止用户访问的时候队列没创建或者没绑定报错，应该丢到消费者中
• 但是我个人觉得两边都可以，因为一般都是先开启服务，再提供给消费者。

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖