通过Spring Cloud Stream 进行 RoctetMQ 消息操作

发布网友 发布时间：2024-10-22 19:47

我来回答

共1个回答

热心网友 时间：2024-10-22 20:04

Spring Cloud Stream 简介

Spring Cloud Stream 是一个框架，用于构建与共享消息系统连接的高度可扩展的事件驱动微服务。 该框架提供了一个灵活的编程模型，该模型基于已经建立和熟悉的 Spring 习惯用法和最佳实践，包括对持久 pub/sub 语义、消费者组和有状态分区的支持。

Spring Cloud Stream支持哪些消息中间件

Spring Cloud Stream 支持的消息中间件如下图所示：

快速使用设置依赖（pom）

对于依赖，我们需要注意 spring-boot``spring-cloud``spring-cloud-alibaba版本对应关系即可。

<properties><maven.compiler.target>1.8</maven.compiler.target><maven.compiler.source>1.8</maven.compiler.source><spring.boot.version>2.2.4.RELEASE</spring.boot.version><spring.cloud.version>Hoxton.SR1</spring.cloud.version><spring.cloud.alibaba.version>2.2.0.RELEASE</spring.cloud.alibaba.version></properties><dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>${spring.boot.version}</version><type>pom</type><scope>import</scope></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-dependencies</artifactId><version>${spring.cloud.version}</version><type>pom</type><scope>import</scope></dependency><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-alibaba-dependencies</artifactId><version>${spring.cloud.alibaba.version}</version><type>pom</type><scope>import</scope></dependency></dependencies></dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-stream-rocketmq</artifactId></dependency></dependencies>设置配置（yml）

配置 ?rocketmq需要的一些基础配置信息。

spring:cloud:stream:bindings:input:destination: TestTopicgroup: TestGroupoutput:destination: TestTopicrocketmq:binder:name-server: 127.0.0.1:9876server:port: 18080设置启动类@EnableBinding({Source.class, Sink.class})@SpringBootApplicationpublic class RocketMQApplication {public static void main(String[] args) {SpringApplication.run(RocketMQApplication.class, args);}}消息生产者

主要是实现发消息

@Componentpublic class RocketMQProducer {@Resourceprivate Source source;public void sendMessage(String msg) {Map<String, Object> headers = new HashMap<>();headers.put(MessageConst.PROPERTY_TAGS, "testTag");MessageHeaders messageHeaders = new MessageHeaders(headers);Message<String> message = MessageBuilder.createMessage(msg,messageHeaders);this.source.output().send(message);}}消息消费者

主要是用来接受消息， 并且打印消息的内容。

@Componentpublic class TestTopicConsumer {@StreamListener(Sink.INPUT)public void onMessage(String messsage) {System.out.println("received message:" + messsage + " from binding:" + Sink.INPUT);}}测试

我们编写一个 TestController类，对外提供一个发消息的接口，去调用发消息，实现发消息。

@RestController@RequestMapping("/MQTest")public class TestController {@Resourceprivate RocketMQProducer producer;@RequestMapping("/sendMessage")public String sendMessage(String message) {producer.sendMessage(message);return "send message success";}}

测试发消息，执行命令 curl http://127.0.0.1:18080/MQTest/sendMessage\?message\=hahha , 返回结果如下：服务端控制台输出如下：

总结

本文简答的对 Spring Cloud Stream ?整合 RockectMQ 做了实践，但是在我们在实际开发过程中，这样来做是远远不够的，注意这个只是案例，后面我也会分享项目中正真使用的方案。

原文：https://juejin.cn/post/71042821130750665