介紹下Netty中常用的編碼器和解碼器

作者：花花和Java 2020-12-22 07:58:46

這篇文章，介紹什么是編碼器、解碼器，也講述了如何實(shí)戰(zhàn)中運(yùn)用編碼器和解碼器。希望能對(duì)有所幫助。

前面文章介紹Netty相關(guān)知識(shí)點(diǎn)。接下來將介紹下在通信過程中用的編碼器和解碼器。這里會(huì)不會(huì)聯(lián)想到諜戰(zhàn)戲里面。發(fā)送情報(bào)者怕情報(bào)泄露，所以對(duì)情報(bào)行加密然后傳給接收者。接收者對(duì)情報(bào)進(jìn)行解密，得到情報(bào)。這里講的編碼器和解碼器是和情報(bào)傳遞很相似?一起查看這篇文章，來揭秘!!!
1一編解碼器

1 1.1 什么叫編解碼器

在網(wǎng)絡(luò)傳輸?shù)倪^程中，數(shù)據(jù)都是以字節(jié)流的方式進(jìn)行傳遞?？蛻舳嗽谙蚍?wù)端發(fā)送數(shù)據(jù)的時(shí)候，將業(yè)務(wù)中其他類型數(shù)據(jù)轉(zhuǎn)化為字節(jié)，叫編碼。服務(wù)端接收到數(shù)據(jù)為字節(jié)流，將字節(jié)流轉(zhuǎn)化為原來的格式，叫解碼。統(tǒng)稱codec。

編解碼器分為兩部分-編碼器和解碼器，編碼器負(fù)責(zé)出站，解碼器負(fù)責(zé)入站。

2 1.2 解碼器

1.2.1 概述

解碼器負(fù)責(zé)入站操作，那么也一定要實(shí)現(xiàn)ChannelInboundHandler接口，所以解碼器本質(zhì) 上也是ChannelHandler。我們自定義編解碼器只需要繼承ByteToMessageDecoder(Netty提供抽象類，繼承 ChannelInboundHandlerAdapter)，實(shí)現(xiàn)decode()。Netty提供一些常用的解碼器實(shí)現(xiàn)，開箱即用。如下：

1 RedisDecoder 基于Redis協(xié)議的解碼器 
2 XmlDecoder 基于XML格式的解碼器 
3 JsonObjectDecoder 基于json數(shù)據(jù)格式的解碼器 
4 HttpObjectDecoder 基于http協(xié)議的解碼器

Netty也提供了MessageToMessageDecoder，將⼀種格式轉(zhuǎn)化為另⼀種格式的解碼器，也提供了⼀些實(shí)現(xiàn)，如下：

1 StringDecoder 將接收到ByteBuf轉(zhuǎn)化為字符串 
2 ByteArrayDecoder 將接收到ByteBuf轉(zhuǎn)化字節(jié)數(shù)組 
3 Base64Decoder 將由ByteBuf或US-ASCII字符串編碼的Base64解碼為ByteBuf。

1.2.2 將字節(jié)流轉(zhuǎn)化為Intger類型(案例)

1. 字節(jié)解碼器

package com.haopt.netty.codec; 
import io.netty.buffer.ByteBuf; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.handler.codec.ByteToMessageDecoder; 
 
import java.util.List; 
public class ByteToIntegerDecoder extends ByteToMessageDecoder { 
    /** 
    * 
    * @param ctx 上下⽂ 
    * @param in 輸⼊的ByteBuf消息數(shù)據(jù) 
    * @param out 轉(zhuǎn)化后輸出的容器 
    * @throws Exception 
    */ 
    @Override 
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
        if(in.readableBytes() >= 4){ //int類型占⽤4個(gè)字節(jié)，所以需要判斷是否存在有4個(gè)字節(jié)，再進(jìn)⾏讀取 
            out.add(in.readInt()); //讀取到int類型數(shù)據(jù)，放⼊到輸出，完成數(shù)據(jù)類型的轉(zhuǎn)化 
        } 
    } 
}

