当前位置： 首页 > news >正文

Springboot整合RabbitMQ

news 来源：原创 2024/4/29 9:20:47

目录

1、创建Springboot项目、导入依赖

2、配置文件中添加rabbitmq的连接信息

3、编写rabbitmq的配置类

4、编写投递消息接口

5、编写邮件消费者

6、编写短信消费者

7、测试

8、使用两个邮件消费者监听邮件队列

9、实现能者多劳和手动消息确认

1、创建Springboot项目、导入依赖

 <dependencies>
        <!-- springboot-web组件 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- 添加springboot对amqp的支持 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </dependency>
        <!--fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.49</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

2、配置文件中添加rabbitmq的连接信息

spring:
  rabbitmq:
    #连接地址
    host: 127.0.0.1
    #端口号
    port: 5672
    #账号
    username: guest
    #密码
    password: guest
    #虚拟主机
    virtual-host: /rkVirtualHost
server:
  port: 8080

3、编写rabbitmq的配置类

package com.rk.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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


/**
 * @author :Rk.
 * @date : 2022/10/8
 * 定义完成后 rabbitmq服务器会自动创建交换机和队列 以及绑定关系
 */
@Configuration
public class RabbitMQConfig {


    /**
     * 定义交换机名称
     */
    private String EXCHANGE_NAME="springboot_falout_exchange";

    /**
     * 短信队列
     */
    private String FANOUT_SMS_QUEUE = "fanout_sms_queuerk";

    /**
     * 邮件队列
     */
    private String FANOUT_EMAIL_QUEUE = "fanout_email_queuerk";


    /**
     * 定义 扇形交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        //参数 交换机名称
        return new FanoutExchange(EXCHANGE_NAME);
    }

    /**
     *参数一:队列名称
     *参数二durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     *参数三exclusive:默认也是false，是否独占队列
     *参数四autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     */
    @Bean
    public Queue smsQueue(){
        return new Queue(FANOUT_SMS_QUEUE,true, false,false);
    }

    /**
     *参数一:队列名称
     *参数二:durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
     *参数三:exclusive:默认也是false，是否独占队列
     *参数四:autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
     */
    @Bean
    public Queue emailQueue(){
        return new Queue(FANOUT_EMAIL_QUEUE,true,false,false);
    }

    /**
     * 短信队列绑定交换机
     * @param smsQueue   短信队列注入到容器的id，也就是方法名 sumQueue
     * @param fanoutExchange   交换机注入到容器的id，也就是方法名 fanoutExchange
     * @return
     */
    @Bean
    public Binding bindingSmsFanoutExchange(Queue smsQueue,FanoutExchange fanoutExchange){
        return  BindingBuilder.bind(smsQueue).to(fanoutExchange);
    }

    /**
     * 邮件队列队列绑定交换机
     * @param emailQueue   短信队列注入到容器的id，也就是方法名 emailQueue
     * @param fanoutExchange   交换机注入到容器的id，也就是方法名 fanoutExchange
     * @return
     */
    @Bean
    public Binding bindingEmailFanoutExchange(Queue emailQueue,FanoutExchange fanoutExchange){
        return  BindingBuilder.bind(emailQueue).to(fanoutExchange);
    }

}

        这里使用的是扇形交换机，交换机会把消息路由到绑定到它的所有队列。

        在Spring启动时，利用Spring Bean管理工厂BeanFactory接口，实现动态创建交换机、队列、交换机和队列的绑定关系，让我们无需进行重复的编码工作。

4、编写投递消息接口

package com.rk.controller;
import com.rk.domain.User;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author :Rk.
 * @date : 2022/10/8
 */

@RestController
public class FanoutProducerController {

    @Autowired
    private AmqpTemplate amqpTemplate;

    /**
     * 模拟注册接口
     * @param
     * @return
     */
    @GetMapping("/registered")
    public String sendMsg(){

        User user=new User("rk",22,"男","陕西西安");
        //1、将user存入数据库

        //2、发送邮件和短信
        //投递消息到交换机   交换机将消息路由到邮件队列和短信队列

        /**
           * 参数一  交换机名称
           * 参数二  路由key
           * 参数三  发送内容
         */
        amqpTemplate.convertAndSend("springboot_falout_exchange","",user);

        return "注册完成";
    }

}

        这里直接发送的user对象，必须要将user对象实现Serializable接口。

