自定义Encoder/Decoder进行对象传递

转载:http://blog.csdn.net/top_code/article/details/50901623

在上一篇文章中，我们使用Netty4本身自带的ObjectDecoder，ObjectEncoder来实现POJO对象的传输，但其使用的是Java内置的序列化，由于Java序列化的性能并不是很好，所以很多时候我们需要用其他高效的序列化方式，例如 protobuf，Hessian， Kryo，Jackson，fastjson等。

本文中Java序列化不是重点，对Java序列化不熟悉的同学的请自行查找资料学习，本篇我们重点介绍如何构造我们的Encoder和Decoder 。

流式传输特点

In a stream-based transport such as TCP/IP, received data is stored into a socket receive buffer. Unfortunately, the buffer of a stream-based transport is not a queue of packets but a queue of bytes. It means, even if you sent two messages as two independent packets, an operating system will not treat them as two messages but as just a bunch of bytes. Therefore, there is no guarantee that what you read is exactly what your remote peer wrote. For example, let us assume that the TCP/IP stack of an operating system has received three packets: 
 
Because of this general property of a stream-based protocol, there’s high chance of reading them in the following fragmented form in your application: 

Therefore, a receiving part, regardless it is server-side or client-side, should defrag the received data into one or more meaningful frames that could be easily understood by the application logic. In case of the example above, the received data should be framed like the following: 

通常情况下有下面几种解决方案：

  1. 消息定长
  2. 在包尾增加一个标识，通过这个标志符进行分割
  3. 将消息分为两部分，也就是消息头和消息尾，消息头中写入要发送数据的总长度，通常是在消息头的第一个字段使用int值（如果消息很大可以考虑用long值）来标识发送数据的长度。

本文中采用第三种方案，自定义Encoder/Decoder进行对象的传输。

准备工作

JDK 7 
Eclipse Juno 
Maven 3.3

序列化框架

本篇我们使用Kryo对POJO对象进行序列化，当然也可以采用protobuf，Hessian做序列化，有兴趣的同学可以自己动手试试。

1、添加Kyro 依赖

<dependency>
            <groupId>com.esotericsoftware</groupId>
            <artifactId>kryo</artifactId>
            <version>3.0.3</version>
        </dependency>

2、自定义Encoder 
首先我们实现一个Encoder，继承自MessageToByteEncoder

package com.ricky.codelab.netty.ch3.serialiaztion;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

import java.io.ByteArrayOutputStream;

import org.apache.commons.io.IOUtils;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoException;
import com.esotericsoftware.kryo.io.Output;
import com.ricky.codelab.netty.model.Car;

/**
 * 自定义Encoder
 * @author Ricky
 *
 */
public class KyroMsgEncoder extends MessageToByteEncoder<Car> {

    private Kryo kryo = new Kryo();

    @Override
    protected void encode(ChannelHandlerContext ctx, Car msg, ByteBuf out) throws Exception {

        byte[] body = convertToBytes(msg);  //将对象转换为byte
        int dataLength = body.length;  //读取消息的长度
        out.writeInt(dataLength);  //先将消息长度写入，也就是消息头
        out.writeBytes(body);  //消息体中包含我们要发送的数据
    }

    private byte[] convertToBytes(Car car) {

        ByteArrayOutputStream bos = null;
        Output output = null;
        try {
            bos = new ByteArrayOutputStream();
            output = new Output(bos);
            kryo.writeObject(output, car);
            output.flush();

            return bos.toByteArray();
        } catch (KryoException e) {
            e.printStackTrace();
        }finally{
            IOUtils.closeQuietly(output);
            IOUtils.closeQuietly(bos);
        }
        return null;
    }

}

在KyroMsgEncoder中我们需要覆盖 encode(ChannelHandlerContext ctx, Car msg, ByteBuf out) 方法，其主要用来将要传输的对象转换为byte数组。

3、自定义Decoder 
自定义Decoder 需继承ByteToMessageDecoder类，并覆盖其decode方法。

package com.ricky.codelab.netty.ch3.serialiaztion;

import java.io.ByteArrayInputStream;
import java.util.List;

import org.apache.commons.io.IOUtils;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoException;
import com.esotericsoftware.kryo.io.Input;
import com.ricky.codelab.netty.model.Car;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

/**
 * 自定义Decoder
 * @author Ricky
 *
 */
public class KyroMsgDecoder extends ByteToMessageDecoder {

    public static final int HEAD_LENGTH = 4;

    private Kryo kryo = new Kryo();

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