2. Handler

package com.haopt.netty.codec; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.channel.ChannelInboundHandlerAdapter; 
public class ServerHandler extends ChannelInboundHandlerAdapter { 
    @Override 
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
        Integer i = (Integer) msg; //這⾥可以直接拿到Integer類型的數(shù)據(jù) 
        System.out.println("服務(wù)端接收到的消息為：" + i); 
    } 
}

3 在pipeline中添加解碼器

@Override 
protected void initChannel(SocketChannel ch) throws Exception { 
    ch.pipeline() 
    .addLast(new ByteToIntegerDecoder()) 
    .addLast(new ServerHandler()); 
}

可以將代碼復(fù)制到IDEA運(yùn)行下，查看下運(yùn)行效果。

3 1.3 編碼器

1.3.1 概述

將原來的格式轉(zhuǎn)化為字節(jié)。我們要實(shí)現(xiàn)自定義解碼器只要繼承MessageToByteEncoder(實(shí)現(xiàn)了ChannelOutboundHandler接⼝)，本質(zhì)上也是ChannelHandler。Netty中一些實(shí)現(xiàn)的編碼器,如下：

1 ObjectEncoder 將對(duì)象（需要實(shí)現(xiàn)Serializable接⼝）編碼為字節(jié)流 
2 SocksMessageEncoder 將SocksMessage編碼為字節(jié)流 
3 HAProxyMessageEncoder 將HAProxyMessage編碼成字節(jié)流

Netty也提供了MessageToMessageEncoder，將⼀種格式轉(zhuǎn)化為另⼀種格式的編碼器，也提供了⼀些實(shí)現(xiàn)：

1 RedisEncoder 將Redis協(xié)議的對(duì)象進(jìn)⾏編碼 
2 StringEncoder 將字符串進(jìn)⾏編碼操作 
3 Base64Encoder 將Base64字符串進(jìn)⾏編碼操作

1.3.2 將Integer類型編碼為字節(jié)進(jìn)⾏傳遞(案例)

1. 自定義編碼器

package com.haopt.netty.codec.client; 
import io.netty.buffer.ByteBuf; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.handler.codec.MessageToByteEncoder; 
public class IntegerToByteEncoder extends MessageToByteEncoder<Integer> { 
    @Override 
    protected void encode(ChannelHandlerContext ctx, Integer msg, ByteBuf out) throws Exception { 
       out.writeInt(msg); 
    } 
}

2. Handler

package com.haopt.netty.codec.client; 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.channel.SimpleChannelInboundHandler; 
import io.netty.util.CharsetUtil; 
public class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> { 
    @Override 
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { 
        System.out.println("接收到服務(wù)端的消息：" + 
        msg.toString(CharsetUtil.UTF_8)); 
    } 
    @Override 
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
     ctx.writeAndFlush(123); 
    } 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
     cause.printStackTrace(); 
     ctx.close(); 
    } 
}

3. pipeline

@Override 
protected void initChannel(SocketChannel ch) throws Exception { 
    ch.pipeline().addLast(new IntegerToByteEncoder()); 
    ch.pipeline().addLast(new ClientHandler()); 
}

2 二開發(fā)Http服務(wù)器

通過Netty中提供的http的解碼器，進(jìn)行http服務(wù)器開發(fā)。建議代碼復(fù)制下來，執(zhí)行下看看效果。

4 2.1 Netty配置

1. server

