spring boot整合RabbitMQ(Fanout模式)

1.Fanout Exchange介绍
Fanout Exchange 消息广播的模式,不管路由键或者是路由模式,会把消息发给绑定给它的全部队列,如果配置了routing_key会被忽略。

如上图所示,即当使用fanout交换器时,他会将消息广播到与该交换器绑定的所有队列上,这有利于你对单条消息做不同的反应。

例如存在以下场景:一个web服务要在用户完善信息时,获得积分奖励,这样你就可以创建两个对列,一个用来处理用户信息的请求,另一个对列获取这条消息是来完成积分奖励的任务。

2.代码示例

1).Queue配置类

FanoutRabbitConfig.java类:

package com.example.rabbitmqfanout;

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 FanoutRabbitConfig {
    //创建队列
    @Bean
    public Queue AMessage() {
        return new Queue("fanout.A");
    }
    //创建队列
    @Bean
    public Queue BMessage() {
        return new Queue("fanout.B");
    }
    //创建队列
    @Bean
    public Queue CMessage() {
        return new Queue("fanout.C");
    }
    //创建Fanout交换器
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
    //将对列绑定到Fanout交换器
    @Bean
    Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(AMessage).to(fanoutExchange);
    }
    //将对列绑定到Fanout交换器
    @Bean
    Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(BMessage).to(fanoutExchange);
    }
    //将对列绑定到Fanout交换器
    @Bean
    Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(CMessage).to(fanoutExchange);
    }    
}
2).消息生产者

FanoutSender.java类:

package com.example.rabbitmqfanout.rabbitmq;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class FanoutSender {
    @Autowired
    private AmqpTemplate rabbitTemplate;

public void send() {
        String context = "hi, fanout msg ";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
    }
}

3).消息消费者

FanoutReceiverA.java类:

package com.example.rabbitmqfanout.rabbitmq;

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

@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {
     @RabbitHandler
     public void process(String message) {
         System.out.println("fanout Receiver A  : " + message);
    }
}

FanoutReceiverB.java类:

package com.example.rabbitmqfanout.rabbitmq;

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

@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {
    @RabbitHandler
    public void process(String message) {
        System.out.println("fanout Receiver B: " + message);
    }
}

FanoutReceiverC.java类:

package com.example.rabbitmqfanout.rabbitmq;

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

@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {
    @RabbitHandler
    public void process(String message) {
        System.out.println("fanout Receiver C: " + message);
    }
}
4).测试

FanoutTest.java类:

package com.example.rabbitmqfanout.rabbitmq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class FanoutTest {
    @Autowired
    private FanoutSender sender;

@Test
    public void fanoutSender() throws Exception {
        sender.send();
    }
}

时间: 2024-12-18 16:56:13

spring boot整合RabbitMQ(Fanout模式)的相关文章

Spring Boot (十三): Spring Boot 整合 RabbitMQ

1. 前言 RabbitMQ 是一个消息队列,说到消息队列,大家可能多多少少有听过,它主要的功能是用来实现应用服务的异步与解耦,同时也能起到削峰填谷.消息分发的作用. 消息队列在比较主要的一个作用是用来做应用服务的解耦,消息从消息的生产者传递到消息队列,消费者从消息队列中获取消息并进行消费,生产者不需要管是谁在消费消息,消费者也无需关注消息是由谁来生产的.在分布式的系统中,消息队列也会被用在其他地方,比如分布式事务的支持,代表如阿里开源的 RocketMQ . 当然,我们本篇文章的主角还是 Ra

Spring Boot 2.X - Spring Boot整合AMQP之RabbitMQ

文章目录Spring Boot 2.X - Spring Boot整合AMQP之RabbitMQRabbitMQ简介引入依赖编写配置编写接口启用Rabbit注解消息监听消息测试Spring Boot 2.X - Spring Boot整合AMQP之RabbitMQSpring Boot 2 整合RabbitMQ案例. RabbitMQ简介简介RabbitMQ是一个由erlang开发的AMQP(Advanved Message Queue Protocol)的开源实现.核心概念Message消息,

Spring Boot之RabbitMQ

