使用消息中间件实现系统间的异步通信和解耦

2024-02-18 06:52
文章标签 实现 系统 使用 通信 异步 消息中间件 和解

本文主要是介绍使用消息中间件实现系统间的异步通信和解耦，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

​​​​​​​目录

引言

一. 选择合适的消息中间件

二. 定义消息格式和通信协议

1. 定义消息格式

消息头

消息体

2. 定义通信协议

发送消息

接收消息

消息处理

3. 示例代码

定义消息格式

发送消息

接收消息

三、发布-订阅模式

1. 定义发布-订阅模式

2. 示例代码

发布消息

订阅消息

3. 运行示例

4. 异步处理消息

5. 解耦系统

6. 实现步骤

7. 实例场景

实例场景：电商系统订单处理

场景描述

实现步骤

示例代码

订单服务发送消息

库存服务接收消息

物流服务接收消息

引言

在现代分布式系统中，异步通信和解耦是非常重要的设计原则。通过使用消息中间件，可以实现系统间的异步通信和解耦，提高系统的可扩展性和可靠性。本文将介绍如何使用消息中间件来实现系统间的异步通信和解耦，并通过一个实际场景来演示。

一. 选择合适的消息中间件

选择合适的消息中间件需要考虑多个因素，包括项目需求、性能要求、可靠性、社区支持等。常见的消息中间件包括 RabbitMQ、Kafka、ActiveMQ、Redis 等，下面针对不同的需求给出一些选择建议：

  1. 消息传递模式：

    • 点对点：适合使用 RabbitMQ、ActiveMQ 等传统消息中间件。
    • 发布-订阅：适合使用 RabbitMQ、Kafka 等支持广播消息的中间件。

  2. 可靠性：

    • 如果对消息的可靠性要求较高，需要确保消息不会丢失，可以考虑使用 RabbitMQ、Kafka 等提供消息持久化和高可靠性的中间件。

  3. 性能：

    • 如果需要处理大量的消息并且需要低延迟，可以考虑使用 Kafka，它是一个高吞吐量的消息中间件，适合大数据场景。
    • 如果对延迟要求较低，可以选择 RabbitMQ、ActiveMQ 等传统消息中间件。

  4. 社区支持和生态系统：

    • 考虑选择一个有活跃社区支持和完善生态系统的消息中间件，这样可以更容易地解决问题和扩展功能。

  5. 技术栈兼容性：

    • 考虑选择一个与你的技术栈兼容的消息中间件，避免出现集成上的问题。

综合考虑以上因素，可以选择最适合项目需求的消息中间件。

二. 定义消息格式和通信协议

定义消息格式和通信协议是使用消息中间件的关键步骤之一，它涉及到消息的结构、内容和交互方式。下面以 RabbitMQ 为例，演示如何定义消息格式和通信协议。

1. 定义消息格式

在 RabbitMQ 中，消息通常由两部分组成：消息头和消息体。消息头包含一些元数据信息，如消息的类型、路由键等；消息体包含实际的业务数据。

消息头
  • Content-Type：消息体的类型，如 application/json、text/plain 等。
  • DeliveryMode：消息持久性标志，标识消息是否需要持久化存储，可选值为 1（持久化）和 2（非持久化）。
  • CorrelationId：消息关联标识，用于关联一组相关消息。
  • 其他自定义的消息头字段，根据业务需求定义。
消息体
  • 消息体可以是任意格式的数据，如 JSON、XML、文本等，根据业务需求定义。

2. 定义通信协议

通信协议定义了消息的交互方式，包括消息的发送、接收和处理流程。通信协议可以包括以下几个方面：

发送消息
  • 客户端向消息队列发送消息，包括指定交换机（Exchange）、路由键（Routing Key）和消息体。
接收消息
  • 服务端从消息队列接收消息，根据消息的交换机和路由键接收对应的消息。
消息处理
  • 客户端接收到消息后，根据消息的内容执行相应的业务逻辑。

3. 示例代码

定义消息格式
public class Message {private String content;private String contentType;private int deliveryMode;private String correlationId;// 省略getter和setter方法
}
发送消息
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class SendMessage {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);Message message = new Message();message.setContent("Hello, RabbitMQ!");message.setContentType("text/plain");message.setDeliveryMode(1); // 持久化message.setCorrelationId("123456");String messageJson = toJson(message);channel.basicPublish("", QUEUE_NAME, null, messageJson.getBytes());System.out.println(" [x] Sent '" + messageJson + "'");}}private static String toJson(Message message) {// 将 message 对象转换成 JSON 格式的字符串return "{ \"content\": \"" + message.getContent() + "\", \"contentType\": \"" + message.getContentType() + "\", \"deliveryMode\": " + message.getDeliveryMode() + ", \"correlationId\": \"" + message.getCorrelationId() + "\" }";}
}
接收消息
import com.rabbitmq.client.*;public class ReceiveMessage {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);System.out.println(" [*] Waiting for messages. To exit press Ctrl+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String messageJson = new String(delivery.getBody(), "UTF-8");Message message = fromJson(messageJson, Message.class);System.out.println(" [x] Received '" + messageJson + "'");};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});}}private static <T> T fromJson(String json, Class<T> clazz) {// 将 JSON 格式的字符串转换成指定类型的对象// 这里可以使用 JSON 框架（如 Jackson、Gson）来实现return null;}
}