package com.haopt.netty.codec.http; 
import io.netty.bootstrap.ServerBootstrap; 
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.NioServerSocketChannel; 
import io.netty.handler.codec.http.HttpObjectAggregator; 
import io.netty.handler.codec.http.HttpRequestDecoder; 
import io.netty.handler.codec.http.HttpResponseEncoder; 
import io.netty.handler.stream.ChunkedWriteHandler; 
public class NettyHttpServer { 
    public static void main(String[] args) throws Exception { 
        // 主線程，不處理任何業(yè)務(wù)邏輯，只是接收客戶的連接請(qǐng)求 
        EventLoopGroup boss = new NioEventLoopGroup(1); 
        // ⼯作線程，線程數(shù)默認(rèn)是：cpu*2 
        EventLoopGroup worker = new NioEventLoopGroup(); 
        try { 
        // 服務(wù)器啟動(dòng)類 
        ServerBootstrap serverBootstrap = new ServerBootstrap(); 
        serverBootstrap.group(boss, worker); 
        //配置server通道 
        serverBootstrap.channel(NioServerSocketChannel.class); 
        serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() { 
            @Override 
            protected void initChannel(SocketChannel ch) throws Exception { 
                ch.pipeline() 
                //http請(qǐng)求的解碼器 
                //將http請(qǐng)求中的uri以及請(qǐng)求體聚合成⼀個(gè)完整的FullHttpRequest對(duì)象 
                .addLast(new HttpRequestDecoder())  
                .addLast(new HttpObjectAggregator(1024 * 128)) 
                .addLast(new HttpResponseEncoder()) //http響應(yīng)的編碼器 
                .addLast(new ChunkedWriteHandler()) //⽀持異步的⼤⽂件傳輸，防⽌內(nèi)存溢出 
                .addLast(new ServerHandler()); 
            } 
          }); //worker線程的處理器 
          ChannelFuture future = serverBootstrap.bind(8080).sync(); 
          System.out.println("服務(wù)器啟動(dòng)完成。。。。。"); 
          //等待服務(wù)端監(jiān)聽端⼝關(guān)閉 
          future.channel().closeFuture().sync(); 
        } finally { 
          //優(yōu)雅關(guān)閉 
          boss.shutdownGracefully(); 
          worker.shutdownGracefully(); 
      } 
     } 
}

2. ServerHandler

package com.haopt.netty.codec.http; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelFutureListener; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.channel.ChannelInboundHandlerAdapter; 
import io.netty.channel.SimpleChannelInboundHandler; 
import io.netty.handler.codec.http.*; 
import io.netty.util.CharsetUtil; 
import java.util.Map; 
public class ServerHandler extends SimpleChannelInboundHandler<FullHttpRequest>{ 
    @Override 
    public void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception { 
        //解析FullHttpRequest，得到請(qǐng)求參數(shù) 
        Map<String, String> paramMap = new RequestParser(request).parse(); 
        String name = paramMap.get("name"); 
        //構(gòu)造響應(yīng)對(duì)象 
        FullHttpResponse httpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); 
        httpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/html;charset=utf-8"); 
        StringBuilder sb = new StringBuilder(); 
        sb.append("<h1>"); 
        sb.append("你好，" + name); 
        sb.append("</h1>"); 
        httpResponse.content().writeBytes(Unpooled.copiedBuffer(sb,CharsetUtil.UTF_8)); 
        //操作完成后，將channel關(guān)閉 
        ctx.writeAndFlush(httpResponse).addListener(ChannelFutureListener.CLOSE);  
    } 
}

3. RequestParser

