详解rabbitmq使用springboot实现fanout模式

 更新时间:2023年07月10日 10:24:14   作者:p&f°  
这篇文章主要介绍了rabbitmq使用springboot实现fanout模式,Fanout特点是发布与订阅模式,是一种广播机制,它是没有路由key的模式,需要的朋友可以参考下

一、fanout模式

  • 类型:fanout
  • 特点:Fanout—发布与订阅模式,是一种广播机制,它是没有路由key的模式。

 二、实现

1、引入相应的pom文件 pom.xml

<?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.1.5.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.xpf</groupId>
    <artifactId>rabbitmq-springboot</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>rabbitmq-springboot</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>8</java.version>
    </properties>
    <dependencies>
        <!--rabbitmq依赖-->
        <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>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter</artifactId>
            <version>RELEASE</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-api</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>

2、配置文件 application.properties

server.port=8080
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtual-host=/
spring.rabbitmq.host=192.168.199.20
spring.rabbitmq.port=5672

3、使用springboot写一个配置文件 RabbitMqConfiguration.java

(当然此配置类就是用来声明交换机类型以及交换机队列绑定关系的,将此配置类放在生产者中,或者放在消费者中都可以,或者两边都放。不管是生产者还是消费者,当生产者发送消息(或者消费者监听队列时),如果发现rabbitmq中没有所发送的交换机(或者监听的队列),都会根据此配置类自动创建交换机和队列以及绑定关系。但是一般建议下,放在消费者会更好一些,因为一般消费者在springboot系统启动时,就会开启监听,如果没有相应的rabbitmq连接有问题能在项目启动时就发现)

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;
@Configuration
public class RabbitMqConfiguration {
    //1、声明注册fanout模式交换机 (生产者发送消息给rabbitmq,其实就是发送到交换机)
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("fanout_order_exchange", true, false);
    }
    //2、声明队列 sms.fanout.queue、email.fanout.queue、duanxin.fanout.queue
    @Bean
    public Queue smsQueue(){
        return new Queue("sms.fanout.queue", true);
    }
    @Bean
    public Queue emailQueue(){
        return new Queue("email.fanout.queue", true);
    }
    @Bean
    public Queue duanxinQueue(){
        return new Queue("duanxin.fanout.queue", true);
    }
    //3、完成绑定关系(队列绑定交换机)
    @Bean
    public Binding smsBinding(){
        return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
    }
    @Bean
    public Binding emailBinding(){
        return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
    }
    @Bean
    public Binding duanxinBinding(){
        return BindingBuilder.bind(duanxinQueue()).to(fanoutExchange());
    }
}

4、写一个生产者 FanoutOrderService.java

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 FanoutOrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 模拟用户下单,发送消息给下游系统
     * @param user
     * @param num
     */
    public void makerOrder(String user,  int num){
        //1、查询库存是否有剩余
        //2、保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生产成功:" + orderId);
        //3、通过mq给下游系统发送消息(交换机的名字千万别写错,和上述配置类应相同)
        String exchangeName = "fanout_order_exchange";
        String routingkey = "";
        rabbitTemplate.convertAndSend(exchangeName, routingkey, orderId);
        System.out.println("完成");
    }
}

5、到这一步就可以再写一个测试类来测试生产者发送消息

import com.xpf.rabbitmqspringboot.service.DirectOrderService;
import com.xpf.rabbitmqspringboot.service.FanoutOrderService;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class RabbitmqSpringbootApplicationTests {
    @Autowired
    private FanoutOrderService fanoutOrderService;
    @Autowired
    private DirectOrderService directOrderService;
    /**
     * Fanout模式生成者发送消息
     */
    @Test
    void setFanoutOrderService() {
        fanoutOrderService.makerOrder("用户1", 10);
    }
}

发送完消息可以检查rabbitmq是否生成新交换机fanout_order_exchange和三个队列sms.fanout.queue、email.fanout.queue、duanxin.fanout.queue,以及是否有在三个队列看到一条刚刚发送的消息(因为是Fanout—发布与订阅模式,是一种没有路由key的广播机制,所有队列都能收到消息)