        if (in.readableBytes() < HEAD_LENGTH) {  //这个HEAD_LENGTH是我们用于表示头长度的字节数。  由于Encoder中我们传的是一个int类型的值，所以这里HEAD_LENGTH的值为4.
            return;
        }
        in.markReaderIndex();                  //我们标记一下当前的readIndex的位置
        int dataLength = in.readInt();       // 读取传送过来的消息的长度。ByteBuf 的readInt()方法会让他的readIndex增加4
        if (dataLength < 0) { // 我们读到的消息体长度为0，这是不应该出现的情况，这里出现这情况，关闭连接。
            ctx.close();
        }

        if (in.readableBytes() < dataLength) { //读到的消息体长度如果小于我们传送过来的消息长度，则resetReaderIndex. 这个配合markReaderIndex使用的。把readIndex重置到mark的地方
            in.resetReaderIndex();
            return;
        }

        byte[] body = new byte[dataLength];  //传输正常
        in.readBytes(body);
        Object o = convertToObject(body);  //将byte数据转化为我们需要的对象
        out.add(o);
    }

    private Object convertToObject(byte[] body) {

        Input input = null;
        ByteArrayInputStream bais = null;
        try {
            bais = new ByteArrayInputStream(body);
            input = new Input(bais);

            return kryo.readObject(input, Car.class);
        } catch (KryoException e) {
            e.printStackTrace();
        }finally{
            IOUtils.closeQuietly(input);
            IOUtils.closeQuietly(bais);
        }

        return null;
    }

}

在KyroMsgDecoder中覆盖父类的decode(ChannelHandlerContext ctx, ByteBuf in, List out)方法，将byte数组转换为对象。

服务端程序

KyroTransferServer.java

package com.ricky.codelab.netty.ch3;

import com.ricky.codelab.netty.ch3.serialiaztion.KyroMsgDecoder;
import com.ricky.codelab.netty.ch3.serialiaztion.KyroMsgEncoder;
import com.ricky.codelab.netty.util.Constant;

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

/**
 * Netty4.x 自定义Decoder，Encoder进行对象传递
 * @author Ricky
 *
 */
public class KyroTransferServer {

    private final int port;

    public KyroTransferServer(int port) {
        this.port = port;
    }

    public void run() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new KyroMsgEncoder(),
                                    new KyroMsgDecoder(),
                                    new KyroServerHandler());
                        }
                    });

            // Bind and start to accept incoming connections.
            b.bind(port).sync().channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {

        new KyroTransferServer(Constant.PORT).run();
    }
}

KyroServerHandler.java

package com.ricky.codelab.netty.ch3;

import com.ricky.codelab.netty.model.Car;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class KyroServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {

        System.out.println("server receive msg:"+msg);

        Car car = new Car();
        car.setName("K5");
        car.setBrand("KIA");
        car.setPrice(24.5);
        car.setSpeed(196);

        System.out.println("server write msg:"+car);
        ctx.writeAndFlush(car);
    }

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

}

客户端程序

KyroTransferClient.java

package com.ricky.codelab.netty.ch3;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import com.ricky.codelab.netty.ch3.serialiaztion.KyroMsgDecoder;
import com.ricky.codelab.netty.ch3.serialiaztion.KyroMsgEncoder;
import com.ricky.codelab.netty.model.Car;
import com.ricky.codelab.netty.util.Constant;

public class KyroTransferClient {
    private String host;
    private int port;
    private Car message;

    public KyroTransferClient(String host, int port, Car message) {
        this.host = host;
        this.port = port;
        this.message = message;
    }

    public void send() throws InterruptedException {

        Bootstrap bootstrap = new Bootstrap();
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

        try {
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new KyroMsgEncoder(),
                                    new KyroMsgDecoder(),
                                    new KyroClientHandler(message));
                        }
                    });

            ChannelFuture future = bootstrap.connect(host, port).sync();

            future.channel().closeFuture().sync();
        } finally {
            eventLoopGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {

        Car message = new Car();
        message.setName("X5");
        message.setBrand("BMW");
        message.setPrice(52.6);
        message.setSpeed(200);

        new KyroTransferClient(Constant.HOST, Constant.PORT, message).send();
    }
}

KyroClientHandler.java

package com.ricky.codelab.netty.ch3;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import com.ricky.codelab.netty.model.Car;

public class KyroClientHandler extends ChannelInboundHandlerAdapter {

    private final Car message;

