【Spring Cloud + RabbitMQ 实现分布式消息总线】—— 每天一点小知识-阿里云开发者社区

【Spring Cloud + RabbitMQ 实现分布式消息总线】—— 每天一点小知识

2023-12-05 252

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 【Spring Cloud + RabbitMQ 实现分布式消息总线】—— 每天一点小知识

🐳使用 Spring Cloud + RabbitMQ 实现分布式消息总线

Spring Cloud 是一个用于构建分布式系统的开发工具包，而 RabbitMQ 是一种功能强大的消息代理。结合使用 Spring Cloud 和 RabbitMQ，我们可以实现一个强大的分布式消息总线。本文将介绍如何使用 Spring Cloud Bus 和 RabbitMQ 实现以下功能：

配置刷新
事件广播
服务监控
微服务间通信

准备工作

💧在开始之前，确保您已经完成以下准备工作：

安装并配置 RabbitMQ。
创建一个 Spring Cloud 微服务项目。

配置刷新

💧配置刷新是一种实现动态配置更新的功能。当配置发生变化时，我们希望能够自动刷新微服务的配置，而不需要重启服务。下面是实现配置刷新的步骤：

在您的 Spring Cloud 微服务项目中，添加以下依赖：

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

配置 RabbitMQ 连接信息，在 application.properties 文件中添加以下配置：

spring.rabbitmq.host=<RabbitMQ 主机名>
spring.rabbitmq.port=<RabbitMQ 端口>
spring.rabbitmq.username=<RabbitMQ 用户名>
spring.rabbitmq.password=<RabbitMQ 密码>

在您的微服务的配置类上添加 @RefreshScope 注解，例如：

@RestController
@RefreshScope
public class MyController {
    // Controller code here
}

配置 RabbitMQ 发送消息的端点。在 application.properties 文件中添加以下配置：

management.endpoints.web.exposure.include=bus-refresh

启动您的微服务应用程序，并进行配置更改。
使用 POST 请求访问 /actuator/bus-refresh 端点，例如：http://localhost:8080/actuator/bus-refresh。这将触发配置刷新并更新微服务的配置。

事件广播

💧事件广播是一种将事件消息广播给所有微服务实例的功能。下面是实现事件广播的步骤：

在您的 Spring Cloud 微服务项目中，添加以下依赖：

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

配置 RabbitMQ 连接信息，在 application.properties 文件中添加以下配置：

spring.rabbitmq.host=<RabbitMQ 主机名>
spring.rabbitmq.port=<RabbitMQ 端口>
spring.rabbitmq.username=<RabbitMQ 用户名>
spring.rabbitmq.password=<RabbitMQ 密码>

创建一个事件消息发布者，用于发送事件消息。例如，创建一个名为 EventPublisher 的类：

import org.springframework.cloud.bus.BusProperties;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
@Component
public class EventPublisher {
    private final ApplicationEventPublisher eventPublisher;
    private final BusProperties busProperties;
    public EventPublisher(ApplicationEventPublisher eventPublisher, BusProperties busProperties) {
        this.eventPublisher = eventPublisher;
        this.busProperties = busProperties;
    }
    public void publishEvent(Object event) {
        String originService = busProperties.getId();
        RemoteApplicationEvent remoteEvent = new RemoteApplicationEvent(event, originService, null);
        eventPublisher.publishEvent(remoteEvent);
    }
}

在需要触发事件广播的地方，使用 EventPublisher 发布事件消息。例如，在一个 REST Controller 中：

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class MyController {
    private final EventPublisher eventPublisher;
    public MyController(EventPublisher eventPublisher) {
        this.eventPublisher = eventPublisher;
    }
    @GetMapping("/trigger-event")
    public String triggerEvent() {
        // 创建事件对象
        MyEvent event = new MyEvent("Hello, world!");
        // 发布事件消息
        eventPublisher.publishEvent(event);
        return "Event triggered";
    }
}