通过以上步骤，可以定义消息格式和通信协议，并使用 RabbitMQ 实现消息的发送和接收。

三、发布-订阅模式

发布-订阅模式是一种常见的消息传递模式，用于实现消息的广播和订阅。在发布-订阅模式中，消息发布者将消息发布到一个主题（Topic），而消息订阅者可以订阅感兴趣的主题，从而接收到相关消息。下面以 RabbitMQ 为例，演示如何使用发布-订阅模式。

1. 定义发布-订阅模式

在发布-订阅模式中，有一个交换机（Exchange）用来接收发布者发布的消息，并根据订阅者的绑定关系将消息路由到对应的队列。订阅者可以创建自己的队列，并将队列绑定到交换机上，从而接收到发布者发布的消息。

2. 示例代码

发布消息
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Publisher {private final static String EXCHANGE_NAME = "logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {channel.exchangeDeclare(EXCHANGE_NAME, "fanout");String message = "Hello, subscribers!";channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}}
}
订阅消息
import com.rabbitmq.client.*;public class Subscriber {private final static String EXCHANGE_NAME = "logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {channel.exchangeDeclare(EXCHANGE_NAME, "fanout");String queueName = channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, "");System.out.println(" [*] Waiting for messages. To exit press Ctrl+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});}}
}

3. 运行示例

  1. 先运行订阅者 Subscriber，它会创建一个队列并绑定到交换机上，开始监听消息。
  2. 然后运行发布者 Publisher，它会向交换机发布一条消息。
  3. 订阅者会接收到发布者发布的消息，并输出到控制台。

通过以上步骤，可以实现基于 RabbitMQ 的发布-订阅模式。

4. 异步处理消息

通过消息中间件实现异步处理消息，即发送消息后不需要立即等待结果，而是继续执行其他任务。这样可以提高系统的响应速度和吞吐量。

5. 解耦系统

通过消息中间件，系统之间的通信变成了基于消息的方式，系统不再直接依赖于对方的接口和实现细节，从而实现了系统之间的解耦。

6. 实现步骤

  • 定义消息格式和通信协议：确定消息的格式和通信协议，包括消息的内容结构、消息的生命周期等。
  • 配置消息中间件：在系统中配置和启动消息中间件，确保消息中间件正常运行。
  • 消息的发布和订阅：编写代码实现消息的发布和订阅逻辑，将消息发布到指定的主题，并订阅感兴趣的主题。
  • 处理接收到的消息：编写代码处理接收到的消息，根据消息的内容执行相应的业务逻辑。
  • 测试和验证：对系统进行测试和验证，确保消息的发布、订阅和处理功能正常运行。

7. 实例场景

实例场景：电商系统订单处理

场景描述

假设有一个电商系统，包含订单服务、库存服务和物流服务。当用户下单时，订单服务需要通知库存服务减少库存，通知物流服务发货。为了提高系统的可扩展性和可靠性，我们可以使用消息中间件来实现订单处理的异步通信和解耦。

实现步骤

  1. 定义消息格式和通信协议：定义订单消息的格式，包括订单号、商品信息等，并确定消息的交换机和队列名称。

  2. 配置消息中间件：在消息中间件中配置交换机和队列，并确保消息的持久化。

  3. 订单服务发送消息：订单服务在用户下单后，将订单消息发送到消息队列中。

  4. 库存服务订阅消息：库存服务订阅订单消息队列，接收并处理订单消息，减少库存。

  5. 物流服务订阅消息：物流服务也订阅订单消息队列，接收并处理订单消息，进行发货。

示例代码

订单服务发送消息
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class OrderService {private static final String EXCHANGE_NAME = "orders";private static final String QUEUE_NAME = "order_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {channel.exchangeDeclare(EXCHANGE_NAME, "fanout");channel.queueDeclare(QUEUE_NAME, false, false, false, null);String message = "New order placed";channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}}
}
库存服务接收消息
import com.rabbitmq.client.*;public class InventoryService {private static final String EXCHANGE_NAME = "orders";private static final String QUEUE_NAME = "order_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {channel.exchangeDeclare(EXCHANGE_NAME, "fanout");channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");System.out.println(" [*] Waiting for orders. To exit press Ctrl+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");// 处理订单消息，减少库存};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });}}
}
物流服务接收消息
import com.rabbitmq.client.*;public class LogisticsService {private static final String EXCHANGE_NAME = "orders";private static final String QUEUE_NAME = "order_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {channel.exchangeDeclare(EXCHANGE_NAME, "fanout");channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");System.out.println(" [*] Waiting for orders. To exit press Ctrl+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");// 处理订单消息，发货};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });}}
}