    /**
     * Creates a client-side handler.
     */
    public KyroClientHandler(Car message) {
        this.message = message;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // Send the message to Server
        super.channelActive(ctx);

        System.out.println("client send message:"+message);
        ctx.writeAndFlush(message);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        // you can use the Object from Server here
        System.out.println("client receive msg:"+msg);
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {

        cause.printStackTrace();
        ctx.close();
    }
}

运行测试

先运行服务端程序，然后运行客户端程序，将看到控制台有内容输出

服务端程序输出

server receive msg:Car [name=X5, brand=BMW, price=52.6, speed=200.0] 
server write msg:Car [name=K5, brand=KIA, price=24.5, speed=196.0]

客户端程序输出

client send message:Car [name=X5, brand=BMW, price=52.6, speed=200.0] 
client receive msg:Car [name=K5, brand=KIA, price=24.5, speed=196.0]

时间： 2024-10-13 05:05:07

自定义Encoder/Decoder进行对象传递的相关文章

java的对象锁和对象传递

1.对象传递 在JAVA的参数传递中,有两种类型,第一种是基本类型传递,例如int,float,double等,这种是值传递,另外一种是对象传递,比如String或者自定义的类,这种是引用传递. 也就是说,基本类型传递的是一个副本,而对象传递的是对象本身. 2.锁 JAVA中,对象锁的概念,就是对对象进行加锁,每个对象都会有一个内存锁,如果加上锁以后,就只能让一个线程进行操作,在操作完成之前,其他线程无法对该对象进行再次操作. 3.例子 package com.itbuluoge.mythrea

Feign发送Get请求时，采用POJO对象传递参数的最终解决方案 Request method &#39;POST&#39; not supported （附带其余好几个坑）

yml: feign: httpclient: enabled: true properties: #feign feign.httpclient.enabled=true <!-- https://mvnrepository.com/artifact/io.github.openfeign/feign-httpclient --> <dependency> <groupId>io.github.openfeign</groupId> <artifac

c#进程之间对象传递方法

1. 起源 KV项目下载底层重构升级决定采用独立进程进行Media下载处理,以能做到模块复用之目的,因此涉及到了独立进程间的数据传递问题. 目前进程间数据传递,多用WM_COPYDATA.共享dll.内存映射.Remoting等方式.相对来说,WM_COPYDATA方式更为简便,网上更到处是其使用方法. 而且Marshal这个静态类,其内置多种方法,可以很方便实现字符串.结构体等数据在不同进程间传递. 那么,对象呢?如何传递? 2.序列化 想到了,Newtonsoft.Json.dll这个神器.

Oracle自定义数据类型 2 (调用对象方法)

调用对象方法 调用对象方法基于类型创建表后,就可以在查询中调用对象方法 A． 创建基于对象的表语法: create   table   <表名>   of   <对象类型>意义:此表具有该类型和member方法的所有属性, 我们不能通过DBA   STUDIO的表数据编辑器来编辑数据. 例:create   table   FLIGHT_SCH_TAB   of   FLIGHT_SCH_TYPE insert   into   FLIGHT_SCH_TAB     values(

Servlet页面间对象传递的方法

Servlet页面间对象传递的方法 1.request 2.session 3.application 4.cookie 5.其他的 Servlet页面间对象传递的方法,布布扣,bubuko.com

java对象传递小解析

先上代码: 1 import org.apache.commons.lang.builder.ToStringBuilder; 2 import org.apache.commons.lang.builder.ToStringStyle; 3 import org.slf4j.Logger; 4 import org.slf4j.LoggerFactory; 5 6 /** 7 * 引用传递demo 8 * @author lishaohua 9 * time : 2016-3-24 下午5:1

使用HashMap对象传递url参数实用工具类

代码如下: package com.yanek.util; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; public class HttpUtil { /** * @param args

asp.net中利用session对象传递、共享数据[session用法]

下面介绍Asp.net中利用session对象传递.共享数据用法: 1.传递值: 首先定义将一个文本值或单独一个值赋予session,如下: session["name"]=textbox1.text:将文本1的值赋给了session变量name,当调查到其它页面时,此值可以传递,依然存在,下面是调用或判断此值. If(session["name"]==null) {} Else { lable1.text=session["name"].tos

Intent之对象传递（Parcelable传递对象和对象集合）

接着上一篇文章,下面我们讨论一下如何利用Parcelable实现Intent之间对象的传递 一.实现对象传递 首先创建User.java实现Parcelable接口: package org.yayun.demo; import java.io.Serializable; import android.R.integer; import android.os.Parcel; import android.os.Parcelable; public class User implements Pa