Netty的序列化之MessagePack

news2024/11/15 20:37:44

目录

引入MessagePack依赖

实体类

服务端代码

客户端代码

执行结果

引入MessagePack依赖

        <dependency>
            <groupId>org.msgpack</groupId>
            <artifactId>msgpack</artifactId>
            <version>0.6.12</version>
        </dependency>

实体类

@Message//MessagePack提供的注解，表明这是一个需要序列化的实体类
public class User {
    private String id;
    private String userName;
    private int age;
    private UserContact userContact;

    public User(String userName, int age, String id) {
        this.userName = userName;
        this.age = age;
        this.id = id;
    }

    public User() {
    }

    public String getUserName() {
        return userName;
    }

    public void setUserName(String userName) {
        this.userName = userName;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public UserContact getUserContact() {
        return userContact;
    }

    public void setUserContact(UserContact userContact) {
        this.userContact = userContact;
    }

    @Override
    public String toString() {
        return "User{" +
                "userName='" + userName + '\'' +
                ", age=" + age +
                ", id='" + id + '\'' +
                ", userContact=" + userContact +
                '}';
    }
}

@Message//MessagePack提供的注解，表明这是一个需要序列化的实体类
public class UserContact {
    private String mail;
    private String phone;

    public UserContact() {
    }

    public UserContact(String mail, String phone) {
        this.mail = mail;
        this.phone = phone;
    }

    public String getMail() {
        return mail;
    }

    public void setMail(String mail) {
        this.mail = mail;
    }

    public String getPhone() {
        return phone;
    }

    public void setPhone(String phone) {
        this.phone = phone;
    }

    @Override
    public String toString() {
        return "UserContact{" +
                "mail='" + mail + '\'' +
                ", phone='" + phone + '\'' +
                '}';
    }
}

服务端代码

public class ServerMsgPackEcho {

    public static final int PORT = 9995;

    public static void main(String[] args) throws InterruptedException {
        ServerMsgPackEcho serverMsgPackEcho = new ServerMsgPackEcho();
        System.out.println("服务器即将启动");
        serverMsgPackEcho.start();
    }

    public void start() throws InterruptedException {
        final MsgPackServerHandler serverHandler = new MsgPackServerHandler();
        EventLoopGroup group = new NioEventLoopGroup();/*线程组*/

        try {
            ServerBootstrap b = new ServerBootstrap();/*服务端启动必须*/
            b.group(group)/*将线程组传入*/
                .channel(NioServerSocketChannel.class)/*指定使用NIO进行网络传输*/
                .localAddress(new InetSocketAddress(PORT))/*指定服务器监听端口*/
                /*服务端每接收到一个连接请求，就会新启一个socket通信，也就是channel，
                所以下面这段代码的作用就是为这个子channel增加handle*/
                .childHandler(new ChannelInitializerImp());
            ChannelFuture f = b.bind().sync();/*异步绑定到服务器，sync()会阻塞直到完成*/
            System.out.println("服务器启动完成，等待客户端的连接和数据.....");
            f.channel().closeFuture().sync();/*阻塞直到服务器的channel关闭*/
        } finally {
            group.shutdownGracefully().sync();/*优雅关闭线程组*/
        }
    }

    private static class ChannelInitializerImp extends ChannelInitializer<Channel> {

        @Override
        protected void initChannel(Channel ch) throws Exception {
            ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535,
                    0,2,0,
                    2));
            ch.pipeline().addLast(new MsgPackDecoder());
            ch.pipeline().addLast(new MsgPackServerHandler());
        }
    }
}


/*基于MessagePack的解码器，反序列化*/
public class MsgPackDecoder extends MessageToMessageDecoder<ByteBuf> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out)
            throws Exception {
        final int length = msg.readableBytes();
        final byte[] array = new byte[length];
        msg.getBytes(msg.readerIndex(),array,0,length);
        MessagePack messagePack = new MessagePack();
        out.add(messagePack.read(array,User.class));
    }
}



@ChannelHandler.Sharable
public class MsgPackServerHandler extends ChannelInboundHandlerAdapter {

    private AtomicInteger counter = new AtomicInteger(0);

