分布式系统通信解决方案：Netty Marshalling 全面解析

🕗 发布于 2025-01-22 08:07 java 后端架构 spring

分布式系统通信解决方案：Netty Marshalling 全面解析

一、引言

在现代网络编程中，Netty 作为一款高性能、异步事件驱动的网络应用框架，因其强大的功能和灵活的扩展性，备受开发者青睐。Netty 广泛应用于分布式系统、RPC 框架以及高并发场景中，是构建高效网络服务的利器。

在我之前的博客《构建高性能网络服务：从 Socket 原理到 Netty 应用实践》中，详细探讨了从传统的 Socket 通信原理到 Netty 应用的基本实践内容，包括：

Socket 编程的局限性：传统阻塞式 IO 在高并发场景中的瓶颈。
NIO 的优势与挑战：通过多路复用和非阻塞 IO 提升效率，但编程复杂度较高。
Netty 的解决方案：屏蔽底层复杂性，提供直观 API，支持高性能网络通信。

在实际开发中，数据的编解码 是网络传输中的关键环节。Netty 提供了强大的工具和机制，如 ByteBuf、编码解码器，以及更高级的序列化方案，极大简化了开发工作。

本文将重点介绍 Netty 的高级编解码技术之一： Marshalling。这是基于 JBoss Marshalling 项目实现的一种高效序列化机制。通过本文，你将学习如何利用 Marshalling 实现 Java 对象的高效传输，同时结合其他编解码工具构建稳定、高性能的网络应用。

如果你还不熟悉 Netty 的基础知识，可以先参考我的博客《构建高性能网络服务：从 Socket 原理到 Netty 应用实践》，获取必要的背景知识。

二、什么是 Marshalling

Marshalling 是 Netty 提供的一种序列化机制，基于 JBoss 的 Marshalling 项目实现。它用于将 Java 对象转换为字节流以便传输，以及将字节流反序列化回 Java 对象。

相比于 Java 自带的 ObjectInputStream 和 ObjectOutputStream，Marshalling 具有以下优势：

性能更优：序列化和反序列化的效率更高。
可扩展性强：支持自定义序列化策略。
更易集成：与 Netty 无缝集成，提供专用的编解码器。

三、Marshalling 的核心组件

在 Netty 中，Marshalling 的主要职责是高效地完成 Java 对象与字节流之间的转换，适用于需要序列化复杂对象的网络传输场景。Netty 提供了两大核心组件与其相关联：

1. MarshallingEncoder

功能：
负责将 Java 对象序列化为字节流并写入到 ByteBuf 中，便于通过网络传输。它是一个 MessageToByteEncoder 的实现类，专为对象序列化设计。

关键特性：

高效性：通过底层优化的序列化机制，提升对象转换的性能。
流式操作：可以将复杂的对象轻松转换为网络流中的字节表示。

典型使用场景：

在客户端或服务端发送包含复杂对象的数据包时，利用 MarshallingEncoder 将对象序列化为可传输的字节流。
适用于传输包含嵌套结构的对象，如 JSON、XML 或复杂 POJO。

2. MarshallingDecoder

功能：
负责将接收到的字节流反序列化为 Java 对象。它是一个 ByteToMessageDecoder 的实现类，能够从 ByteBuf 中读取字节流并还原为原始对象。

关键特性：

自动拆包：根据字节流解析完整对象，避免由于 TCP 拆包或黏包导致的数据不完整问题。
灵活性：可结合其他解码器（如 DelimiterBasedFrameDecoder 或 LengthFieldBasedFrameDecoder）实现定制化数据解析。

典型使用场景：

在服务端或客户端接收包含序列化对象的数据包时，使用 MarshallingDecoder 将其反序列化为具体的 Java 对象。
适用于需要还原复杂对象的场景，如 RPC 调用、分布式系统通信等。

3. Marshalling 编解码器的协作工作流程

为了让 MarshallingEncoder 和 MarshallingDecoder 正常协作工作，我们需要将它们添加到 Netty 的 ChannelPipeline 中。整个工作流程如下：