package com.haopt.netty.codec.http; 
import io.netty.handler.codec.http.FullHttpRequest; 
import io.netty.handler.codec.http.HttpMethod; 
import io.netty.handler.codec.http.QueryStringDecoder; 
import io.netty.handler.codec.http.multipart.Attribute; 
import io.netty.handler.codec.http.multipart.HttpPostRequestDecoder; 
import io.netty.handler.codec.http.multipart.InterfaceHttpData; 
import java.io.IOException; 
import java.util.HashMap; 
import java.util.List; 
import java.util.Map; 
/** 
* HTTP請(qǐng)求參數(shù)解析器, ⽀持GET, POST 
*/ 
public class RequestParser { 
    private FullHttpRequest fullReq; 
    /** 
    * 構(gòu)造⼀個(gè)解析器 
    * @param req 
    */ 
    public RequestParser(FullHttpRequest req) { 
     this.fullReq = req; 
    } 
    /** 
    * 解析請(qǐng)求參數(shù) 
    * @return 包含所有請(qǐng)求參數(shù)的鍵值對(duì), 如果沒有參數(shù), 則返回空Map 
    * 
    * @throws IOException 
    */ 
    public Map<String, String> parse() throws Exception { 
        HttpMethod method = fullReq.method(); 
        Map<String, String> parmMap = new HashMap<>(); 
        if (HttpMethod.GET == method) { 
          // 是GET請(qǐng)求 
          QueryStringDecoder decoder = new QueryStringDecoder(fullReq.uri()); 
          decoder.parameters().entrySet().forEach( entry -> { 
          // entry.getValue()是⼀個(gè)List, 只取第⼀個(gè)元素 
          parmMap.put(entry.getKey(), entry.getValue().get(0)); 
          }); 
        } else if (HttpMethod.POST == method) { 
          // 是POST請(qǐng)求 
          HttpPostRequestDecoder decoder = new 
          HttpPostRequestDecoder(fullReq); 
          decoder.offer(fullReq); 
          List<InterfaceHttpData> parmList = decoder.getBodyHttpDatas(); 
          for (InterfaceHttpData parm : parmList) { 
          Attribute data = (Attribute) parm; 
          parmMap.put(data.getName(), data.getValue()); 
          } 
        } else { 
          // 不⽀持其它⽅法 
          throw new RuntimeException("不⽀持其它⽅法"); // 可以用自定義異常來替代 
        } 
        return parmMap; 
    } 
}

4. 對(duì)象

package com.haopt.netty.codec.obj; 
public class User implements java.io.Serializable { 
    private static final long serialVersionUID = -89217070354741790L; 
    private Long id; 
    private String name; 
    private Integer age; 
    public Long getId() { 
     return id; 
    } 
    public void setId(Long id) { 
     this.id = id; 
    } 
    public String getName() { 
     return name; 
    } 
    public void setName(String name) { 
     this.name = name; 
    } 
    public Integer getAge() { 
     return age; 
    } 
    public void setAge(Integer age) { 
     this.age = age; 
    } 
    @Override 
    public String toString() { 
      return "User{" + 
        "id=" + id + 
        ", name='" + name + '\'' + 
        ", age=" + age + 
        '}'; 
      } 
}

5 2.2 服務(wù)端

1. NettyObjectServer

package com.haopt.netty.codec.obj; 
import io.netty.bootstrap.ServerBootstrap; 
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.NioServerSocketChannel; 
import io.netty.handler.codec.serialization.ClassResolvers; 
import io.netty.handler.codec.serialization.ObjectDecoder; 
public class NettyObjectServer { 
    public static void main(String[] args) throws Exception { 
        // 主線程，不處理任何業(yè)務(wù)邏輯，只是接收客戶的連接請(qǐng)求 
        EventLoopGroup boss = new NioEventLoopGroup(1); 
        // ⼯作線程，線程數(shù)默認(rèn)是：cpu*2 
        EventLoopGroup worker = new NioEventLoopGroup(); 
        try { 
        // 服務(wù)器啟動(dòng)類 
        ServerBootstrap serverBootstrap = new ServerBootstrap(); 
        serverBootstrap.group(boss, worker); 
        //配置server通道 
        serverBootstrap.channel(NioServerSocketChannel.class); 
        serverBootstrap.childHandler(new ChannelInitializer<SocketChannel> () { 
            @Override 
            protected void initChannel(SocketChannel ch) throws Exception { 
                ch.pipeline() 
                .addLast(new ObjectDecoder(ClassResolvers.weakCachingResolver( 
                this.getClass().getClassLoader() 
                ))) 
                .addLast(new ServerHandler()); 
            } 
        }); //worker線程的處理器 
        ChannelFuture future = serverBootstrap.bind(6677).sync(); 
        System.out.println("服務(wù)器啟動(dòng)完成。。。。。"); 
        //等待服務(wù)端監(jiān)聽端⼝關(guān)閉 
        future.channel().closeFuture().sync(); 
        } finally { 
        //優(yōu)雅關(guān)閉 
        boss.shutdownGracefully(); 
        worker.shutdownGracefully(); 
        } 
    } 
}

