SpringBoot整合消息中间件（ActiveMQ，RabbitMQ，RocketMQ，Kafka）

消息中间件

  • 消息
    • 消息队列
      • JMS
    • AMQP
    • MQTT
    • Kafka
  • Spring整合消息队列
    • 模拟消息队列的工作流程
    • Spring整合ActiveMQ
    • Spring整合RabbitMQ
      • 直连交换机模式
      • 主题交换机模式
    • Spring整合RocketMQ
    • Spring整合kafka

消息

消息的发送方：生产者
消息的接收方：消费者
同步消息：发送方发送消息到接收方，接收方有所回应后才能够进行下一次的消息发送
异步消息：不需要接收方回应就可以进行下一步的发送

消息队列

什么是消息队列？
消息队列
当此时有很多个用户同时访问服务器，需要服务器进行操作，但此时由于操作太多服务器运转不过来，这时将非常多的操作转换成消息的格式储存器来，所有的子服务器从中获取到消息进行操作分担主服务器的压力，而这个中间存储消息的容器我们一般称为消息队列

  • 企业级应用中广泛使用的三种异步消息传递技术（实现高并发的有效处理）：
  1. JMS
  2. AMQP
  3. MQTT

JMS

（java Message Service）：一个规范，等同于JDBC规范，提供了与消息服务相关的API接口

  • JMS消息模型
  1. peer-2-peer: 点对点模型，消息发送到一个队列中，队列保存信息，队列的消息只能被一个消费者消费，或超时
  2. publish-subscribe:发布订阅模式，消息可以被多个消费者消费，生产者和消费者完全独立，不需要感知对方存在
  • JMS消息种类

TextMessage，MapMessage, BytesMessage,StreamMessage,ObjectMessage,Message(Message只有消息头和属性)

  • 实现JMS规范的MQ

ActiveMQ,Redis,HornetMQ,RabbitMQ,RocketMQ(RocketMQ并未完全遵守JMS规范)

AMQP

AMQP（advanced message queuing protocol）:一种协议（高级队列协议，消息代理规范），规范了网络交换的数据格式，兼容JMS

JMS存在一定的问题，JMS规范对对应的语言进行了规范，但若是我使用不是规范语言进行操作的时候就会出现问题，这时我们推出AMQP，这更像是一种协议，规范消息的格式，就是无论用什么语言什么环境都无所谓，它只人消息的格式

优点：跨平台性，服务器供应商，生产者，消费者可以使用不同的语言来实现

  • AMQP的消息模型

direct exchange,fanout exchange,topic exchange,headers exchange,system exchange

AMQP的消息种类：byte[]

  • 实现AMQP的MQ：

RabbitMQ，StormMQ，RocketMQ

MQTT

(Message Queueing Telemetry Transport)消息队列遥测传输，专为小设备设计，是物联网(IOT)生态系统中主要成分之一

Kafka

kafka，一种高吞吐量的分布式订阅消息系统，提供实时消息功能

Spring整合消息队列

模拟消息队列的工作流程

模拟消息队列的处理过程

import java.util.ArrayList;

@Service
public class Messageservice implements MessageService {
    private ArrayList<String> megList=new ArrayList<String>();

    @Override
    public void sendMessage(String id) {
        System.out.println("将待发送的消息订单纳入到处理队列.id:"+id);
        megList.add(id);
    }

    @Override
    public String doMessage() {
        String remove = megList.remove(0);
        System.out.println("已完成短信业务的发送,id："+remove);
        return remove;
    }
}

模拟将消息导入到消息队列

@Service
public class orderserviceimpl implements orderService {
    @Autowired
    private MessageService messageService;
    @Override
    public void order(String id) {
        //发送消息队列
        messageService.sendMessage(id);
    }
}

Spring整合ActiveMQ

首先下载activeMQ
下载地址：https://activemq.apache.org/components/classic/download/
下载之后进行解压缩

  • 启动服务

打开x64的bin目录下执行activemq.bat命令启动服务
在这里插入图片描述
在这里插入图片描述
启动成功 ，其中给出其web控制台的访问地址：
在这里插入图片描述
进入其管理界面：
在这里插入图片描述
默认用户名&密码：admin

  • SpringBoot进行整合activemq

添加依赖：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>

添加配置：
配置Spirng连接的地址，以及后边消息存入的位置

server:
  port: 80
spring:
  activemq:
  # 说明spring连接的active的端口地址
    broker-url: tcp://localhost:61616

进行消息队列的操作：


