Netty實踐：ChannelInboundHandlerAdapter與ChannelOutboundHandlerAdapter

本文轉載自查看原文 2017-06-27 23:20 9876

package netty.test1;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.jboss.netty.channel.ChannelFutureListener;

import java.util.concurrent.ScheduledExecutorService;


/**
 * Created by Administrator on 2017/5/17.
 */
@ChannelHandler.Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {


    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("注冊事件");
    }

    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("取消注冊事件");
    }

    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("有新客戶端連接接入。。。"+ctx.channel().remoteAddress());
    }

    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("失去連接");
    }

    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf in = (ByteBuf) msg;
        System.out.println("讀客戶端傳入數據="+in.toString(CharsetUtil.UTF_8));
        ctx.writeAndFlush(Unpooled.copiedBuffer("channelRead Netty rocks!", CharsetUtil.UTF_8));
        //ctx.fireChannelActive();
    }

    public void channelReadComplete(ChannelHandlerContext ctx){
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (future.isSuccess()) {
                    System.out.println("執行成功="+future.isSuccess());
                }
            }
        });
        ctx.writeAndFlush(Unpooled.copiedBuffer("channelReadComplete Netty rocks!", CharsetUtil.UTF_8)).addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (future.isSuccess())  {

                }else {

                }
            }
        });
    }

    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        System.out.println("userEventTriggered");
    }

    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channelWritabilityChanged");
    }

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

package netty.test1;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.ReferenceCounted;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;


/**
 * Created by Administrator on 2017/5/17.
 */
@ChannelHandler.Sharable
public class EchoOtherServerHandler extends ChannelInboundHandlerAdapter {


    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Other注冊事件");
        ctx.fireChannelRegistered();
    }

    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Other取消注冊事件");
        ctx.fireChannelUnregistered();
    }

    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Other有新客戶端連接接入。。。"+ctx.channel().remoteAddress());
        ctx.fireChannelActive();
    }

    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Other失去連接");
        ctx.fireChannelInactive();
    }

    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf in = (ByteBuf) msg;
        System.out.println("Other讀客戶端傳入數據="+in.toString(CharsetUtil.UTF_8));
        final ByteBuf byteBuf = Unpooled.copiedBuffer("Other channelRead Netty rocks!", CharsetUtil.UTF_8);
        ctx.writeAndFlush(byteBuf);
        ctx.fireChannelRead(msg);
        //ReferenceCountUtil.release(msg);
    }

    public void channelReadComplete(ChannelHandlerContext ctx){
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (future.isSuccess()) {
                    System.out.println("Other執行成功="+future.isSuccess());
                }
            }
        });
        final ByteBuf byteBuf = Unpooled.copiedBuffer("Other channelReadComplete Netty rocks!", CharsetUtil.UTF_8);
        ctx.writeAndFlush(byteBuf).addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                if (future.isSuccess())  {
                }else {

                }
                ReferenceCountUtil.release(byteBuf);
            }
        });
        ctx.fireChannelReadComplete();
    }

    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        System.out.println("Other  userEventTriggered");
    }

    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Other  channelWritabilityChanged");
    }

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

package netty.test1;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandler;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;

import java.nio.charset.Charset;

/**
 * Created by Administrator on 2017/6/8.
 */
public class EchoServerOutHandler extends ChannelOutboundHandlerAdapter {

    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println("EchoServerOutHandler   write "+((ByteBuf)msg).toString(Charset.defaultCharset()));
        ctx.write(msg, promise);
    }
}

package netty.test1;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelFuture;
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 java.net.InetSocketAddress;

/**
 * Created by Administrator on 2017/5/17.
 */
public class EchoServer {

    private final  int port;

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

    public static void main(String[] args) throws InterruptedException {
        int port = 8080;
        ByteBufAllocator allocator;

        new EchoServer(port).start();
    }