RabbitMQ 即一个消息队列,主要是用来实现应用程序的异步和解耦,同时也能起到消息缓冲,消息分发的作用. 消息中间件在互联网公司的使用中越来越多,刚才还看到新闻阿里将 RocketMQ 捐献给了 Apache,当然了今天的主角还是讲 RabbitMQ.消息中间件最主要的作用是解耦,中间件最标准的用法是生产者生产消息传送到队列,消费者从队列中拿取消息并处理,生产者不用关心是谁来消费,消费者不用关心谁在生产消息,从而达到解耦的目的.在分布式的系统中,消息队列也会被用在很多其它的方面,比如:分布式

activeMQ入门+spring boot整合activeMQ

最近想要学习MOM(消息中间件:Message Oriented Middleware),就从比较基础的activeMQ学起,rabbitMQ.zeroMQ.rocketMQ.Kafka等后续再去学习. 上面说activeMQ是一种消息中间件,可是为什么要使用activeMQ? 在没有使用JMS的时候,很多应用会出现同步通信(客户端发起请求后需要等待服务端返回结果才能继续执行).客户端服务端耦合.单一点对点(P2P)通信的问题,JMS可以通过面向消息中间件的方式很好的解决了上面的问题. JMS规

Spring Boot整合Elasticsearch

Spring Boot整合Elasticsearch Elasticsearch是一个全文搜索引擎,专门用于处理大型数据集.根据描述,自然而然使用它来存储和搜索应用程序日志.与Logstash和Kibana一起,它是强大的解决方案Elastic Stack的一部分,我之前的一些文章中已经对此进行了描述. 保留应用程序日志不是Elasticsearch的唯一使用场景.它通常用作应用程序的辅助数据库,是一个主关系数据库.如果您必须对大型数据集执行全文搜索或仅存储应用程序不再修改的许多历史记录,这个方

Spring Boot 整合模板引擎 Freemaker、thymeleaf

1.常用的 Spring Boot 2.x 模板引擎和官方推荐案例 1)JSP(后端渲染,消耗性能) Java Server Pages 动态网页技术,由应用服务器中的 JSP 引擎来编译和执行,再将生成的整个页面返回给客户端.优点是:可以写java代码.支持表达式语言(el.jstl).内建函数. 但 JSP 本质上是 Servlet,它非常占用 JVM 内存.Java Web 官方推荐,但 Spring Boot 不推荐https://docs.spring.io/spring-boot/d

spring boot整合activeMQ

spring boot整合activeMQ spring boot整合MQ以后,对于消息的发送和接收操作更加便捷.本文将通过四个案例,分别讲解spring boot整合MQ: spring boot整合MQ发送queue消息 spring boot整合MQ发送topic消息 spring boot整合MQ以后如何让queue和topic消息共存 spring boot整合MQ以后topic消息如何持久化 下面分别进行讲解: 一. spring boot 整合MQ发送queue消息 搭建测试工程,

spring boot 整合 quartz 集群环境 实现 动态定时任务配置【原】

最近做了一个spring boot 整合 quartz  实现 动态定时任务配置,在集群环境下运行的 任务.能够对定时任务,动态的进行增删改查,界面效果图如下: 1. 在项目中引入jar 2. 将需要的表导入数据库 官网上有不同数据库的脚本,找到对应的,导入即可 3. java 代码 将quartz 的相关配置文件,配置为暴露bean,方便后期引用. 有一处关键的地方,就是注入spring 上下文,也可以算是一个坑.如果,不注入spring 上下文,那么新添加的定时任务job,是新new 的一个

spring boot 整合spring Data JPA+Spring Security+Thymeleaf框架(上)

最近上班太忙所以耽搁了给大家分享实战springboot 框架的使用. 下面是spring boot 整合多个框架的使用. 首先是准备工作要做好. 第一  导入框架所需的包,我们用的事maven 进行对包的管理. 以上的举例是本人的H5DS的真实的后台管理项目,这个项目正在盛情融资中,各位多多捧点人场.关注一下软件发展的动态,说不定以后就是您的生活不可或缺的软件哟. 点击打开链接.闲话少说.现在切入正题. 第二,写点配置文件 第三,spring data -设计一个简单的po关系,这里需要下载一