@Service
public class Messageservice implements MessageService {
    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;
    @Override
    public void sendMessage(String id) {
        System.out.println("将待发送的消息订单纳入到处理队列.id:"+id);
        jmsMessagingTemplate.convertAndSend(id);
    }
    @Override
    public String doMessage() {
        //将消息队列中的类型转移出来，并在参数中规定转移出来的消息类型
        String s = jmsMessagingTemplate.receiveAndConvert(String.class);
        System.out.println("已完成短信业务的发送,id："+s);
        return s;
    }
}

在发送和获取期间也可以规定名称

 jmsMessagingTemplate.convertAndSend("order.shishi.id",id);
 String s = jmsMessagingTemplate.receiveAndConvert("order.shishi.id",String.class);

上述之中也有一个小问题，就是在并不是每次消费都需要进行访问，而是当消息队列中有消息就开始消费我们可以创建一个Listener


@Component
public class MessageListener {
    @JmsListener(destination = "order.shishi.id")
    public void receive(String id){
        System.out.println("已完成的短信业务:id:"+id);
    }
}

这样就自动监听指定位置下的消息，一有消息就自动开始消费，从服务开始就一直存在
还有一个消息转发的操作：

@Component
public class MessageListener {
    @JmsListener(destination = "order.shishi.id")
    @SendTo("order.bushi.id")
    public void receive(String id){
        System.out.println("已完成的短信业务:id:"+id);
    }
}

注解 @SendTo的作用是将监听到的消息消费之后将返回值返回到对应的消息中去
上述使用的都是点对点的模型，如果要使用发布订阅的模型，可以在配置文件中进行配置：

spring:
  activemq:
    broker-url: tcp://localhost:61616
  jms:
    template:
      default-destination: shishi
    pub-sub-domain: true

Spring整合RabbitMQ

rabbitMQ基于Erlang语言编写，需要安装Erlang
首先需要下载Erlang：
下载地址：https://www.erlang.org/downloads
下载完成之需要重启一下操作系统（重启电脑）
配置环境变量
在这里插入图片描述
添加path：
在这里插入图片描述
安装完成后下载RabbitMQ
下载地址：https://rabbitmq.com/install-windows.html

  • 启动rabbitMQ

在这里插入图片描述
注意：要启动rabbitMQ服务需要命令行进入到管理员身份运行
rabbitMQ的控制台界面（需要手动配置插件）：
在sbin目录下找到：rabbitmq:plugins.bat命令
在这里插入图片描述
执行命令展示其插件列表，通过命令开启插件
在这里插入图片描述
这样就可以访问它的控制台界面，端口号是15672，地址：http://localhost:15672
在这里插入图片描述
输入默认密码：guest

Spring进行整合rabbitMQ首先添加依赖：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

在配置文件中进行rabbits的配置：

spring:
 activemq:
   broker-url: tcp://localhost:61616
 jms:
   template:
     default-destination: shishi
   pub-sub-domain: true
 rabbitmq:
   host: localhost
   port: 5672

直连交换机模式

使用直连模式的交换机进行消息队列的开发：
首先需要在配置类中进行直连交换机与消息队列的绑定


import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class ConfigQM {
   @Bean
   public Queue directQueue(){
       //第一个是消息队列的名称，第一个true表示消息持久化，第二个表示当前的消息队列是否是连接专用（连接一关消息队列就关闭），第三个参数是是否删除（当消费者生产者都不使用就删除）
       return new Queue("direct_queue",true,true,true);
   }
   //我们需要一个交换机去绑定消息队列,此处设置一个交换机
   @Bean
   public DirectExchange directExchange(){
       return new DirectExchange("directexchange");
   }
   @Bean
   public Binding binding(){
       //将消息队列与交换机进行绑定
       return BindingBuilder.bind(directQueue()).to(directExchange()).with("direct");
   }
}

绑定之后通过直连交换机进行消息的存储

@Service
public class amqpservice implements MessageService{

   @Autowired
   private AmqpTemplate amqpTemplate;
   @Override
   public void sendMessage(String id) {
       //使用直连交换机
       amqpTemplate.convertAndSend("directExchange","direct",id);
   }
   @Override
   public String doMessage() {
       return null;
   }
}

然后从消息队列中读取消息写在rabbitMQ监听器下面：

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class RabbitMQListener {
   @RabbitListener(queues = "direct_queue")
   public void reveive(String id){
       System.out.println("已完成短信发送业务 id:"+id);
   }
}

主题交换机模式