2. ServerHandler

package com.haopt.netty.codec.obj; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.channel.SimpleChannelInboundHandler; 
import io.netty.util.CharsetUtil; 
public class ServerHandler extends SimpleChannelInboundHandler<User> { 
    @Override 
    public void channelRead0(ChannelHandlerContext ctx, User user) throws Exception { 
        //獲取到user對(duì)象 
        System.out.println(user); 
        ctx.writeAndFlush(Unpooled.copiedBuffer("ok", CharsetUtil.UTF_8)); 
    } 
}

6 2.3 客戶端

1. NettyObjectClient

package com.haopt.netty.codec.obj; 
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 io.netty.handler.codec.serialization.ObjectEncoder; 
public class NettyObjectClient { 
public static void main(String[] args) throws Exception{ 
    EventLoopGroup worker = new NioEventLoopGroup(); 
        try { 
            // 服務(wù)器啟動(dòng)類 
            Bootstrap bootstrap = new Bootstrap(); 
            bootstrap.group(worker); 
            bootstrap.channel(NioSocketChannel.class); 
            bootstrap.handler(new ChannelInitializer<SocketChannel>() { 
                @Override 
                protected void initChannel(SocketChannel ch) throws Exception { 
                    ch.pipeline().addLast(new ObjectEncoder()); 
                    ch.pipeline().addLast(new ClientHandler()); 
                } 
            }); 
            ChannelFuture future = bootstrap.connect("127.0.0.1", 6677).sync(); 
            future.channel().closeFuture().sync(); 
        } finally { 
            worker.shutdownGracefully(); 
        } 
    } 
}

2. ClientHandler

package com.haopt.netty.codec.obj; 
import io.netty.buffer.ByteBuf; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.channel.SimpleChannelInboundHandler; 
import io.netty.util.CharsetUtil; 
public class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> { 
    @Override 
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { 
        System.out.println("接收到服務(wù)端的消息：" + 
        msg.toString(CharsetUtil.UTF_8)); 
    } 
    @Override 
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
        User user = new User(); 
        user.setId(1L); 
        user.setName("張三"); 
        user.setAge(20); 
        ctx.writeAndFlush(user); 
    } 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
        cause.printStackTrace(); 
        ctx.close(); 
    } 
}

7 2.4 JDK序列化的優(yōu)化

JDK序列化使⽤是⽐較⽅便，但是性能較差，序列化后的字節(jié)⽐較⼤，所以⼀般在項(xiàng)⽬中不會(huì)使⽤⾃帶的序列化，⽽是會(huì)采⽤第三⽅的序列化框架Hessian編解碼。

1. 導(dǎo)入依賴

<dependency> 
  <groupId>com.caucho</groupId> 
  <artifactId>hessian</artifactId> 
  <version>4.0.63</version> 
</dependency>

2. User對(duì)象

package com.haopt.netty.codec.hessian; 
public class User implements java.io.Serializable{ 
    private static final long serialVersionUID = -8200798627910162221L; 
    private Long id; 
    private String name; 
    private Integer age; 
    public Long getId() { 
     return id; 
    } 
    public void setId(Long id) { 
     this.id = id; 
    } 
    public String getName() { 
     return name; 
    } 
    public void setName(String name) { 
     this.name = name; 
    } 
    public Integer getAge() { 
     return age; 
    } 
    public void setAge(Integer age) { 
     this.age = age; 
    } 
    @Override 
    public String toString() { 
      return "User{" + 
      "id=" + id + 
      ", name='" + name + '\'' + 
      ", age=" + age + 
      '}'; 
    } 
}

