030-从零搭建微服务-消息队列（二）

写在最前

如果这个项目让你有所收获，记得 Star 关注哦，这对我是非常不错的鼓励与支持。

源码地址（后端）：mingyue: 🎉 基于 Spring Boot、Spring Cloud & Alibaba 的分布式微服务架构基础服务中心

源码地址（前端）：mingyue-ui: 🎉 基于 Vue3 + TS + Vite + Element plus 等技术，适配 MingYue 后台微服务

文档地址：Wiki - Gitee.com

mingyue-common-mq

添加依赖

根据需要在 mingyue-common-mq 模块中添加所需的 MQ 中间件，例如：RocketMQ、Kafka。

<dependencies><!-- RocketMQ --><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-stream-rocketmq</artifactId></dependency>
<!-- Kafka --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-kafka</artifactId></dependency>
</dependencies>

集成 RocketMQ

引入依赖

<!-- MQ工具 -->
<dependency><groupId>com.csp.mingyue</groupId><artifactId>mingyue-common-mq</artifactId>
</dependency>

Nacos 配置

spring:cloud:stream:function:# 重点配置 与 binding 名与消费者对应definition: rocketmqDemorocketmq:binder:# rocketmq 地址name-server: 192.168.21.32:9876bindings:rocketmqDemo-out-0:producer:# 必须得写group: defaultbindings:rocketmqDemo-out-0:content-type: application/jsondestination: stream-rocketmq-demo-topicgroup: demo-groupbinder: rocketmqrocketmqDemo-in-0:content-type: application/jsondestination: stream-rocketmq-demo-topicgroup: demo-groupbinder: rocketmq

RocketMQ 生产者

@Component
public class RocketMqProducer {
@Resourceprivate StreamBridge streamBridge;
public void rocketMqDemoMsg(String msg) {// 构建消息对象MqMessageDto messageDto = new MqMessageDto().setMsgId(IdUtil.fastSimpleUUID()).setMsgText(msg);
streamBridge.send("rocketmqDemo-out-0", MessageBuilder.withPayload(messageDto).build());}

}

RocketMQ 消费者

@Slf4j
@Component
public class RocketMqConsumer {
@BeanConsumer<MqMessageDto> rocketmqDemo() {log.info("Rocket MQ 初始化订阅");return msg -> {log.info("通过 Rocket MQ 消费到消息 => {}", msg.toString());};}

}

推送消息到 RocketMQ

@GetMapping("/sendRocketMq")
@Operation(summary = "发送消息到RocketMQ", parameters = { @Parameter(name = "msg", description = "推送的消息体", required = true) })
public R<Void> sendRocketMq(String msg) {rocketMqProducer.rocketMqDemoMsg(msg);return R.ok();
}

集成 Kafka

引入依赖

<!-- MQ工具 -->
<dependency><groupId>com.csp.mingyue</groupId><artifactId>mingyue-common-mq</artifactId>
</dependency>

Nacos 配置

spring:cloud:stream:function:# 重点配置 与 binding 名与消费者对应definition: kafkaDemokafka:binder:brokers: 192.168.21.32:9092bindings:kafkaDemo-out-0:destination: stream-kafka-demo-topiccontentType: application/jsongroup: demo-groupbinder: kafkakafkaDemo-in-0:destination: stream-kafka-demo-topiccontentType: application/jsongroup: demo-groupbinder: kafka

Kafka 生产者

@Component
public class KafkaProducer {
@Resourceprivate StreamBridge streamBridge;
public void kafkaDemoMsg(String msg) {// 构建消息对象MqMessageDto messageDto = new MqMessageDto().setMsgId(IdUtil.fastSimpleUUID()).setMsgText(msg);
streamBridge.send("kafkaDemo-out-0", MessageBuilder.withPayload(messageDto).build());}

}

Kafka 消费者

@Slf4j
@Component
public class KafkaConsumer {
@BeanConsumer<MqMessageDto> kafkaDemo() {log.info("Kafka 初始化订阅");return msg -> {log.info("通过 Kafka 消费到消息 => {}", msg.toString());};}

}

推送消息到 Kafka

@GetMapping("/sendKafka")
@Operation(summary = "发送消息到Kafka", parameters = { @Parameter(name = "msg", description = "推送的消息体", required = true) })
public R<Void> sendKafka(String msg) {kafkaProducer.kafkaDemoMsg(msg);return R.ok();
}

拓展 RabbitMQ

mingyue-common-mq 添加依赖

<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

Nacos 配置

--- # rabbitmq 配置
spring:rabbitmq:host: rabbitmqIpport: 5672username: rootpassword: rootcloud:stream:function:# 重点配置 与 binding 名与消费者对应definition: rabbitmqDemorabbit:bindings:rabbitmqDemo-in-0:consumer:delayedExchange: truerabbitmqDemo-out-0:producer:delayedExchange: truebindings:rabbitmqDemo-in-0:destination: delay.exchange.democontent-type: application/jsongroup: delay-groupbinder: rabbitrabbitmqDemo-out-0:destination: delay.exchange.democontent-type: application/jsongroup: delay-groupbinder: rabbit