启动多个微服务实例，并访问其中一个实例的 /trigger-event 路径。事件消息将被广播到所有微服务实例，每个实例都能接收到该事件并执行相应的操作。

服务监控

💧服务监控可以实现对微服务实例的监控和告警。下面是实现服务监控的步骤：

在您的 Spring Cloud 微服务项目中，添加以下依赖：

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

配置 RabbitMQ 连接信息，在 application.properties 文件中添加以下配置：

spring.rabbitmq.host=<RabbitMQ 主机名>
spring.rabbitmq.port=<RabbitMQ 端口>
spring.rabbitmq.username=<RabbitMQ 用户名>
spring.rabbitmq.password=<RabbitMQ 密码>

配置 RabbitMQ 发送消息的端点。在 application.properties 文件中添加以下配置：

management.endpoints.web.exposure.include=bus-health, bus-env

启动您的微服务应用程序，并访问 /actuator/bus-health 和 /actuator/bus-env 端点，例如：http://localhost:8080/actuator/bus-health 和 http://localhost:8080/actuator/bus-env。这将提供有关微服务实例的健康状态信息和环境变量信息。
配置监控系统以订阅并处理 RabbitMQ 发送的监控消息，实现实时监控和告警功能。

微服务间通信

💧微服务间通信是通过消息传递实现解耦和异步处理的方式。下面是实现微服务间通信的步骤：

在您的 Spring Cloud 微服务项目中，添加以下依赖：

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

配置 RabbitMQ 连接信息，在 application.properties 文件中添加以下配置：

spring.rabbitmq.host=<RabbitMQ 主机名>
spring.rabbitmq.port=<RabbitMQ 端口>
spring.rabbitmq.username=<RabbitMQ 用户名>
spring.rabbitmq.password=<RabbitMQ 密码>

创建一个消息消费者，用于处理接收到的消息。例如，创建一个名为 MessageConsumer 的类：

import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
@Component
public class MessageConsumer implements ApplicationListener<RemoteApplicationEvent> {
    @Override
    public void onApplicationEvent(RemoteApplicationEvent event) {
        if (event instanceof MyMessageEvent) {
            MyMessageEvent messageEvent = (MyMessageEvent) event;
            // 处理接收到的消息
            handleMessage(messageEvent.getMessage());
        }
    }
    private void handleMessage(String message) {
        // 处理消息的逻辑
        System.out.println("Received message: " + message);
    }
}

创建一个消息发布者，用于发送消息给其他微服务。例如，创建一个名为 MessagePublisher 的类：

import org.springframework.cloud.bus.BusProperties;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
@Component
public class MessagePublisher {
    private final ApplicationEventPublisher eventPublisher;
    private final BusProperties busProperties;
    public MessagePublisher(ApplicationEventPublisher eventPublisher, BusProperties busProperties) {
        this.eventPublisher = eventPublisher;
        this.busProperties = busProperties;
    }
    public void publishMessage(String message) {
        String originService = busProperties.getId();
        MyMessageEvent messageEvent = new MyMessageEvent(message, originService);
        eventPublisher.publishEvent(messageEvent);
    }
}

创建一个自定义的消息事件类，用于封装消息内容。例如，创建一个名为 MyMessageEvent 的类：

import org.springframework.cloud.bus.event.RemoteApplicationEvent;
public class MyMessageEvent extends RemoteApplicationEvent {
    private String message;
    // 构造函数、getter、setter 省略...
    public MyMessageEvent(String message, String originService) {
        super(message, originService);
        this.message = message;
    }
}

在需要发送消息的地方，使用 MessagePublisher 发布消息。例如，在一个 REST Controller 中：

import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class MyController {
    private final MessagePublisher messagePublisher;
    public MyController(MessagePublisher messagePublisher) {
        this.messagePublisher = messagePublisher;
    }
    @PostMapping("/send-message")
    public String sendMessage(@RequestBody String message) {
        // 发送消息给其他微服务
        messagePublisher.publishMessage(message);
        return "Message sent";
    }
}