发送端：将 Java 对象交由 MarshallingEncoder 序列化为字节流，再通过其他编码器（如长度字段或分隔符解码器）添加传输标识。
接收端：接收到字节流后，MarshallingDecoder 将其反序列化为 Java 对象。

示意图：

lua复制编辑+-----------------------+          +-----------------------+
|    MarshallingEncoder | ----->  |    网络传输（字节流）  | ----->  |    MarshallingDecoder |
+-----------------------+          +-----------------------+
    （对象序列化）                         （传输数据）                （对象反序列化）

4. 配合定制化工具的使用

MarshallingEncoder 和 MarshallingDecoder 可与其他 Netty 编解码器（如 DelimiterBasedFrameDecoder 或 LengthFieldBasedFrameDecoder）结合使用，以解决 TCP 拆包黏包问题并优化网络传输。具体例子将在后续示例中详细展示。

通过对 MarshallingEncoder 和 MarshallingDecoder 的合理应用，我们可以高效、安全地实现对象的序列化与反序列化操作，为构建复杂的网络传输方案提供坚实基础。

四、快速入门

场景描述

我们实现一个简单的库存管理系统，服务端接收客户端发送的库存变动请求，并返回处理结果。每次请求和响应都包含复杂的 Java 对象，例如 InventoryRequest 和 InventoryResponse。

这些对象包含多层嵌套结构和集合数据，适合展示 Marshalling 的序列化能力。

4.1 引入依赖

在 Maven 项目中添加以下依赖：

<dependency>
    <groupId>org.jboss.marshalling</groupId>
    <artifactId>jboss-marshalling</artifactId>
    <version>2.0.12.Final</version>
</dependency>
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.97.Final</version>
</dependency>

4.2 初始化 Marshalling 编解码器

示例代码

1. 创建复杂的传输对象

定义两个对象：InventoryRequest 和 InventoryResponse，并实现 Serializable 接口以支持序列化。

创建一个工具类 MarshallingCodeCFactory 来初始化编解码器：

import java.io.Serializable;
import java.util.List;

// 库存请求对象
public class InventoryRequest implements Serializable {
    private static final long serialVersionUID = 1L;
    private String productId; // 产品 ID
    private int quantity;     // 请求的数量
    private String operation; // 操作类型：增加(add) 或 减少(remove)

    // 构造方法、getter 和 setter
    public InventoryRequest(String productId, int quantity, String operation) {
        this.productId = productId;
        this.quantity = quantity;
        this.operation = operation;
    }

    @Override
    public String toString() {
        return "InventoryRequest{" +
                "productId='" + productId + '\'' +
                ", quantity=" + quantity +
                ", operation='" + operation + '\'' +
                '}';
    }
}

// 库存响应对象
public class InventoryResponse implements Serializable {
    private static final long serialVersionUID = 1L;
    private String productId; // 产品 ID
    private boolean success;  // 是否成功
    private String message;   // 响应消息

    // 构造方法、getter 和 setter
    public InventoryResponse(String productId, boolean success, String message) {
        this.productId = productId;
        this.success = success;
        this.message = message;
    }

    @Override
    public String toString() {
        return "InventoryResponse{" +
                "productId='" + productId + '\'' +
                ", success=" + success +
                ", message='" + message + '\'' +
                '}';
    }
}

4.3 服务端代码

2. 服务端实现

服务端处理逻辑（Handler）：

服务端接收 InventoryRequest 对象，解析后根据操作类型更新库存，并返回 InventoryResponse 对象。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class InventoryServerHandler extends SimpleChannelInboundHandler<InventoryRequest> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, InventoryRequest request) {
        System.out.println("收到客户端请求：" + request);

        // 模拟库存处理逻辑
        boolean success = "add".equals(request.getOperation()) || "remove".equals(request.getOperation());
        String message = success ? "操作成功！" : "操作失败，未知操作类型：" + request.getOperation();

        // 构造响应对象
        InventoryResponse response = new InventoryResponse(request.getProductId(), success, message);

        // 发送响应
        ctx.writeAndFlush(response);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

服务端启动类

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class InventoryServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
                            ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
                            ch.pipeline().addLast(new InventoryServerHandler());
                        }
                    });

            ChannelFuture future = bootstrap.bind(8080).sync();
            System.out.println("服务端启动，端口：8080");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