    /*** 服务端读取到网络数据后的处理*/
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //将上一个handler生成的数据强制转型
        User user = (User)msg;
        System.out.println("Server Accept["+user
                +"] and the counter is:"+counter.incrementAndGet());
        //服务器的应答
        String resp = "I process user :"+user.getUserName()
                + System.getProperty("line.separator");
        ctx.writeAndFlush(Unpooled.copiedBuffer(resp.getBytes()));
        ctx.fireChannelRead(user);
    }

    /*** 发生异常后的处理*/
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

客户端代码

public class ClientMsgPackEcho {

    private final String host;

    public ClientMsgPackEcho(String host) {
        this.host = host;
    }

    public void start() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
        try {
            final Bootstrap b = new Bootstrap();;/*客户端启动必须*/
            b.group(group)/*将线程组传入*/
                    .channel(NioSocketChannel.class)/*指定使用NIO进行网络传输*/
                    /*配置要连接服务器的ip地址和端口*/
                    .remoteAddress(
                            new InetSocketAddress(host, ServerMsgPackEcho.PORT))
                    .handler(new ChannelInitializerImp());
            ChannelFuture f = b.connect().sync();
            System.out.println("已连接到服务器.....");
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully().sync();
        }
    }

    private static class ChannelInitializerImp extends ChannelInitializer<Channel> {

        @Override
        protected void initChannel(Channel ch) throws Exception {
            /*告诉netty，计算一下报文的长度，然后作为报文头加在前面*/
            ch.pipeline().addLast(new LengthFieldPrepender(2));
            /*对服务器的应答也要解码，解决粘包半包*/
            ch.pipeline().addLast(new LineBasedFrameDecoder(1024));

            /*对我们要发送的数据做编码-序列化*/
           ch.pipeline().addLast(new MsgPackEncode());
           ch.pipeline().addLast(new MsgPackClientHandler(5));
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new ClientMsgPackEcho("127.0.0.1").start();
    }
}



/*基于MessagePack的编码器，序列化*/
public class MsgPackEncode extends MessageToByteEncoder<User> {
    @Override
    protected void encode(ChannelHandlerContext ctx, User msg, ByteBuf out)
            throws Exception {
        MessagePack messagePack = new MessagePack();
        byte[] raw = messagePack.write(msg);
        out.writeBytes(raw);

    }
}


public class MsgPackClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private final int sendNumber;

    public MsgPackClientHandler(int sendNumber) {
        this.sendNumber = sendNumber;
    }

    private AtomicInteger counter = new AtomicInteger(0);

    /*** 客户端读取到网络数据后的处理*/
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        System.out.println("client Accept["+msg.toString(CharsetUtil.UTF_8)
                +"] and the counter is:"+counter.incrementAndGet());
    }

    /*** 客户端被通知channel活跃后，做事*/
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        User[] users = makeUsers();
        //发送数据
        for(User user:users){
            System.out.println("Send user:"+user);
            ctx.write(user);
        }
        ctx.flush();
    }

    /*** 发生异常后的处理*/
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    /*生成用户实体类的数组，以供发送*/
    private User[] makeUsers(){
        User[] users=new User[sendNumber];
        User user =null;
        for(int i=0;i<sendNumber;i++){
            user=new User();
            user.setAge(i);
            String userName = "ABC--->"+i;
            user.setUserName(userName);
            user.setId("No:"+(sendNumber-i));
            user.setUserContact(
                    new UserContact(userName+"@9527.com","133333333"));
            users[i]=user;
        }
        return users;
    }
}

核心就是通过MessagePack提供的处理器来处理数据。

执行结果

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1440493.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Stable Diffusion 模型下载：GhostMix（幽灵混合）

Stable Diffusion 模型下载：GhostMix（幽灵混合）

文章目录 模型介绍生成案例案例一案例二案例三案例四案例五案例六案例七案例八案例九案例十 下载地址 模型介绍 GhostMix 是绝对让你惊艳的模型&#xff0c;也是自己认为现在最强的2.5D模型。我认为模型的更新应该是基于现有的画面整体不大变的前提下&#xff0c;提高模型的成…
阅读更多...
springboot167基于springboot的医院后台管理系统的设计与实现

springboot167基于springboot的医院后台管理系统的设计与实现

简介 【毕设源码推荐 javaweb 项目】基于springbootvue 的 适用于计算机类毕业设计&#xff0c;课程设计参考与学习用途。仅供学习参考&#xff0c; 不得用于商业或者非法用途&#xff0c;否则&#xff0c;一切后果请用户自负。 看运行截图看 第五章 第四章 获取资料方式 **项…
阅读更多...
大数据术语系列（1）——COW和MOR，我如何使用chatgpt通俗易懂地理解了hudi这两种表类型