3. Hessian序列化⼯具類

package com.haopt.netty.codec.hessian.codec; 
import com.caucho.hessian.io.HessianInput; 
import com.caucho.hessian.io.HessianOutput; 
import java.io.ByteArrayInputStream; 
import java.io.ByteArrayOutputStream; 
import java.io.IOException; 
/** 
* Hessian序列化⼯具類 
* 
*/ 
public class HessianSerializer { 
    public <T> byte[] serialize(T obj) { 
        ByteArrayOutputStream os = new ByteArrayOutputStream(); 
        HessianOutput ho = new HessianOutput(os); 
        try { 
          ho.writeObject(obj); 
          ho.flush(); 
          return os.toByteArray(); 
        } catch (IOException e) { 
         throw new RuntimeException(e); 
        } finally { 
          try { 
           ho.close(); 
          } catch (IOException e) { 
           throw new RuntimeException(e); 
          } 
          try { 
           os.close(); 
          } catch (IOException e) { 
          throw new RuntimeException(e); 
          } 
        } 
     } 
      
     public <T> Object deserialize(byte[] bytes, Class<T> clazz) { 
        ByteArrayInputStream is = new ByteArrayInputStream(bytes); 
        HessianInput hi = new HessianInput(is); 
        try { 
          return (T) hi.readObject(clazz); 
        } catch (IOException e) { 
          throw new RuntimeException(e); 
        } finally { 
            try { 
               hi.close(); 
            } catch (Exception e) { 
            throw new RuntimeException(e); 
        } 
        try { 
          is.close(); 
        } catch (IOException e) { 
          throw new RuntimeException(e); 
        } 
      } 
    } 
}

4. 編碼器

package com.haopt.netty.codec.hessian.codec; 
import cn.itcast.netty.coder.hessian.User; 
import io.netty.buffer.ByteBuf; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.handler.codec.MessageToByteEncoder; 
public class HessianEncoder extends MessageToByteEncoder<User> { 
    private HessianSerializer hessianSerializer = new HessianSerializer(); 
    protected void encode(ChannelHandlerContext ctx, User msg, ByteBuf out) throws Exception { 
          byte[] bytes = hessianSerializer.serialize(msg); 
          out.writeBytes(bytes); 
    } 
}

5. 解碼器

public class HessianDecoder extends ByteToMessageDecoder { 
    private HessianSerializer hessianSerializer = new HessianSerializer(); 
 
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> 
            out) throws Exception { 
        //復(fù)制⼀份ByteBuf數(shù)據(jù)，輕復(fù)制，⾮完全拷⻉ 
        //避免出現(xiàn)異常：did not read anything but decoded a message 
        //Netty檢測(cè)沒有讀取任何字節(jié)就會(huì)拋出該異常 
        ByteBuf in2 = in.retainedDuplicate(); 
        byte[] dst; 
        if (in2.hasArray()) {//堆緩沖區(qū)模式 
            dst = in2.array(); 
        } else { 
            dst = new byte[in2.readableBytes()]; 
            in2.getBytes(in2.readerIndex(), dst); 
        } 
        //跳過所有的字節(jié)，表示已經(jīng)讀取過了 
        in.skipBytes(in.readableBytes()); 
        //反序列化 
        Object obj = hessianSerializer.deserialize(dst, User.class); 
        out.add(obj); 
    } 
}

6. 服務(wù)端