4.4 客户端代码

客户端发送逻辑（Handler）：

客户端发送一个 InventoryRequest，并接收服务端返回的 InventoryResponse。

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class InventoryClientHandler extends SimpleChannelInboundHandler<InventoryResponse> {
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 构造一个请求对象并发送
        InventoryRequest request = new InventoryRequest("P12345", 10, "add");
        ctx.writeAndFlush(request);
        System.out.println("客户端已发送请求：" + request);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, InventoryResponse response) {
        // 打印服务端响应
        System.out.println("收到服务端响应：" + response);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

客户端启动类

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class InventoryClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
                            ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
                            ch.pipeline().addLast(new InventoryClientHandler());
                        }
                    });

            ChannelFuture future = bootstrap.connect("localhost", 8080).sync();
            future.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

五、TCP 拆包与黏包问题

虽然 Marshalling 解决了数据的序列化与反序列化问题，但在基于 TCP 的网络通信中，可能会遇到拆包和黏包现象。

5.1 什么是拆包和黏包？

拆包和黏包是 TCP 协议的常见问题，其本质原因在于 TCP 是一种流式传输协议，不保证消息边界：

拆包： 发送的数据包过大，接收端一次读取不完整，导致数据被分割成多个包。
黏包： 发送的数据包较小，多个消息被拼接在一个 TCP 包中，接收端无法区分边界。

在我之前的博客《构建高性能网络服务：从 Socket 原理到 Netty 应用实践》中，已详细分析了 TCP 拆包与黏包的成因和常见处理方法，建议读者参考相关内容。

5.2 Netty 提供的解决方案

针对 TCP 拆包和黏包问题，Netty 提供了以下通用解码器，可以与 Marshalling 协作使用：

定长帧解码器：
使用 FixedLengthFrameDecoder，通过指定消息的固定长度，强制按照长度切分数据包。

适用场景： 消息格式固定、长度已知的简单协议。
```
ch.pipeline().addLast(new FixedLengthFrameDecoder(1024)); // 每次读取固定 1024 字节
```
特殊分隔符解码器：
使用 DelimiterBasedFrameDecoder，指定特殊字符（如 \n 或自定义符号）作为消息边界。

适用场景： 消息中含有明确的结束符，如基于文本的协议。
```
ByteBuf delimiter = Unpooled.copiedBuffer("_$".getBytes());
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(8192, delimiter));
```

长度字段解码器：
使用 LengthFieldBasedFrameDecoder，通过消息头携带的长度信息解析数据包。

适用场景： 自定义协议中包含消息长度字段。

ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(
        65536, // 最大帧长度
        0,     // 长度字段的偏移
        4,     // 长度字段的字节数
        0,     // 长度调整值
        4      // 跳过的初始字节数
));

5.3 Marshalling 的协作

在基于 Marshalling 的对象传输中，消息往往是复杂 Java 对象，直接使用 MarshallingDecoder 和 MarshallingEncoder 可能会因为拆包和黏包问题导致数据不完整或出错。为避免此问题，需结合上述解码器解决。

典型组合示例：

ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4)); // 处理拆包黏包
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());   // 反序列化
ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());   // 序列化

小结：

通过明确消息边界，可以有效避免拆包和黏包问题。
在本博客的复杂对象传输示例中，Marshalling 通常结合 LengthFieldBasedFrameDecoder 使用，确保消息的完整性和可靠性。

六、总结

在本篇博客中，我们探讨了 Netty 提供的高级编解码工具 Marshalling，并结合实际示例展示了如何通过它高效地实现 Java 对象的序列化与反序列化。以下是关键点回顾：

1. 核心内容回顾

