当前位置：首页 > 编程语言 > 网络编程 -- RPC实现原理 -- Netty -- 迭代版本V3 -- 编码解码

网络编程 -- RPC实现原理 -- Netty -- 迭代版本V3 -- 编码解码

2021-07-14 23:06

阅读：851

标签：style   pool   flush   worker   .net   sign   自动   lib   sage   

　网络编程 -- RPC实现原理 -- 目录

　　啦啦啦

V2——Netty -- pipeline.addLast(io.netty.handler.codec.MessageToMessageCodec) 覆写编码解码方法。

　pipeline相当于拦截器。在pipeline中添加MessageToMessageCodec接口的实现类，该接口的实现类中的encode()方法自动将发送的Object对象转换为ByteBuf，decode()方法自动将接收的ByteBuf对象转换为Object

　

　　Class ： Server

package lime.pri.limeNio.netty.netty03;

import java.net.InetSocketAddress;
import java.util.Date;
import java.util.List;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
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.MessageToMessageCodec;
import io.netty.util.CharsetUtil;
import lime.pri.limeNio.netty.netty03.entity.User;

public class Server {

    public static void main(String[] args) throws Exception {
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();
        serverBootstrap.group(boss, worker);
        serverBootstrap.channel(NioServerSocketChannel.class);
        serverBootstrap.childHandler(new ChannelInitializer() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new MessageToMessageCodec() {
                    @Override
                    protected void encode(ChannelHandlerContext ctx, Object msg, List out) throws Exception {
                        System.out.println("-- -- 服务端编码");
                        out.add(Unpooled.buffer().writeBytes(JSON.toJSONString(msg,SerializerFeature.WriteClassName).getBytes(CharsetUtil.UTF_8)));
                    }

                    @Override
                    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List out) throws Exception {
                        System.out.println("-- -- 服务端解码");
                        out.add(JSON.parse(msg.toString(CharsetUtil.UTF_8)));
                    }
                }).addLast(new ChannelHandlerAdapter() {

                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        System.out.println("客户端请求数据：" + msg);
                        String request = (String) msg;
                        Object response = "请求参数不正确";
                        if("Query Date".equalsIgnoreCase(request)){
                            response = "当前系统时间：" + new Date().toString();
                        }else if("Query User".equalsIgnoreCase(request)){
                            response = new User(1,"lime",new Date());
                        }
                        ChannelFuture channelFuture = ctx.writeAndFlush(response);
                        channelFuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
                        channelFuture.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                        channelFuture.addListener(ChannelFutureListener.CLOSE);
                    }
                    @Override
                    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                        cause.printStackTrace();
                    }

                });
            }

        });
        ChannelFuture channelFuture = serverBootstrap.bind(new InetSocketAddress(9999)).sync();
        channelFuture.channel().closeFuture().sync();
        boss.close();
        worker.close();
    }
}

　　Class ： Client

package lime.pri.limeNio.netty.netty03;

import java.net.InetSocketAddress;
import java.util.List;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
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.MessageToMessageCodec;
import io.netty.util.CharsetUtil;

public class Client {

    public static void main(String[] args) throws Exception {
        for (int i = 0; i ) {
            new Thread() {
                {
                    setDaemon(false);
                }

                public void run() {
                    try {
                        client();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                };
            }.start();
            Thread.sleep(1000);
        }
    }

    private static void client() throws Exception {
        Bootstrap bootstrap = new Bootstrap();
        EventLoopGroup worker = new NioEventLoopGroup();
        bootstrap.group(worker);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new MessageToMessageCodec() {

                    @Override
                    protected void encode(ChannelHandlerContext ctx, Object msg, List out) throws Exception {
                        System.out.println("-- -- 客户端编码");
                        out.add(Unpooled.buffer().writeBytes(JSON.toJSONString(msg, SerializerFeature.WriteClassName).getBytes(CharsetUtil.UTF_8)));
                    }

                    @Override
                    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List out) throws Exception {
                        System.out.println("-- -- 客户端解码");
                        out.add(JSON.parse(msg.toString(CharsetUtil.UTF_8)));
                    }
                }).addLast(new ChannelHandlerAdapter() {
                    /**
                     * 默认只捕获网络连接异常
                     */
                    @Override
                    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                        System.out.println(cause);
                    }

                    /**
                     * 客户端发送经过JSON编码的byteBuf
                     */
                    @Override
                    public void channelActive(ChannelHandlerContext ctx) throws Exception {
                        String request = null;
                        switch ((int) (Math.random() * 10) % 3) {
                        case 0:
                            request = "Query Date";
                            break;
                        case 1:
                            request = "Query User";
                            break;

                        default:
                            request = "Query What?";
                            break;
                        }
                        ChannelFuture channelFuture = ctx.writeAndFlush(request);
                        channelFuture.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);
                        channelFuture.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                    }

                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        System.out.println("服务端响应数据 --> " + msg);
                    }

                });
            }
        });
        ChannelFuture channelFuture;

        channelFuture = bootstrap.connect(new InetSocketAddress(9999)).sync();
        channelFuture.channel().closeFuture().sync();
        worker.close();
    }
}

啦啦啦

网络编程 -- RPC实现原理 -- Netty -- 迭代版本V3 -- 编码解码

标签：style   pool   flush   worker   .net   sign   自动   lib   sage   

原文地址：http://www.cnblogs.com/ClassNotFoundException/p/7074482.html

上一篇：Web Service学习-CXF开发Web Service实例demo（一）

下一篇：dhtmlxTree 加载大数据量树

文章来自：搜素材网的编程语言模块，转载请注明文章出处。
文章标题：网络编程 -- RPC实现原理 -- Netty -- 迭代版本V3 -- 编码解码
文章链接：http://soscw.com/essay/105312.html

评论

亲，登录后才可以留言！

热门文章

推荐文章

最新文章

置顶文章

关于我们 | 版权声明 | 常见问题 | 素材投稿 | 联系我们 | 网站地图 |
搜素材网素材除本站原创外均由用户分享，若发现权利被侵害，请联系及时联系我们，我们会在第一时间进行处理。
特别说明：本站所有资源除本站原创外仅供学习与参考，请勿用于商业用途,如有侵犯您的版权请联系客服服务QQ：点击这里给我发消息
Copyright © 2025 soscw.com 搜素材网素材网版权所有 蜀ICP备18015633号-1