通过以上步骤的简单演示，订单服务可以异步发送订单消息，库存服务和物流服务可以订阅订单消息并处理，实现了订单处理的异步通信和解耦。

通过以上步骤，可以使用消息中间件实现系统间的异步通信和解耦，提高系统的可扩展性和可维护性。

这篇关于使用消息中间件实现系统间的异步通信和解耦的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/720388。

相关文章

使用Python实现IP地址和端口状态检测与监控

使用Python实现IP地址和端口状态检测与监控

《使用Python实现IP地址和端口状态检测与监控》在网络运维和服务器管理中,IP地址和端口的可用性监控是保障业务连续性的基础需求,本文将带你用Python从零打造一个高可用IP监控系统,感兴趣的小伙... 目录概述：为什么需要IP监控系统使用步骤说明1. 环境准备2. 系统部署3. 核心功能配置系统效果展
阅读更多...
Python实现微信自动锁定工具

Python实现微信自动锁定工具

《Python实现微信自动锁定工具》在数字化办公时代,微信已成为职场沟通的重要工具,但临时离开时忘记锁屏可能导致敏感信息泄露,下面我们就来看看如何使用Python打造一个微信自动锁定工具吧... 目录引言：当微信隐私遇到自动化守护效果展示核心功能全景图技术亮点深度解析1. 无操作检测引擎2. 微信路径智能获
阅读更多...
使用Java将各种数据写入Excel表格的操作示例

使用Java将各种数据写入Excel表格的操作示例

《使用Java将各种数据写入Excel表格的操作示例》在数据处理与管理领域,Excel凭借其强大的功能和广泛的应用,成为了数据存储与展示的重要工具,在Java开发过程中,常常需要将不同类型的数据,本文... 目录前言安装免费Java库1. 写入文本、或数值到 Excel单元格2. 写入数组到 Excel表格
阅读更多...
redis中使用lua脚本的原理与基本使用详解

redis中使用lua脚本的原理与基本使用详解

《redis中使用lua脚本的原理与基本使用详解》在Redis中使用Lua脚本可以实现原子性操作、减少网络开销以及提高执行效率,下面小编就来和大家详细介绍一下在redis中使用lua脚本的原理... 目录Redis 执行 Lua 脚本的原理基本使用方法使用EVAL命令执行 Lua 脚本使用EVALSHA命令
阅读更多...
Python中pywin32 常用窗口操作的实现

Python中pywin32 常用窗口操作的实现

《Python中pywin32常用窗口操作的实现》本文主要介绍了Python中pywin32常用窗口操作的实现,pywin32主要的作用是供Python开发者快速调用WindowsAPI的一个... 目录获取窗口句柄获取最前端窗口句柄获取指定坐标处的窗口根据窗口的完整标题匹配获取句柄根据窗口的类别匹配获取句
阅读更多...
Java 中的 @SneakyThrows 注解使用方法(简化异常处理的利与弊)

Java 中的 @SneakyThrows 注解使用方法(简化异常处理的利与弊)

《Java中的@SneakyThrows注解使用方法(简化异常处理的利与弊)》为了简化异常处理,Lombok提供了一个强大的注解@SneakyThrows,本文将详细介绍@SneakyThro... 目录1. @SneakyThrows 简介 1.1 什么是 Lombok？2. @SneakyThrows
阅读更多...
在 Spring Boot 中实现异常处理最佳实践

在 Spring Boot 中实现异常处理最佳实践

《在SpringBoot中实现异常处理最佳实践》本文介绍如何在SpringBoot中实现异常处理,涵盖核心概念、实现方法、与先前查询的集成、性能分析、常见问题和最佳实践,感兴趣的朋友一起看看吧... 目录一、Spring Boot 异常处理的背景与核心概念1.1 为什么需要异常处理？1.2 Spring B
阅读更多...
Python位移操作和位运算的实现示例

Python位移操作和位运算的实现示例

《Python位移操作和位运算的实现示例》本文主要介绍了Python位移操作和位运算的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录1. 位移操作1.1 左移操作 (<<)1.2 右移操作 (>>)注意事项：2. 位运算2.1
阅读更多...
如何在 Spring Boot 中实现 FreeMarker 模板

如何在 Spring Boot 中实现 FreeMarker 模板

《如何在SpringBoot中实现FreeMarker模板》FreeMarker是一种功能强大、轻量级的模板引擎,用于在Java应用中生成动态文本输出（如HTML、XML、邮件内容等）,本文... 目录什么是 FreeMarker 模板？在 Spring Boot 中实现 FreeMarker 模板1. 环
阅读更多...
Qt实现网络数据解析的方法总结

Qt实现网络数据解析的方法总结

《Qt实现网络数据解析的方法总结》在Qt中解析网络数据通常涉及接收原始字节流,并将其转换为有意义的应用层数据,这篇文章为大家介绍了详细步骤和示例,感兴趣的小伙伴可以了解下... 目录1. 网络数据接收2. 缓冲区管理（处理粘包/拆包）3. 常见数据格式解析3.1 jsON解析3.2 XML解析3.3 自定义
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2