大数据术语系列（1）——COW和MOR，我如何使用chatgpt通俗易懂地理解了hudi这两种表类型

从传统数据库到大数据的转变&#xff0c;首当其冲的是各种术语的理解。 所以我与chatgpt发生了一系列对话&#xff0c;以便于我能快速理解这些术语。 我先把汇总的结果放在前边&#xff0c;后边会一步步地来说明我是如何获取这些信息的。前边我也发过一些关于chatgpt提示词相…
阅读更多...
图像处理入门：OpenCV的基础用法解析

图像处理入门：OpenCV的基础用法解析

图像处理入门&#xff1a;OpenCV的基础用法解析 引言OpenCV的初步了解深入理解OpenCV&#xff1a;计算机视觉的开源解决方案什么是OpenCV&#xff1f;OpenCV的主要功能1. 图像处理2. 图像分析3. 结构分析和形状描述4. 动态分析5. 三维重建6. 机器学习7. 目标检测 OpenCV的应用场…
阅读更多...
Go 语言中如何大小端字节序？int 转 byte 是如何进行的？

Go 语言中如何大小端字节序？int 转 byte 是如何进行的？

嗨&#xff0c;大家好&#xff01;我是波罗学。 本文是系列文章 Go 技巧第十五篇&#xff0c;系列文章查看&#xff1a;Go 语言技巧。 我们先看这样一个问题&#xff1a;“Go 语言中&#xff0c;将 byte 转换为 int 时是否涉及字节序&#xff08;endianness&#xff09;&#x…
阅读更多...
Verilog刷题笔记26

Verilog刷题笔记26

题目&#xff1a; Build a combinational circuit with 100 inputs, in[99:0]. There are 3 outputs: out_and: output of a 100-input AND gate. out_or: output of a 100-input OR gate. out_xor: output of a 100-input XOR gate. 解题&#xff1a; module top_module( …
阅读更多...
springboot157基于springboot的线上辅导班系统的开发与设计

springboot157基于springboot的线上辅导班系统的开发与设计

简介 【毕设源码推荐 javaweb 项目】基于springbootvue 的 适用于计算机类毕业设计&#xff0c;课程设计参考与学习用途。仅供学习参考&#xff0c; 不得用于商业或者非法用途&#xff0c;否则&#xff0c;一切后果请用户自负。 看运行截图看 第五章 第四章 获取资料方式 **项…
阅读更多...
数字图像处理实验记录八（图像压缩实验）

数字图像处理实验记录八（图像压缩实验）

前言&#xff1a;做这个实验的时候很忙&#xff0c;就都是你抄我我抄你了 一、基础知识 1&#xff0e;为什么要进行图像压缩&#xff1a; 图像的数据量巨大&#xff0c;对计算机的处理速度、存储容量要求高。传输信道带宽、通信链路容量一定&#xff0c;需要减少传输数据量&a…
阅读更多...
字节跳动公益平台“公益聚力计划”上线

字节跳动公益平台“公益聚力计划”上线

为更好地联合社会多方力量参与社会公益&#xff0c;字节跳动公益平台于近日正式推出“公益聚力计划”&#xff08;以下简称“计划”&#xff09;。“计划”支持公益项目的策划与筛选、公益机构撮合&#xff0c;以及多种定制化的产品功能&#xff0c;如定制版公益证书、爱心回礼…
阅读更多...
深入探索：缓冲区溢出漏洞及其防范策略

深入探索：缓冲区溢出漏洞及其防范策略

在网络安全的广阔领域中&#xff0c;缓冲区溢出漏洞一直是一个重要的议题。这种漏洞&#xff0c;如果被恶意利用&#xff0c;可能会导致严重的安全问题&#xff0c;包括数据泄露、系统崩溃&#xff0c;甚至可能被攻击者利用来执行恶意代码。在本文中&#xff0c;我们将深入探讨…
阅读更多...
2月3日作业

2月3日作业

