SpringCloud 集成 RocketMQ 及配置解析

文章目录

  • 前言
  • 一、SpringCloud 集成 RocketMQ
    • 1. pom 依赖
    • 2. yml 配置
    • 3. 操作实体
    • 4. 生产消息
      • 4.1. 自动发送消息
      • 4.2. 手动发送消息
    • 5. 消费消息
  • 二、配置解析
    • 1. spring.cloud.stream.function.definition


前言

  1. 定义
    Spring Cloud Stream 是一个用来为微服务应用构建消息驱动能力的框架。它可以基于 Spring Boot 来创建独立的、可用于生产的 Spring 应用程序。Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现,并引入了发布-订阅、消费组、分区这三个核心概念。简单的说,Spring Cloud Stream本质上就是整合了Spring Boot和Spring Integration,实现了一套轻量级的消息驱动的微服务框架。

  2. 抽象模型
    我们都知道市面上有很多消息中间件,Sping Cloud Stream 为了可以集成各种各样的中间件,它抽象出了 Binder 的概念,每个消息中间件都需要有对应自己的 Binder。这样它就可以根据不同的 Binder 集成不同的中间件。下图的input和output是channel,Binder则是消息中间件和通道之间的桥梁。
    在这里插入图片描述

  3. 绑定器
    通过使用 Spring Cloud Stream,可以有效简化开发人员对消息中间件的使用复杂度,让系统开发人员可以有更多的精力关注于核心业务逻辑的处理。但是目前 Spring Cloud Stream 只支持 RabbitMQ 和 Kafka 的自动化配置。
    Spring Cloud Stream 提供了 Binder (负责与消息中间件进行交互),我们则通过 inputs 或者 outputs 这样的消息通道与 Binder 进行交互。

Binder 绑定器是 Spring cloud Stream 中一个非常重要的概念,实现了应用程序和消息中间件之间的隔离,同时我们也可以通过应用程序实现,消息中间件之间的通信。在我们的项目的可以继承多种绑定器,我们可以根据不同特性的消息使用不同的消息中间件。Spring Cloud Stream 为我们实现了 RabbitMQ 和Kafka 的绑定器。如果你想使用其他的消息中间件需要自己去实现绑定器接口。

一、SpringCloud 集成 RocketMQ

1. pom 依赖

<!-- rocketmq -->
<dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
</dependency>

2. yml 配置

spring:cloud:stream:function:definition: producer1;consumer1 # 方法定义(用于定义发送者或消费者方法)# 配置消息通道通用属性(适用于所有消息中间件)bindings:# 配置channel消息通道consumer1-in-0:destination: consumer_topic # topic消息主题content-type: application/json # 内容格式group: consumer-group # 消费者组producer1-out-0:destination: producer_topic # topic消息主题content-type: application/json # 内容格式rocketmq:binder:name-server: 127.0.0.1:9876 # rocketmq服务地址vipChannelEnabled: true # 是否开启vip通道(兼容老版本使用。多监听一个端口用于接受处理消息,防止端口占用。)# 配置消息通道独特属性(仅适用于rocketmq)bindings:# 配置channel消息通道(生产者:[functionName]-out-[index],消费者:[functionName]-in-[index])producer1-out-0:producer:group: consumer-groupsync: true # 是否开启同步发送consumer1-in-0: consumer:subscription: myTag  # 消费tagdelayLevelWhenNextConsume: -1suspendCurrentQueueTimeMillis: 99999999broadcasting: false # 是否使用广播消费,默认为false使用集群消费

3. 操作实体

package com.demo.model;import lombok.AllArgsConstructor;
import lombok.Data;/*** 消息model*/
@Data
@AllArgsConstructor
public class MsgModel {/*** 消息id*/private String msgId;/*** 消息内容*/private String message;
}

4. 生产消息

4.1. 自动发送消息

通过 MessageBuilder 自动发送消息。

package com.demo;import com.demo.model.MsgModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;import java.util.function.Supplier;/*** 消息生产者类*/
@Configuration
@Slf4j
public class MyProducer {/*** 消息生产者1*/@Beanpublic Supplier<Message<MsgModel>> producer1() {return () -> {MsgModel msgModel = new MsgModel(System.currentTimeMillis(), "测试消息");log.info("producer1发送消息:" + msgModel);return MessageBuilder.withPayload(entity).build();};}
}

这种方式定义 suppelier 会 默认1000ms 发送一次记录。可以修改 spring.cloud.stream.poller.fixedDelay 设置延迟毫秒值。

4.2. 手动发送消息

通过 StreamBridge 手动发送消息。

