SpringBoot实现RabbitMQ三种使用方式

更新时间：2023年07月09日 12:12:44 作者：流殇꧂

本文主要介绍了SpringBoot实现RabbitMQ三种使用方式，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧

基于API的方式

1.使用AmqpAdmin定制消息发送组件

     @Autowired
    private AmqpAdmin amqpAdmin;
    @Test
    public void amqpAdmin(){
        //1.定义fanout类型的交换器
        amqpAdmin.declareExchange(new FanoutExchange("fanout_exchange"));
        //2.定义两个默认持久化队列,分别处理email和sms
        amqpAdmin.declareQueue(new Queue("fanout_queue_email"));
        amqpAdmin.declareQueue(new Queue("fanout_queue_sms"));
        //3.将队列分别与交换器进行绑定
                                                //               队列名             是队列                                 交换机的名称           路由         其它参数
        amqpAdmin.declareBinding(new Binding("fanout_queue_email",Binding.DestinationType.QUEUE,"fanout_exchange","",null));
        amqpAdmin.declareBinding(new Binding("fanout_queue_sms",Binding.DestinationType.QUEUE,"fanout_exchange","",null));
    }

2.消息发送者发送消息

创建实体类

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class User {
    private Integer id;
    private String name;
}

发送消息

    @Autowired
    private RabbitTemplate re;
    @Test//消息发送者
    public void subPublisher(){
        User user = new User(1,"小满");
        re.convertAndSend("fanout_exchange", "", user);
    }

如图所以,如果我们直接发送的话就会报这个错,有两种解决方法,第一种是比较常用的让实体类User实现序列化Serializable接口,这里我们不做演示,第二种是写一个配置类,只有在RabbitMQ可以使用

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    //定制JSON格式的消息转化器
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

加上配置类后我们发送就不会报错了,我们也可以在RabbitMQ的可视化端口看到我们发送的消息

3.发送完消息后接下来就是消费消息了,定义接收消息的业务

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
@Service
public class RabbitMQService {
    //发布订阅模式: @RabbitListener可以指定当前方法监控哪一个队列
   @RabbitListener(queues = "fanout_queue_email")//消费者可以消费多个队列的消息 
    public void subConsumerEmail(Message message){
        //当队列中有内容是方法会自动执行   推荐Object来接收
        //官网推荐Message
        byte[] body = message.getBody();//Message将数据存放在body中
        String msg = new String(body);
        System.out.println("邮件业务接收到消息:"+msg);
    }
    @RabbitListener(queues = "fanout_queue_sms") 
    public void subConsumerSms(Message message){
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("短信业务接收到消息:"+msg);
    }
}

4.重新运行发送端就可以接收到我们发送的数据,接收的数据可能打印在任意一个控制台中,这是idea的机制,我们不需要管

基于配置类的方式

1.在config配置类中定义

 
import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    //定制JSON格式的消息转化器
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
   // 1.fanout创建一个交换机
    @Bean
    public Exchange fanoutExchange(){
        return ExchangeBuilder.fanoutExchange("fanout_exchange").build();
    }
    //2.定义消息队列
    @Bean
    public Queue fanoutQueueEmail(){
        return new Queue("fanout_queue_email");
    }
    @Bean
    public Queue fanoutQueueSms(){
        return new Queue("fanout_queue_sms");
    }
    //3.将创建的队列绑定到对应的交换机上
    @Bean
    public Binding bingingEmail(){
        return BindingBuilder.bind(fanoutQueueEmail()).to(fanoutExchange()).with("").noargs();
    }
    @Bean
    public Binding bingingSms(){
        return BindingBuilder.bind(fanoutQueueSms()).to(fanoutExchange()).with("").noargs();
    }
}

2.为了避免api的影响,我们可以在可视化端口将基于api创建的交换机和队列删除

1)删除交换机

2)删除队列,前面也是点击队列的名字