5、编写邮件消费者

import com.rk.domain.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author :Rk.
 * @date : 2022/10/8
/* */
@Slf4j
@Component
@RabbitListener(queues = "fanout_email_queuerk") //监听队列
public class EmailConsumer {

    //监听到队列消息后执行
    @RabbitHandler
    public void process(User user) {
        System.err.println("邮件消费者接收到消息发送邮件:"+user);
    }

}

6、编写短信消费者

import com.rk.domain.User;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author :Rk.
 * @date : 2022/10/8
 */

@Component
@RabbitListener(queues = "fanout_sms_queuerk")
public class SumConsumer {
    @RabbitHandler
    public void process(User user){
        System.out.println("短信消费者接收到消息发送短信:"+user);
    }
}

7、测试

  访问注册接口，会立即响应。

8、使用两个邮件消费者监听邮件队列

邮件消费者1:

import com.rk.domain.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author :Rk.
 * @date : 2022/10/8
/* */
@Slf4j
@Component
@RabbitListener(queues = "fanout_email_queuerk") //监听队列
public class EmailConsumer1 {


    //监听到队列消息后执行
    @RabbitHandler
    public void process(User user) throws InterruptedException {
        //休眠10ms  模拟处理时间
        Thread.sleep(10);
        System.err.println("邮件消费者1接收到消息发送邮件:"+user);
    }

}

邮件消费者2:

import com.rk.domain.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @author :Rk.
 * @date : 2022/10/8
/* */
@Slf4j
@Component
@RabbitListener(queues = "fanout_email_queuerk")
public class EmailConsumer2 {

    @RabbitHandler
    public void process(User user) throws InterruptedException {
        //休眠20ms  模拟处理时间
        Thread.sleep(20);
        System.out.println("邮件消费者2接收到消息发送邮件:"+user);
    }

}

测试: 生产者往交换机发送20条消息:

        两个消费者使用轮询的方式对消息进行消费，而且不存在重复消费。 这里消费者1的处理时间是10ms，消费者2的处理时间为20ms，所以使用轮询的方式来进行消费并不合理，需要实现能者多劳。

9、实现能者多劳和手动消息确认

更改配置文件的参数来解决这一个问题：

邮件消费者1:

package com.rk.service;
import com.rabbitmq.client.Channel;
import com.rk.domain.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author :Rk.
 * @date : 2022/10/8
/* */
@Slf4j
@Component
@RabbitListener(queues = "fanout_email_queuerk")
public class EmailConsumer1 {