public class NettyHessianServer { 
    public static void main(String[] args) throws Exception { 
        // System.setProperty("io.netty.noUnsafe", "true"); 
        // 主線程，不處理任何業(yè)務(wù)邏輯，只是接收客戶的連接請(qǐng)求 
        EventLoopGroup boss = new NioEventLoopGroup(1); 
        // ⼯作線程，線程數(shù)默認(rèn)是：cpu*2 
        EventLoopGroup worker = new NioEventLoopGroup(); 
        try { 
            // 服務(wù)器啟動(dòng)類 
            ServerBootstrap serverBootstrap = new ServerBootstrap(); 
            serverBootstrap.group(boss, worker); 
            //配置server通道 
            serverBootstrap.channel(NioServerSocketChannel.class); 
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel> 
                    () { 
                @Override 
                protected void initChannel(SocketChannel ch) throws Exception { 
                    ch.pipeline() 
                            .addLast(new HessianDecoder()) 
                            .addLast(new ServerHandler()); 
                } 
            }); //worker線程的處理器 
            // serverBootstrap.childOption(ChannelOption.ALLOCATOR, 
            UnpooledByteBufAllocator.DEFAULT); 
            ChannelFuture future = serverBootstrap.bind(6677).sync(); 
            System.out.println("服務(wù)器啟動(dòng)完成。。。。。"); 
            //等待服務(wù)端監(jiān)聽端⼝關(guān)閉 
            future.channel().closeFuture().sync(); 
        } finally { 
            //優(yōu)雅關(guān)閉 
            boss.shutdownGracefully(); 
            worker.shutdownGracefully(); 
        } 
    } 
}

public class ServerHandler extends SimpleChannelInboundHandler<User> { 
    @Override 
    public void channelRead0(ChannelHandlerContext ctx, User user) throws 
            Exception { 
        //獲取到user對(duì)象 
        System.out.println(user); 
        ctx.writeAndFlush(Unpooled.copiedBuffer("ok", CharsetUtil.UTF_8)); 
    } 
}

7. 客戶端(配置類)

public class NettyHessianClient { 
    public static void main(String[] args) throws Exception { 
        EventLoopGroup worker = new NioEventLoopGroup(); 
        try { 
            // 服務(wù)器啟動(dòng)類 
            Bootstrap bootstrap = new Bootstrap(); 
            bootstrap.group(worker); 
            bootstrap.channel(NioSocketChannel.class); 
            bootstrap.handler(new ChannelInitializer<SocketChannel>() { 
                @Override 
                protected void initChannel(SocketChannel ch) throws Exception { 
                    ch.pipeline().addLast(new HessianEncoder()); 
                    ch.pipeline().addLast(new ClientHandler()); 
                } 
            }); 
            ChannelFuture future = bootstrap.connect("127.0.0.1", 6677).sync(); 
            future.channel().closeFuture().sync(); 
        } finally { 
            worker.shutdownGracefully(); 
        } 
    } 
}

public class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> { 
    @Override 
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws 
            Exception { 
        System.out.println("接收到服務(wù)端的消息：" + 
                msg.toString(CharsetUtil.UTF_8)); 
    } 
 
    @Override 
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
        User user = new User(); 
        user.setId(1L); 
        user.setName("張三"); 
        user.setAge(20); 
        ctx.writeAndFlush(user); 
    } 
 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) 
            throws Exception { 
        cause.printStackTrace(); 
        ctx.close(); 
    } 
}

這篇文章，介紹了什么是編碼器、解碼器，也講述了如何實(shí)戰(zhàn)中運(yùn)用編碼器和解碼器。希望能對(duì)有所幫助。在開頭提到的我們本文的編碼器解碼器和情報(bào)信息交互是否相似?在我看來，是相似的。發(fā)報(bào)人將自己看的懂得信息，按照某種規(guī)則進(jìn)行加密。收?qǐng)?bào)人接收到信息是加密后的數(shù)據(jù)，需要進(jìn)行按照規(guī)則進(jìn)行解密才能看懂。我們客戶端在進(jìn)行發(fā)送數(shù)據(jù)，需要將程序中的數(shù)據(jù)變?yōu)槎M(jìn)制流發(fā)送。服務(wù)端接收到數(shù)據(jù)，需要將二進(jìn)制流轉(zhuǎn)化為程序可以操作數(shù)據(jù)類型。

責(zé)任編輯：姜華來源：花花和Java

Netty 編碼器解碼器

自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

介紹下Netty中常用的編碼器和解碼器