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

消息传递模式：
- 点对点：适合使用 RabbitMQ、ActiveMQ 等传统消息中间件。
- 发布-订阅：适合使用 RabbitMQ、Kafka 等支持广播消息的中间件。
可靠性：
- 如果对消息的可靠性要求较高，需要确保消息不会丢失，可以考虑使用 RabbitMQ、Kafka 等提供消息持久化和高可靠性的中间件。
性能：
- 如果需要处理大量的消息并且需要低延迟，可以考虑使用 Kafka，它是一个高吞吐量的消息中间件，适合大数据场景。
- 如果对延迟要求较低，可以选择 RabbitMQ、ActiveMQ 等传统消息中间件。
社区支持和生态系统：
- 考虑选择一个有活跃社区支持和完善生态系统的消息中间件，这样可以更容易地解决问题和扩展功能。
技术栈兼容性：
- 考虑选择一个与你的技术栈兼容的消息中间件，避免出现集成上的问题。

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

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

定义消息格式和通信协议是使用消息中间件的关键步骤之一，它涉及到消息的结构、内容和交互方式。下面以 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. 运行示例

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

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

4. 异步处理消息

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

5. 解耦系统

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

6. 实现步骤

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

7. 实例场景

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

场景描述

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

实现步骤

定义消息格式和通信协议：定义订单消息的格式，包括订单号、商品信息等，并确定消息的交换机和队列名称。
配置消息中间件：在消息中间件中配置交换机和队列，并确保消息的持久化。
订单服务发送消息：订单服务在用户下单后，将订单消息发送到消息队列中。
库存服务订阅消息：库存服务订阅订单消息队列，接收并处理订单消息，减少库存。
物流服务订阅消息：物流服务也订阅订单消息队列，接收并处理订单消息，进行发货。

示例代码

订单服务发送消息

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 -> { });
        }
    }
}

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

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

原文地址：https://blog.csdn.net/2201_75809246/article/details/136132655

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：揭秘铷原子钟：北斗卫星系统的“心脏”
下一篇：win32汇编获取系统信息

[CUDA] cuda kernel开发记录
包括kernel的一些使用注意事项， launch_bound, __device__, debug排查技巧
阅读更多2024-11-16
【python】掌握 Flask：轻量级 Web 开发框架解析
路由是 Web 开发的基础，负责管理 URL 到视图函数的映射。在 Flask 中，路由定义非常简单，只需使用装饰器即可。这个代码段定义了一个路由，访问该路径时会返回 “Hello, Flask!通过
阅读更多2024-11-16
AI图片分析接口LiteAIServer摄像机实时接入分析平台未戴安全帽检测算法
随着人工智能技术的飞速发展，摄像机实时接入分析平台LiteAIServer工地未佩戴安全帽检测算法应运而生，为工地安全管理带来了革命性的变革。
阅读更多2024-11-16
2024新版pycharm如何切换anaconda虚拟环境
回归正题，导入项目后点击文件=>设置，找到解释器。不得不说这界面改的真不错，看着很舒服。另外在终端用指令切换也是可以的。添加解释器=>添加本地解释器。
阅读更多2024-11-16
计算机提示mfc140u.dll丢失的五种解决方法，了解mfc140u.dll错误的几种修复方法
当你尝试打开某些程序时，突然出现错误提示，告知你系统缺少 mfc140u.dll 文件，这可能让你感到困惑和无助。mfc140u.dll 是 Microsoft Foundation Class (M
阅读更多2024-11-16
k8s 中传递参数给docker容器
在 Kubernetes 中，可以通过多种方式将参数传递给 Dockerfile 或其运行的容器，常见的方式包括使用环境变量、命令行参数、配置文件等。
阅读更多2024-11-16
设计模式之工厂模式，但是宝可梦
作为一个细分了三个种类的设计模式，到底该如何取舍？比起直接new一个对象，使用对应模式的好处到底在哪？简单工厂模式：根据传入的参数决定产出的对象，可以隐藏一些创建的细节适用于需要根据条件创建不同对象的
阅读更多2024-11-16
【深度学习】wsl-ubuntu深度学习基本配置
这里注意一点，你换了源之后就最好不要开代理了，要不然搞不好下载失败，pip和conda都是。
阅读更多2024-11-16
nodejs和npm在gitbash中提示Not Found情况的解决办法
很多小伙伴学习了node以后，在cmd命令行中可以正常的获取node版本和npm版本，但是我们经常使用gitbash来管理git，这时候下载完gitbash后，在gitbash中输入node -v和n
阅读更多2024-11-16
判断子序列
给定一个长度为 n的整数序列 a1,a2,…,an以及一个长度为 m的整数序列 b1,b2,…,bm。请你判断 a序列是否为 b序列的子序列。子序列指序列的一部分项按原有次序排列而得的序列，例如序列
阅读更多2024-11-16

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

引言

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

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

1. 定义消息格式

消息头

消息体

2. 定义通信协议

发送消息

接收消息

消息处理

3. 示例代码

定义消息格式

发送消息

接收消息

三、发布-订阅模式

1. 定义发布-订阅模式

2. 示例代码

发布消息

订阅消息

3. 运行示例

4. 异步处理消息

5. 解耦系统

6. 实现步骤

7. 实例场景

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

场景描述

实现步骤

示例代码

订单服务发送消息

库存服务接收消息

物流服务接收消息

相关文章