[Java知识库] SpringBoot整合RabbitMQ

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> Java知识库 -> SpringBoot整合RabbitMQ -> 正文阅读

[Java知识库]SpringBoot整合RabbitMQ

文章目录

参考蚂蚁课堂

1.简介

我们的生产者向消息队列中投递消息，不用非得把他转化成json格式，可以直接投递对象，因为SpringBoot已经帮你实现了序列化，你投递的对象只要实现Seralized接口就行了。同时SpringBoot还可以帮你自动创建交换机，队列这些组件，你不用SpringBoot的话你不创建他就会报错。SpringBoot就相当于一个小女友，十分的贴心处处为你着想。

2.准备工作

在这里插入图片描述

我们大概要实现一个这样的效果，一个生产者，通过fanout交换机将消息投递给两个消息队列当中，然后两个消费者获取队列当中的消息。所以我们需要3个微服务一个生产者服务，两个消费者服务。

2.1生产者

生产者配置信息

首先我们要在配置类中规定一些东西，将队列交换机之类的注入到Spring容器当中,然后将队列和交换机进行绑定

@Component
public class RabbitMQConfig {

    /**
     * 定义交换机
     */
    private String EXCHANGE_SPRINGBOOT_NAME = "fanout_exchange";


    /**
     * 短信队列
     */
    private String FANOUT_SMS_QUEUE = "fanout_sms_queue";
    /**
     * 邮件队列
     */
    private String FANOUT_EMAIL_QUEUE = "fanout_email_queue";

    // 1.注入队列和交换机注入到spring容器中
    // 2.关联交换机  <bean id="smsQueue" class="";>

    /**
     * 邮件和短信队列注入到spring容器中
     *
     * @return
     */
    @Bean
    public Queue smsQueue() {
        return new Queue(FANOUT_SMS_QUEUE);
    }

    @Bean
    public Queue emailQueue() {
        return new Queue(FANOUT_EMAIL_QUEUE);
    }

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(EXCHANGE_SPRINGBOOT_NAME);
    }

    /**
     * 关联交换机
     * 根据参数名称 ioc获取 Queue对象
     */
    @Bean
    public Binding BindingSmsFanoutExchange(Queue smsQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(smsQueue).to(fanoutExchange);
    }

    @Bean
    public Binding BindingEmailFanoutExchange(Queue emailQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(emailQueue).to(fanoutExchange);
    }
}

由于生产者的作用是向队列当中投递消息，所以我们写一下他的service类，投递信息的方法。

service投递信息

@RestController
public class ProducerService {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @RequestMapping("/sendMsg")
    public void sendMsg() {
        /**
         * 参数1 交换机名称
         * 参数2 路由key
         * 参数3 发送内容
         */
        MsgEntity msgEntity = new MsgEntity(UUID.randomUUID().toString(),
                "1234", "181111111", "2197206001@qq.com");
        amqpTemplate.convertAndSend("fanout_exchange", "", msgEntity);
    }
}

实体类

@Data
public class MsgEntity implements Serializable {
    private String msgId;
    private String userId;
    private String phone;
    private String email;

    public MsgEntity(String msgId, String userId, String phone, String email) {
        this.msgId = msgId;
        this.userId = userId;
        this.phone = phone;
        this.email = email;
    }
}

这个实体类一定要实现Serializable接口

yml文件

spring:
  rabbitmq:
    ####连接地址
    host: 127.0.0.1
    ####端口号
    port: 5672
    ####账号
    username: guest
    ####密码
    password: guest
    ###
    virtual-host: /wjzVirtualHost
server:
  port: 9092

启动类
```
@SpringBootApplication
public class AppProducer {
    public static void main(String[] args) {
        SpringApplication.run(AppProducer.class);
    }
}
```
然后我们启动这个服务然后访问到sendMsg这个方法http://localhost:9092/sendMsg，看看我们的消息有没有投递成功。

通过查看控制台我们可以看到消息投递成功。

2.2邮件消费者

消费者这一端只需要消费队列中的消息。所以我们以邮件消费者为例。

@Slf4j
@Component
@RabbitListener(queues = "fanout_email_queue")
public class FanoutEmailConsumer {

    @RabbitHandler
    public void process(MsgEntity msgEntity) {
        log.info("email：msgEntity:" + msgEntity);
    }
}

最主要的方法就是这个通过@RabbitListener注解说明要从哪个队列里获取消息进行消费。然后在方法上加上@RabbitHandler这个注解，就能够消费消息队列里的消息。

yml

spring:
  rabbitmq:
    ####连接地址
    host: 127.0.0.1
    ####端口号
    port: 5672
    ####账号
    username: guest
    ####密码
    password: guest
    ###
    virtual-host: /wjzVirtualHost

启动类

@SpringBootApplication
public class AppEmailConsumer {
    public static void main(String[] args) {
        SpringApplication.run(AppEmailConsumer.class);
    }
}