package com.demo.controller;import com.demo.model.MsgModel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;/*** 消息controller*/
@RestController
@RequiredArgsConstructor
@RequestMapping("/msg")
@Slf4j
public class MsgController {private final StreamBridge streamBridge;/*** 发送消息*/@GetMapping("/send")public void sendMsg() {MsgModel msgModel = new MsgModel(System.currentTimeMillis(), "测试消息");log.info("producer1发送消息:" + msgModel);streamBridge.send("producer1-out-0", MessageBuilder.withPayload(entity).setHeader("MyHearder", "这是一个请求头").build());}
}

5. 消费消息

package com.demo;import com.demo.model.MsgModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;import java.util.function.Consumer;/*** 消息消费者类*/
@Configuration
@Slf4j
public class ReceiveMQ {/*** 消息消费者1*/@Beanpublic Consumer<Message<MsgModel>> consumer1(){return (message)->{MessageHeaders headers = message.getHeaders();MsgModel msgModel = message.getPayload();log.info("consumer1接收消息,消息头:" + headers.get("MyHeader"));log.info("consumer1接收消息,消息内容:" + msgModel);};}
}

二、配置解析

1. spring.cloud.stream.function.definition

进行生产者或消费者方法定义,在 rocketmq 初始时会加载这些方法以创建生产者或消费者列表。

不管是创建 Consumer 还是 Supplier 或者是 Function Stream 都会将其方法名称进行一个 topic 拆封和绑定。假设创建了一个 Consumer< String > myTopic 的方法,Stream 会将其 拆分成 In 和 out 两个通道:

  • 输入通道(消费者): [functionName]-in-[index]
    consumer1-in-0
  • 输出通道(生产者): [functionName]-out-[index]
    producer1-out-0

注意:这里的 functionName 需要和生产者或消费者方法名称以及 spring.cloud.stream.function.definition 下的名称保持一致。

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

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

idea 使用 git

可以看见项目地址&#xff0c; git clone 地址 就可以拉新项目了 命令 git remote -v

宁夏银川市起名专家的老师颜廷利:死神(死亡)并不可怕,可怕的是...

在中国优秀传统文化之中&#xff0c;汉语‘巳’字与‘四’同音&#xff0c;在阿拉伯数字里面&#xff0c;通常用‘4’来表示&#xff1b; 湖南长沙、四川成都、重庆、宁夏银川最靠谱最厉害的起名大师的老师颜廷利教授指出&#xff0c;作为汉语‘九’字&#xff0c;倘若是换一个…

八、e2studio VS STM32CubeIDE之内存使用情况窗口

目录 一、概述/目的 二、STM32CubeIDE Build Analyzer 三、e2studio Memory Usage 八、e2studio VS STM32CubeIDE之内存使用情况窗口 一、概述/目的 1、嵌入开发最大特点之一就是资源受限&#xff0c;关注芯片资源使用详情是优秀工程师的技能之一 2、Keil和IAR都不支持内存…

使用Docker+Jar方式部署微服务工程(前后端分离)看着一篇就够了

本篇教程的使用到的技术有springboot、springcloud、Nacos、Docker、Nginx部署前后端分离访问的微服务。 部署一下Nacos 首先我们需要在服务器中&#xff08;或者本地部署启动一下Nacos&#xff09;&#xff0c;这里我采用服务器的方式进行部署&#xff0c;这里有一点不一样的…

【XR806开发板试用】SPI驱动数码管显示

准备工作 安装repo 创建repo安装目录。 mkdir ~/bin下载repo wget https://storage.googleapis.com/git-repo-downloads/repo -P ~/bin/改变执行权限 chmod ax ~/bin/repo设置环境变量&#xff0c;在~/.bashrc文件的最后输入 export PATH~/bin:$PATH和export REPO_URLhttps://…

redis深入理解之实战

1、SpringBoot整合redis 1.1 导入相关依赖 <dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId> </dependency> <dependency><groupId>org.springframework.boot</groupId><artifactId&g…

JavaEE之线程(4)——线程安全、线程安全的原因,synchronized关键字

前言 在本栏的前面的内容中&#xff0c;我们介绍了线程的创建、Thread 类及常见方法、线程的状态&#xff0c;今天我们来介绍一下关于线程的另一个重点知识——线程安全。 一、线程安全 基本概念&#xff1a; 线程安全的确切定义是复杂的&#xff0c;但我们可以这样认为&…

vue3 antd-vue 超简单方式实现a-table跨页勾选

效果如下&#xff1a; 一、第一种方法 定义一个变量 selectKeysBack 用于维护所有勾选的数据value, 然后用 onSelect 拿到当前操作的数据&#xff0c;以及时候勾选。 这个方法便于拿到后端的数据之后用于回显 1、template <a-tableclass"custom-table-data":da…

类与对象(二)

封装 封装作为面向对象三大特性&#xff08;封装&#xff0c;继承&#xff0c;多态&#xff09;之一&#xff0c;那如何实现封装性的呢&#xff1f;就又得拿出上面的访问修饰限定符的图 public: 就是在任何地方都可以访问 protected: 涉及子类在介绍继承时详细介绍 default: …