主题交换机可以模糊设置交换机绑定的名称来达到分发的目的
例如：

    @Bean
   public Binding binding(){
       //将消息队列与交换机进行绑定
       return BindingBuilder.bind(directQueue()).to(directExchange()).with("topic_*_id");
   }

在消息进入消息队列的时候：

        amqpTemplate.convertAndSend("directExchange","topic_ni_id",id);
        amqpTemplate.convertAndSend("directExchange","topic_bu_id",id);

这个两种消息都可以进入到消息队列中去，而且通过这种方式也可以使消息进入到不同的消息队列中去

  • 绑定案件的规则：

*（星号）：用来表示一个单词，且该单词必须出现
#（井号）：用来表示任意数量
在这里插入图片描述

Spring整合RocketMQ

下载地址：https://rocketmq.apache.org/
默认服务端口：9876
配置环境变量：ROCKETMQ_HOME，PATH，NAMESER_ADDR(建议)：127.0.0.1：9876

  • 命名服务器与broker

在这里插入图片描述
当后期的业务服务器增多时，就需要不停的进行服务器之间的连接，会变得非常繁琐，但是如果我们有一台服务器将所有的业务服务器注册进行，消费者与生产者只需要连接命名服务器即可

  • 首先启动命名服务器
    在这里插入图片描述
    双击文件启动命名服务器
    在这里插入图片描述
    然后双击mqbroker文件启动服务器：
    在这里插入图片描述
    如何测试服务器是否正常启动：
    在bin目录下启动cmd：
    在这里插入图片描述

首先使用第一个命名生成对应的消息：
在这里插入图片描述
再使用第二个命令对生成的消息进行消费：
在这里插入图片描述
进行整合：
首先导入依赖坐标：

        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.2.1</version>
        </dependency>

在配置文件中配置其命名服务器：
rocketmq是与spring在同一层次下

rocketmq:
   name-server: localhost:9876
   producer:
      group: group_rocketmq

进行消息队列的相关操作：

@Service
public class MessageRocketmqimpl implements MessageService {
   @Autowired
   private RocketMQTemplate rocketMQTemplate;
   @Override
   public void sendMessage(String id) {
       rocketMQTemplate.convertAndSend("sdasda",id);
   }
   @Override
   public String doMessage() {
       return null;
   }
}

消费者监听器：

 
@Component
@RocketMQMessageListener(topic = "sdasda",consumerGroup = "group_rocketmq")
public class MessageRocketmqListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String s) {
        System.out.println("id:"+s);
        
    }
}

使用异步方式进行发送：


@Service
public class MessageRocketmqimpl implements MessageService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Override
    public void sendMessage(String id) {
        SendCallback callback=new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("消息发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("消息发送失败");
            }
        }
//        rocketMQTemplate.convertAndSend("sdasda",id);
        rocketMQTemplate.asyncSend("sdasda",id,callback);
    }
  • 同步发送与异步发送的区别：

同步发送和异步发送是两种不同的消息发送方式。在同步发送中，发送线程会等待消息发送完成并收到发送结果后继续执行，而在异步发送中，发送线程不会阻塞，可以立即执行后续逻辑。选择哪种方式取决于业务需求和对消息发送结果的要求。

Spring整合kafka

下载地址：https://kafka.apache.org/downloads
下载之后进行解压缩文件
解压之后首先需要运行：zookeeper-server-start.bat文件
这个文件相当于一个注册中心，需要先进行注册才能够启动kafka服务器，作用相当于RocketMQ中的命名服务器，需要在对应目录下cmd命令携带参数进行启动：
在这里插入图片描述
启动注册服务器后，然后开启kafka服务器：
在这里插入图片描述
在这里插入图片描述
spring进行整合kafka：
导入依赖坐标:

        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

在配置文件中进行配置，配置注册服务器的地址：

  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order

@Service
public class kafka implements MessageService {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    @Override
    public void sendMessage(String id) {
        kafkaTemplate.send("adad",id);
    }

    @Override
    public String doMessage() {
        return null;
    }
}

创建消费者监听器：