1.编程实现单向循环链表的头插&#xff0c;头删、尾插、尾删 尾插/头插&#xff0c;头删&#xff0c;尾删&#xff1a; 头文件&#xff1a; #ifndef __HEAD_H_ #define __HEAD_H_#include<stdio.h> #include<string.h> #include<stdlib.h>enum {FALSE-1,SU…
阅读更多...
18：蜂鸣器

18：蜂鸣器

蜂鸣器 1、蜂鸣器的介绍2、编程让蜂鸣器响起来3、通过定时控制蜂鸣器4、蜂鸣器发出滴滴声&#xff08;间歇性鸣叫&#xff09; 1、蜂鸣器的介绍 蜂鸣器内部其实是2个金属片&#xff0c;当一个金属片接正电&#xff0c;一个金属片接负电时&#xff0c;2个金属片将合拢&#xff…
阅读更多...
二叉搜索树题目：验证二叉搜索树

二叉搜索树题目：验证二叉搜索树

文章目录 题目标题和出处难度题目描述要求示例数据范围 解法一思路和算法代码复杂度分析 解法二思路和算法代码复杂度分析 解法三思路和算法代码复杂度分析 题目 标题和出处 标题&#xff1a;验证二叉搜索树 出处&#xff1a;98. 验证二叉搜索树 难度 3 级 题目描述 要求…
阅读更多...
K8S之Pod常见的状态和重启策略

K8S之Pod常见的状态和重启策略

Pod常见的状态和重启策略 常见的Pod状态PendingPodScheduledUnschedulablePodInitializingImagePullBackOffInitializedRunningErrorCrashLoopBackOffTerminatingSucceededFailedEvictedUnknown Pod的重启策略使用Always重启策略使用Never重启策略使用OnFailure重启策略(常用) …
阅读更多...
16、prometheus + grafana + alertmanager

16、prometheus + grafana + alertmanager

16、prometheus grafana alertmanager k8s 手撕方式安装 prometheus grafana alertmanager k8s版本&#xff1a;k8s-1.29.1 prometheus grafana alertmanager 监控报警 1、k8s 手撕方式安装 prometheus mkdir ~/prometheus-ymlkubectl create ns monitoringcat > ~/…
阅读更多...
python创建udf函数步骤

python创建udf函数步骤

一、目标 实现一个函数&#xff0c;传入两个datetime类型的参数&#xff0c;返回double类型的工作日天数 二、思路 如何计算差值&#xff1f; 如果开始时间和结束时间在同一天&#xff1a;实现同 datediff(end, start, ‘ss’) / 86400.0 如果开始时间和结束时间在不同天&am…
阅读更多...
c语言--一维数组传参的本质（详解）

c语言--一维数组传参的本质（详解）

目录 一、前言二、代码三、形式3.1形式13.2形式2 四、总结 一、前言 首先从⼀个问题开始&#xff0c;我们之前都是在函数外部计算数组的元素个数&#xff0c;那我们可以把函数传给⼀个函数后&#xff0c;函数内部求数组的元素个数吗&#xff1f; 二、代码 直接上代码&#x…
阅读更多...
代码随想录算法训练营第十五天|102.二叉树的层序遍历、226.翻转二叉树

代码随想录算法训练营第十五天|102.二叉树的层序遍历、226.翻转二叉树

102.二叉树的层序遍历 刷题https://leetcode.cn/problems/binary-tree-level-order-traversal/description/文章讲解https://programmercarl.com/0102.%E4%BA%8C%E5%8F%89%E6%A0%91%E7%9A%84%E5%B1%82%E5%BA%8F%E9%81%8D%E5%8E%86.html视频讲解https://www.bilibili.com/video…
阅读更多...
Mountain Lake - Forest Pack

Mountain Lake - Forest Pack

从头开始构建的50个岩石森林资源集合，充分利用了HDRP。还支持Universal 和Built-In。 支持Unity 2020.3+、高清渲染管线、通用渲染管线、标准渲染管线。导入包后，按照README中的说明进行操作。 Mountain Lake - Rock & Tree Pack是一个由50个准备好的资源组成的集合，从头…
阅读更多...
Python datetime 模块的高级应用

Python datetime 模块的高级应用

Python datetime 模块的高级应用 介绍方法时区处理日期格式化日期计算常见问题及解决方案代码日历应用时间序列分析 介绍 datetime 模块是 Python 中用于处理日期和时间的标准库模块。它提供了日期和时间类型&#xff08;date、time、datetime&#xff09;以及与日期和时间相关…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023