STK12 RPO模块学习 (1)

一、背景介绍 在STK12中&#xff0c;在Astrogator的模块上开发了新的模块&#xff08;Rendezvous and proximity operations)。轨道交会接近通常来说是一个很复杂的过程。RPO实现需要对轨道动力学有一个清晰的理解&#xff0c;并且对于Astrogator模块具备很强的背景和经验&…

【教学类-55-02】20240512图层顺序挑战(四格长条纸加黑色边框、4*4、7张 、43200张去掉非7色有23040张,去掉重复样式有几种?)

作品展示 背景需求&#xff1a; 之前的代码吗存在几个问题&#xff0c;最大的问题是不能生成“”长条黑边框”” 【教学类-55-01】20240511图层顺序挑战&#xff08;四格长条纸&#xff09;&#xff08;4*4&#xff09;和“手工纸自制参考图”-CSDN博客文章浏览阅读485次&…

第十篇:数字堡垒:操作系统安全深度解析与实战指南

数字堡垒&#xff1a;操作系统安全深度解析与实战指南 1 *引言 1.1 数字世界的守护者 在遥远的比特海中&#xff0c;有一座名为“操作系统”的数字堡垒&#xff0c;它守护着我们的数据宝藏&#xff0c;确保每一次计算的航行都能安全抵达彼岸。然而&#xff0c;这片海域并非风…

记录:robot_localization传感器数据融合学习

一、参考资料 官方&#xff1a; http://wiki.ros.org/robot_localizationhttp://docs.ros.org/en/noetic/api/robot_localization/html/index.html2015 ROSCon 演讲官方网址&#xff08;youyube上也有这个视频&#xff09; 实践教程 https://kapernikov.com/the-ros-robot_…

深入理解Python的类,实例和type函数

问题起源&#xff1a; class t():pass s1 t() s2 type("Student2",(),{}) isinstance(s1, type), isinstance(s2, type)为什么第一个是false&#xff0c;第二个是true呢 根因定位&#xff1a; 在Python中&#xff0c;一切皆对象&#xff0c;类是对象&#xff0c…

Linux字符设备驱动设计

Linux字符设备驱动设计 概述 驱动的定义与功能 计算机系统中存在着大量的设备&#xff0c; 操作系统要求能够控制和管理这些硬件&#xff0c; 而驱动就是帮助操作系统完成这个任务。 驱动相当于硬件的接口&#xff0c; 它直接操作、 控制着我们的硬件&#xff0c; 操作系统通…

升级Microsoft 365后,SAP GUI中无法打开Excel的解决方案

最近&#xff0c;我们遇到了一个棘手的问题&#xff0c;一位客户在升级到Microsoft 365后&#xff0c;无法在SAP GUI中打开Excel。这个问题不仅影响了工作效率&#xff0c;也给用户的日常操作带来了不便。在本文中&#xff0c;我们将探讨问题的成因&#xff0c;并提供一种解决方…

gin自定义验证器+中文翻译

gin自定义验证器中文翻译 1、说明2、global.go3、validator.go4、eg&#xff1a;main.go5、调用接口测试 1、说明 gin官网自定义验证器给的例子相对比较简单&#xff0c;主要是语法级别&#xff0c;便于入门学习&#xff0c;并且没有给出翻译相关的处理&#xff0c;因此在这里记…

PaddleOCR_PP-Structure

静态IP设置 # 修改网卡配置文件 vim /etc/sysconfig/network-scripts/ifcfg-ens33# 修改文件内容 TYPEEthernet PROXY_METHODnone BROWSER_ONLYno BOOTPROTOstatic IPADDR192.168.15.132 NETMASK255.255.255.0 GATEWAY192.168.15.2 DEFROUTEyes IPV4_FAILURE_FATALno IPV6INIT…

YOLOv8小白中的小白安装环境教程!没一个字废话,看一遍不踩坑!

文章目录 去哪里下代码?怎么下代码?怎么装环境?命令行界面(CLI)指令和Python脚本区别?附录1 conda常用指令附录2 git常用指令附录3 项目代码文件作用去哪里下代码? 下载代码请大家直接去 YOLOv8的官方仓库下载,名字叫 ultralytics,有些镜像网站和个人发的等来历不明的代…

CMakeLists.txt语法规则:foreach 循环基本用法

一. 简介 cmake 中除了 if 条件判断之外&#xff0c;还支持循环语句&#xff0c;包括 foreach()循环、while()循环。 本文学习 CMakeLists.txt语法中的循环语句。 CMakeLists.txt语法中 有两种 循环实现方式&#xff1a;foreach循环与 while循环。 二. CMakeLists.txt语法规则…