    public void start() throws InterruptedException {
        final EchoServerHandler echoServerHandler = new EchoServerHandler();
        final EchoOtherServerHandler echoOtherServerHandler = new EchoOtherServerHandler();
        final EchoServerOutHandler echoServerOutHandler = new EchoServerOutHandler();
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(eventLoopGroup).
                    channel(NioServerSocketChannel.class).//指定channel使用Nio傳輸
                    localAddress(new InetSocketAddress(port)).//執行端口設置套接字地址
                    childHandler(new ChannelInitializer<SocketChannel>() {//添加echoServerHandler到Channel的channelpipeline上
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline channelPipeline = socketChannel.pipeline();
                            channelPipeline.addFirst(echoOtherServerHandler);
                            channelPipeline.addFirst(echoServerOutHandler);
                            channelPipeline.addLast(echoServerHandler);
                        }
                    });
            ChannelFuture f = serverBootstrap.bind().sync();//異步綁定服務器，調用sync()方法阻塞等待直到綁定完成
            f.channel().closeFuture().sync();//獲得Channel的closefutrue，並且阻塞當前線程直到它完成
        } catch (InterruptedException e) {
            eventLoopGroup.shutdownGracefully().sync();
        }
    }
}

package netty.test1;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.jboss.netty.channel.ChannelFutureListener;


/**
 * Created by Administrator on 2017/5/17.
 */
@ChannelHandler.Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8));
    }

    public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        ByteBuf in = msg;
        System.out.println("讀取服務端channelRead0="+in.toString(CharsetUtil.UTF_8));
    }

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

package netty.test1;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
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.oio.OioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.channel.socket.oio.OioSocketChannel;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;

/**
 * Created by Administrator on 2017/5/17.
 */
public class EchoClient {

    private final  int port;

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

    public static void main(String[] argsw) {
        ByteBuffer byteBuffer;
        String host = "127.0.0.1";
        int port = 8080;
        try {
            new EchoClient(8080).start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void start() throws Exception {
        final EchoClientHandler clientHandler = new EchoClientHandler();
        EventLoopGroup eventLoopGroup = new OioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(eventLoopGroup).
                    channel(OioSocketChannel.class).
                    remoteAddress(new InetSocketAddress("localhost",port)).
                    handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(clientHandler);
                        }
                    });
            //異步連接遠程服務，連接遠程服務成功后，輸出"已經連接到服務器！"
            final ChannelFuture f = b.connect();
            f.addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    if (future.isSuccess()) {
                        System.out.println("已經連接到服務器！");
                        ByteBuf byteBuf = Unpooled.copiedBuffer("創建ByteBuf", Charset.defaultCharset());
                        ChannelFuture channelFuture = f.channel().writeAndFlush(byteBuf);
                    }else {
                        Throwable throwable = future.cause();
                        throwable.printStackTrace();
                    }
                }
            });
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            eventLoopGroup.shutdownGracefully().sync();
        }
    }
}

服務端日志：

Other注冊事件
注冊事件
Other有新客戶端連接接入。。。/127.0.0.1:58678
有新客戶端連接接入。。。/127.0.0.1:58678
Other讀客戶端傳入數據=創建ByteBufNetty rocks!
EchoServerOutHandler   write Other channelRead Netty rocks!
讀客戶端傳入數據=創建ByteBufNetty rocks!
EchoServerOutHandler   write channelRead Netty rocks!
EchoServerOutHandler   write
Other執行成功=true
EchoServerOutHandler   write Other channelReadComplete Netty rocks!
EchoServerOutHandler   write
執行成功=true
EchoServerOutHandler   write channelReadComplete Netty rocks!

------------------------------------------------------------------------------------------------

客戶端日志：

已經連接到服務器！
讀取服務端channelRead0=Other channelRead Netty rocks!
讀取服務端channelRead0=channelRead Netty rocks!
讀取服務端channelRead0=Other channelReadComplete Netty rocks!
讀取服務端channelRead0=channelReadComplete Netty rocks!

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 Netty實踐與NIO原理 SimpleChannelInboundHandler與ChannelInboundHandlerAdapter netty廣播（三）ChannelGroup實踐 Netty 3升級Netty4實踐 Google Protobuf結合Netty實踐 SimpleChannelInboundHandler和ChannelInboundHandlerAdapter區別 ChannelInboundHandlerAdapter和SimpleChannelInboundHandler區別 Netty基礎招式——ChannelHandler的最佳實踐 netty搭建Tcp服務器實踐 netty5 HTTP協議棧淺析與實踐