Marshalling 的作用：
- 提供了一种高效的序列化机制，适用于 Java 对象的网络传输。
- 基于 JBoss Marshalling 项目实现，性能优于传统的 ObjectInputStream 和 ObjectOutputStream。
Marshalling 的核心组件：
- MarshallingEncoder：负责对象序列化为字节流。
- MarshallingDecoder：负责字节流反序列化为 Java 对象。
典型使用场景：
- 复杂 Java 对象的网络传输，如分布式系统通信、RPC 框架中数据的序列化与反序列化。

2. 实践中的重要建议

结合解码器处理拆包与黏包问题：
在 TCP 网络传输中，结合 Netty 提供的 LengthFieldBasedFrameDecoder 等解码器使用，可以有效避免因拆包或黏包导致的传输错误。
性能优化：
- 针对性能需求较高的场景，可通过自定义序列化器（如实现 Externalizer 接口）优化特定对象的序列化效率。
- 对序列化后的数据进行压缩，进一步减少传输开销。
安全性与兼容性：
- 使用 serialVersionUID 确保对象序列化的向后兼容。
- 在反序列化过程中，限制可加载的类，防范反序列化漏洞。

3. 学习与拓展

要全面掌握 Marshalling，可以从以下方面进一步学习和实践：

深入理解序列化机制： 研究 Marshalling 的内部实现，了解其性能优势和扩展能力。
对比其他序列化方案： 例如 Protobuf、Kryo 等，分析其优劣与适用场景。
实际项目应用： 在微服务、分布式系统或 RPC 框架中尝试集成和优化 Marshalling。

原文地址：https://blog.csdn.net/qq_36534560/article/details/145284919

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

上一篇：Java 在包管理与模块化中的优势：与其他开发语言的比较
下一篇：索引（MySQL）

【优选算法】3----快乐数
这是第三篇算法题，也是十分的不好想，虽然定位在简单，但我感觉还是有些难度的~
阅读更多2025-01-22
算法竞赛之差分进阶——等差数列差分 python
给定区间 [ l, r ]，把数组[ l, r ] 区间中的数加上一个首项s、末项e、公差为d的等差数列如何实现？
阅读更多2025-01-22
Linux:进程控制
• 另外，进程⼀旦变成僵⼫状态，那就⼑枪不⼊，“杀⼈不眨眼”的kill -9 也⽆能为⼒，因为谁也没有办法杀死⼀个已经死去的进程。通常，⽗⼦代码共享，⽗⼦再不写⼊时，数据也是共享的，当任意⼀⽅试图写⼊
阅读更多2025-01-22
2024：CSDN上的收获与蜕变——我的技术成长之旅
2024：CSDN上的收获与蜕变——我的技术成长之旅
阅读更多2025-01-22
【C】memory 详解
【C】memory详解
阅读更多2025-01-22
OpenMP并行编程实例系列3 —— 线程设置
等待同步：用在并行域内，所有线程执行到 barrier 都要停下等待，直到所有线程都执行到 barrier，然后再继续往下执行。指定并行域中的串行任务，创建仅由一个线程执行的任务，先到先执行，其他线程
阅读更多2025-01-22
【GitHub】登录时的2FA验证
如何进行2FA认证，以及验证后如何在新设备或新浏览器登录
阅读更多2025-01-22
【Python项目】小区监控图像拼接系统
科技的介入正在重塑我们的生活方式，而越来越多的人也开始享受这种由科技带来的便利和变化。例如，监控画面分散，安保人员需要在多个屏幕上查看不同的区域，这不仅增加了管理难度，也降低了监控效率。该系统利用先进
阅读更多2025-01-22
Mysql触发器（学习自用）
仅供学习使用，如有错误，欢迎指正。
阅读更多2025-01-22
堆的实现（C语言详解版）
（Heap）是一种特殊的完全二叉树，它满足父节点的值总是不大于或不小于其子节点的值。这种数据结构常用于实现优先队列，以及在各种排序算法中快速找到最大或最小元素。和。在最大堆中，父节点的值总是大于或等于
阅读更多2025-01-22

分布式系统通信解决方案：Netty Marshalling 全面解析