6、写一个消费者。(可以另启一个springboot项目,模拟另外的系统来消费我们发送的消息,也可像我一样在原项目编写,自己消费)

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
 * @Author xpf
 * @Date 2023/7/9 0:44
 * @Version 1.0
 * Fanout—发布与订阅模式消费者(广播机制)
 */
@Component
@RabbitListener(queues = "sms.fanout.queue")
public class smsFanoutConsumer {
    @RabbitHandler
    public void receiveMessage(String message){
        System.out.println("接收到来自队列sms.fanout.queue消息订单的message是:" + message);
    }
}

这里写了一个监听sms发送消息的消费者,剩下两个email和duanxin类似。

特别注意:如果生成者发送的消息是什么类型,消费者接收时,

public void receiveMessage(String message)方法中的参数应该也是相同的类型,否则会报错。比如这里的string类型,因为生产者发送的也是sting类型

直接启动会发现sms.fanout.queue队列中的消息被消费掉了

到此这篇关于rabbitmq使用springboot实现fanout模式的文章就介绍到这了,更多相关springboot实现fanout模式内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • Java进阶教程之String类

    Java进阶教程之String类

    这篇文章主要介绍了Java进阶教程之String类,String类对象是不可变对象(immutable object),String类是唯一一个不需要new关键字来创建对象的类,需要的朋友可以参考下
    2014-09-09
  • SpringBoot实现EMQ设备的上下线告警

    SpringBoot实现EMQ设备的上下线告警

    EMQX 的上下线系统消息通知功能在客户端连接成功或者客户端断开连接,需要实现设备的上下线状态监控,所以本文给大家介绍了如何通过SpringBoot实现EMQ设备的上下线告警,文中有详细的代码示例,需要的朋友可以参考下
    2023-10-10
  • Java Idea TranslationPlugin翻译插件使用解析

    Java Idea TranslationPlugin翻译插件使用解析

    这篇文章主要介绍了Java Idea TranslationPlugin翻译插件使用解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-04-04
  • Java中的WeakHashMap源码分析

    Java中的WeakHashMap源码分析

    这篇文章主要介绍了Java中的WeakHashMap源码分析,WeakHashMap可能平时使用的频率并不高,但是你可能听过WeakHashMap会进行自动回收吧,下面就对其原理进行分析,需要的朋友可以参考下
    2023-09-09
  • 详解Java多线程与并发

    详解Java多线程与并发

    多线程是一个进程在执行过程中产生多个更小的程序单元,这些更小的单元称为线程,这些线程可以同时存在,同时运行,一个进程可能包含多个同时执行的线程。多线程是实现并发机制的一种有效手段。进程和线程一样,都是实现并发的一个基本单位。
    2021-06-06
  • Java 中如何使用 stream 流

    Java 中如何使用 stream 流

    这篇文章主要介绍了Java中如何使用stream流,文章围绕Stream API支持的许多操作展开主题,感兴趣的小伙伴可以参考一下
    2022-09-09
  • Java方法重写_动力节点Java学院整理

    Java方法重写_动力节点Java学院整理

    在Java和其他一些高级面向对象的编程语言中,子类可继承父类中的方法,而不需要重新编写相同的方法。但有时子类并不想原封不动地继承父类的方法,而是想作一定的修改,这就需要采用方法的重写。方法重写又称方法覆盖,下文给大家介绍java方法重写及重写规则,一起学习吧
    2017-04-04
  • java selenium 操作浏览器实例

    java selenium 操作浏览器实例

    本文主要介绍java selenium 操作浏览器,这里整理了相关资料,并附示例代码,有需要的小伙伴可以参考下
    2016-08-08
  • Java数组声明、创建、初始化基础

    Java数组声明、创建、初始化基础

    本文讲述了Java数组的几个相关的方面,讲述了对Java数组的声明、创建和初始化,并给出其对应的代码
    2012-12-12
  • Spring系统属性及spring.properties配置文件示例详解

    Spring系统属性及spring.properties配置文件示例详解

    spring中有一个SpringProperties类,来保存spring的系统属性,本文结合实例代码对Spring系统属性及spring.properties配置文件相关知识给大家介绍的非常详细,需要的朋友参考下吧
    2023-07-07

最新评论