@Component
public class kafkaListener {
    @KafkaListener(topics = "adad")
    public void onMessage(ConsumerRecord<String,String> consumerRecord){
        System.out.println("id:"+consumerRecord.value());
    }
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/547097.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

vite - WebAssembly入门

vite - WebAssembly入门

1. 初始化 vite 项目 1.1 安装 nvm&#xff08;可选&#xff09; brew update brew install nvm在 ~/.zshrc 添加 export NVM_DIR~/.nvm source $(brew --prefix nvm)/nvm.sh执行如下命令 source ~/.zshrc1.2 安装 node nvm install nodenvm ls -> …
阅读更多...
1260. 二维网格迁移

1260. 二维网格迁移

1260. 二维网格迁移 原题链接&#xff1a;完成情况&#xff1a;解题思路&#xff1a;参考代码&#xff1a;错误经验吸取 原题链接&#xff1a; 1260. 二维网格迁移 https://leetcode.cn/problems/shift-2d-grid/description/ 完成情况&#xff1a; 解题思路&#xff1a; 这…
阅读更多...
android不同版本（支持＞10）获取当前连接的wifi名称

android不同版本（支持＞10）获取当前连接的wifi名称

1、AndroidManifest.xml 配置权限 <uses-permission android:name"android.permission.ACCESS_COARSE_LOCATION" /> <uses-permission android:name"android.permission.CHANGE_NETWORK_STATE" /> <uses-permission android:name&q…
阅读更多...
磁盘管理和文件系统

磁盘管理和文件系统

一.磁盘基础 1.磁盘结构 &#xff08;1&#xff09;物理结构&#xff1a; 盘片&#xff1a;硬盘有多个盘片&#xff0c;每盘片2面 磁头&#xff1a;每面一个磁头 &#xff08;2&#xff09;硬盘的数据结构 扇区&#xff1a;盘片被分为多个扇形区域&#xff0c;每个扇区存…
阅读更多...
山姆·奥特曼是如何成为亿万富豪的？

山姆·奥特曼是如何成为亿万富豪的？

2017年夏天&#xff0c;Superhuman公司首席执行官拉胡尔沃拉&#xff08;Rahul Vohra&#xff09;开始疯狂向投资者一一发消息&#xff0c;缘由是他的初创公司尝试了谷歌浏览器Chrome的一项即将推出的更新。由于一个看似无害的代码更改&#xff0c;Superhuman的智能电子邮件服务…
阅读更多...
Jackson 2.x 系列【23】注解内省 AnnotationIntrospector

Jackson 2.x 系列【23】注解内省 AnnotationIntrospector

有道无术&#xff0c;术尚可求&#xff0c;有术无道&#xff0c;止于术。 本系列Jackson 版本 2.17.0 源码地址&#xff1a;https://gitee.com/pearl-organization/study-jaskson-demo 文章目录 1. 前言2. AnnotationIntrospector3. JacksonAnnotationIntrospector4. Annotati…
阅读更多...
SQL12 获取每个部门中当前员工薪水最高的相关信息

SQL12 获取每个部门中当前员工薪水最高的相关信息

题目&#xff1a;获取每个部门中当前员工薪水最高的相关信息 注意了&#xff0c;这道题目&#xff0c;分组函数只能查出来&#xff1a;每个部门的最高薪水&#xff0c;group by dept_no &#xff0c;根据部门分组&#xff0c;绝对不能group by dept_no,emp_no&#xff0c;不能…
阅读更多...
【刷题笔记】第二天

【刷题笔记】第二天

一道图论相关的题目 3108. 带权图里旅途的最小代价 结论&#xff1a; 做与运算&#xff0c;结果不会大于当前值&#xff0c;也就是说与运算只能导致结果不变或越来越小&#xff0c;所以要使得边的and值最小&#xff0c;就是把每一个联通块的所有边都and一遍。 方法1&#xf…
阅读更多...
Vue项目实战：基于用户身份的动态路由管理

Vue项目实战：基于用户身份的动态路由管理

&#x1f31f; 前言 欢迎来到我的技术小宇宙&#xff01;&#x1f30c; 这里不仅是我记录技术点滴的后花园&#xff0c;也是我分享学习心得和项目经验的乐园。&#x1f4da; 无论你是技术小白还是资深大牛&#xff0c;这里总有一些内容能触动你的好奇心。&#x1f50d; &#x…
阅读更多...
201403-3 命令行选项

201403-3 命令行选项

100分 #include <bits/stdc.h> using namespace std;int main() {string line;cin >> line;map<char, bool> dct; // true:带参数 false:不带参数for (int i 0; i < line.size(); i){if (line[i] ! :){dct.insert(pair<char, bool>(line[i], fals…
阅读更多...
ubuntu 23.10.1 mysql 安装

ubuntu 23.10.1 mysql 安装

注&#xff1a;请进入root用户模式下操作&#xff0c;若没有&#xff0c;输入命令前加上sudo 1、更新软件包列表 apt update2、安装最新版的Mysql服务器 apt install mysql-server -y如果不加-y 会在安装过程中&#xff0c;系统将提示你设置MySQL的root密码。确保密码足够强…
阅读更多...
职场成长之路：如何规划与实现

职场成长之路：如何规划与实现

在职场中&#xff0c;每个人都希望实现自己的职业目标和成长。然而&#xff0c;职场成长并非一蹴而就&#xff0c;需要有明确的规划和方法。本文将探讨如何在职场中规划与实现成长&#xff0c;帮助您迈向成功之路。 一、明确职业目标 1. 自我认知&#xff1a;了解自己的兴趣、…
阅读更多...
C语言双向链表

C语言双向链表

1. 链表的分类 链表的种类很多&#xff0c;主要由三个要素决定&#xff1a;是否带头&#xff0c;单向还是双向&#xff0c;是否循环。 根据这三个要素的组合&#xff0c;共可得到8&#xff08;2*2*2&#xff09;种链表 而我们常用的链表有两种&#xff1a; 1. 单链表&#xf…
阅读更多...
鸿蒙原生应用元服务-访问控制（权限）开发Stage模型向用户申请授权

鸿蒙原生应用元服务-访问控制（权限）开发Stage模型向用户申请授权

一、向用户申请授权 当应用需要访问用户的隐私信息或使用系统能力时&#xff0c;例如获取位置信息、访问日历、使用相机拍摄照片或录制视频等&#xff0c;应该向用户请求授权。这需要使用 user_grant 类型权限。在此之前&#xff0c;应用需要进行权限校验&#xff0c;以判断当前…
阅读更多...
02_按键控制LED

02_按键控制LED

按键控制LED 按键控制LED 按键控制LED while (1){//按键控制LEDif(HAL_GPIO_ReadPin(GPIOC,GPIO_PIN_5)GPIO_PIN_RESET)//读取PC5引脚状态&#xff0c;即检测按键是否按下{while(HAL_GPIO_ReadPin(GPIOC,GPIO_PIN_5)GPIO_PIN_RESET);//松手检测HAL_GPIO_WritePin(GPIOA,GPIO_PI…
阅读更多...
【JavaWeb】Day45.Mybatis——入门程序

【JavaWeb】Day45.Mybatis——入门程序

什么是MyBatis? MyBatis是一款优秀的持久层框架&#xff0c;用于简化JDBC的开发。 &#xff08;持久层&#xff1a;指的是就是数据访问层(dao)&#xff0c;是用来操作数据库的。&#xff09; &#xff08;框架&#xff1a;是一个半成品软件&#xff0c;是一套可重用的、通用…
阅读更多...
【软考】UML中的图之类图

【软考】UML中的图之类图

目录 1. 说明2. 图示3. 类图使用方式3.1 对系统的词汇建模3.2 对简单的协作建模3.3 对逻辑数据库模式建模 1. 说明 1.类图&#xff08;Class Diagram&#xff09;展现了一组对象、接口、协作和它们之间的关系。2.在面向对象系统的建模中所建立的最常见的图是类图。3.类图给出系…
阅读更多...
Openwrt21.02支持SKW78(MT7621)

Openwrt21.02支持SKW78(MT7621)

1.获取SDK 1.下载Openwrt源码 下载链接&#xff1a; git clone --branch openwrt-21.02 https://gitee.com/cocos_yang/openwrt.git 下载完后&#xff0c;会有一个openwrt目录&#xff0c;进入openwrt目录 cd openwrt 修改feeds.conf.default的内容&#xff0c;如下所示&#x…
阅读更多...
操作系统（1）计算机存储结构

操作系统（1）计算机存储结构

文章目录 一、计算机存储结构1、基本概念2、计算机存储结构的组成部分3、局部性原理4、高速缓存4.1、基本概念4.2、工作原理4.3、层次结构4.4、对系统性能的影响4.5、一致性问题 5、寄存器5.1&#xff0c;种类5.2&#xff0c;特点5.3、作用5.4、寄存器与缓存的差异 前言 计算机…
阅读更多...
MybatisPlus实现数据权限隔离

MybatisPlus实现数据权限隔离

引言 Mybatis Plus对Mybatis做了无侵入的增强&#xff0c;非常的好用&#xff0c;今天就给大家介绍它的其中一个实用功能&#xff1a;数据权限插件。 数据权限插件的应用场景和多租户的动态拦截拼接SQL一样。建议点赞收藏关注&#xff0c;方便以后复习查阅。 依赖 首先导入M…
阅读更多...
最新文章

Copyright @ 2022~2023