    @RabbitHandler
    public void process(User user, Message message, Channel channel) throws IOException {
        try {
             //休眠20ms  模拟处理时间
            Thread.sleep(10);
            System.err.println("邮件消费者1接收到消息发送邮件:"+user);
            //确认一条消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }catch (Exception e){
            //消费者告诉队列信息消费失败
            /**
             * 拒绝确认消息:
             * channel.basicNack(long deliveryTag, boolean multiple, boolean requeue) ;
             * deliveryTag:该消息的index
             * multiple：是否批量true:将一次性拒绝所有小于deliveryTag的消息
             * requeue：被拒绝的是否重新入队列
             */
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}

邮件消费者2:

import com.rabbitmq.client.Channel;
import com.rk.domain.User;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author :Rk.
 * @date : 2022/10/8
/* */
@Slf4j
@Component
@RabbitListener(queues = "fanout_email_queuerk")
public class EmailConsumer4 {

    @RabbitHandler
    public void process(User user, Message message, Channel channel) throws IOException {
        try {
            //确认一条消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            //休眠20ms  模拟处理时间
            Thread.sleep(20);
            System.out.println("邮件消费者2接收到消息发送邮件:"+user);
        }catch (Exception e){
            //消费者告诉队列信息消费失败
            /**
             * 拒绝确认消息:
             * channel.basicNack(long deliveryTag, boolean multiple, boolean requeue) ;
             * deliveryTag:该消息的index
             * multiple：是否批量true:将一次性拒绝所有小于deliveryTag的消息
             * requeue：被拒绝的是否重新入队列
             */
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }

}

        实现能者多劳和手动确认。上面的演示过程都是将生产者和消费者写在同一个项目中，在实际开发过程中，一般使用2个项目，一个项目来投递消息，一个专门用来消费消息，使用不同的JVM来运行生产者和消费者。

整合直连交换机和主题交换机可以看下面这篇博客:

Springboot 整合RabbitMq ，用心看完这一篇就够了_小目标青年的博客-CSDN博客_springboot整合rabbitmq

相关文章：

  • 量化投资交易接口究竟能否取缔手工交易？
  • didChangeDependencies什么时候被调用
  • 【测试面试】2022年软件测试面试题大全（精品带回答的）
  • Java --- JVM年轻代与老年代
  • 制作地图的布局、元素和设计介绍
  • C/C++面试准备——运算符
  • 【华为机考】ACM输入输出（中等+）
  • 美团四年，字节三年这七年测试之路希望能让正在迷茫的你少走弯路
  • Vue 基础
  • Ts/Typescript基础运用
  • 嵌入注意力机制的多尺度深度可分离表情识别--2021.宋玉琴
  • 页面登录功能的思路
  • ShardingSphere 5.2.0：分片审计功能拦截多分片场景下的不合理请求
  • 毕业设计 单片机stm32智能大棚监控护理系统 - lora 远程通信
  • 关于QCefView的一些事
  • 〔开发系列〕一次关于小程序开发的深度总结
  • Angularjs之国际化
  • electron原来这么简单----打包你的react、VUE桌面应用程序
  • Git初体验
  • java多线程
  • Java方法详解
  • Js基础知识（四） - js运行原理与机制
  • JS正则表达式精简教程（JavaScript RegExp 对象）
  • node-glob通配符
  • python docx文档转html页面
  • REST架构的思考
  • SQLServer之索引简介
  • vue.js框架原理浅析
  • Vue--数据传输
  • 基于Android乐音识别(2)
  • 前端知识点整理（待续）
  • 如何学习JavaEE,项目又该如何做？
  • 深度学习在携程攻略社区的应用
  • 使用 Docker 部署 Spring Boot项目
  • 在 Chrome DevTools 中调试 JavaScript 入门
  • 【干货分享】dos命令大全
  • 通过调用文摘列表API获取文摘
  • ​ArcGIS Pro 如何批量删除字段
  • ​低代码平台的核心价值与优势
  • ​软考-高级-信息系统项目管理师教程 第四版【第23章-组织通用管理-思维导图】​
  • # Python csv、xlsx、json、二进制(MP3) 文件读写基本使用
  • ### Error querying database. Cause: com.mysql.jdbc.exceptions.jdbc4.CommunicationsException
  • #单片机（TB6600驱动42步进电机）
  • (+3)1.3敏捷宣言与敏捷过程的特点
  • (html5)在移动端input输入搜索项后 输入法下面为什么不想百度那样出现前往？ 而我的出现的是换行...
  • (非本人原创)我们工作到底是为了什么？​——HP大中华区总裁孙振耀退休感言（r4笔记第60天)...
  • （入门自用）--C++--抽象类--多态原理--虚表--1020
  • （学习日记）2024.02.29：UCOSIII第二节
  • (一）使用Mybatis实现在student数据库中插入一个学生信息
  • (转)linux自定义开机启动服务和chkconfig使用方法
  • (转)视频码率,帧率和分辨率的联系与区别
  • （转）微软牛津计划介绍——屌爆了的自然数据处理解决方案(人脸/语音识别，计算机视觉与语言理解)...
  • （转载）跟我一起学习VIM - The Life Changing Editor
  • .NET : 在VS2008中计算代码度量值
  • .NET “底层”异步编程模式——异步编程模型（Asynchronous Programming Model，APM）...