可以看到我已经将交换机和消息队列都已经删除,接下来我们重新启动项目 ,配置类可以在启动的时候自动创建

我们的订阅发布模式也是可以正常运行

基于注解类的方式

1.我们要现将基于配置类的方式注释掉,避免影响我们测试

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;
@Service
public class RabbitMQService { 
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("fanout_queue_email"),
            exchange=@Exchange(value = "fanout_exchange",type = "fanout")
    ))
    public void subConsumerEmail(Message message){
        //当队列中有内容是方法会自动执行   推荐Object来接收
        //官网推荐Message
        byte[] body = message.getBody();//Message将数据存放在body中
        String msg = new String(body);
        System.out.println("邮件业务接收到消息:"+msg);
    }
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("fanout_queue_sms"),
            exchange=@Exchange(value = "fanout_exchange",type = "fanout")
    ))
    public void subConsumerSms(Message message){
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("短信业务接收到消息:"+msg);
    }
}

提前将交换机和队列删除,然后运行,就会发现会在启动时会自动生成交换机和队列,测试也不会有影响

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

您可能感兴趣的文章:

SpringBoot-Session管理问题及处理
本文介绍了SpringBoot中Session管理的多种实现方式,包括内存、Redis、数据库和缓存等,并详细分析了Session失效、并发访问、共享、数据丢失和性能问题,以及相应的解决策略,如延长超时时间、使用分布式Session、优化存取频率和确保Session安全等
2026-04-04
Java list foreach修改元素方式
这篇文章主要介绍了Java list foreach修改元素方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2023-11-11
Java中常用的四种引用类型详解
Java中常用的四种引用类型，分别为，强引用、软引用、弱引用以及虚引用，这篇文章主要为大家介绍了这四种引用的用法，需要的可以参考一下
2023-06-06
微服务之Feign的介绍与使用小结
Feign 是一个声明式的伪RPC的REST客户端，它用了基于接口的注解方式，很方便的客户端配置，Spring Cloud 给 Feign 添加了支持Spring MVC注解，并整合Ribbon及Eureka进行支持负载均衡，这篇文章主要介绍了微服务之Feign的介绍与使用,需要的朋友可以参考下
2022-07-07
Java 中通过 key 获取锁的方法
这篇文章主要介绍了Java 中通过 key 获取锁,本文演示如何对某个 key 加锁，以保证对该 key 的并发操作限制，可以实现同一个 key 一个或者多个线程同时执行，需要的朋友可以参考下
2022-11-11
SqlSession的创建与生命周期源码解读
文章解析了MyBatis SqlSession的核心作用、工厂模式、Executor类型及生命周期管理,说明其设计优势与性能优化,为后续源码分析奠定基础,本文给大家详细介绍SqlSession的创建与生命周期,感兴趣的朋友一起看看吧
2025-09-09
Java如何将ResultSet结果集遍历到List中
这篇文章主要介绍了Java如何将ResultSet结果集遍历到List中问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-02-02
剖析SpringCloud Feign中所隐藏的坑
这篇文章主要为大家介绍了剖析SpringCloud Feign中所隐藏的坑示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2022-08-08
Java使用WebView实现桌面程序的技术指南
在现代软件开发中,许多应用需要在桌面程序中嵌入 Web 页面,例如,你可能需要在 Java 桌面应用中嵌入一部分 Web 前端,或者加载一个 HTML5 界面以增强用户体验,所以本文给大家介绍了Java使用WebView实现桌面程序的技术指南,需要的朋友可以参考下
2025-05-05
SpringBoot结合Redis实现序列化的方法详解
Spring提供了一个RedisTemplate来进行对Redis的操作，但是RedisTemplate默认配置的是使用Java本机序列化。如果要对对象操作，就不是那么的方便。所以本文为大家介绍了另一种SpringBoot结合Redis实现序列化的方法，需要的可以参考一下
2022-06-06