netty简单介绍

 

ServerBootstrap

 

启动netty只需要三行代码：

package com.duitang.test;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.socket.oio.OioServerSocketChannelFactory;

/**
 * StringDecoder tester
 * @author yunpeng
 *
 */
public class StringDecoderTester {

    public static void main(String[] args) {
        ChannelFactory channelFactory = new OioServerSocketChannelFactory(Executors.newCachedThreadPool(),Executors.newCachedThreadPool());
        ServerBootstrap bootStrap = new ServerBootstrap(channelFactory);
        bootStrap.bind(new InetSocketAddress(8888));
    }
}

 

 

  ServerBootstrap是入口，复制启动netty。具体实现是委托给了ChannelFactory。这里是使用的OioServerSocketChannelFactory，OioServerSocketChannelFactory需要两个参数，一个是BossExcutor,一个WorkExcutor。BossExcutor负责不停的accept()到新的socket，然后提交给workExcutor，workExcutor负责监听socket的InputStream，当发现socket的InputStrea中有新的数据写入时，读取数据，然后通知给netty的消息框架进行处理。流程如下：

1. OioWorker.run()   负责监听一个socket的InputStream
2. Channels.fireMessageReceived()  发送消息  
3. channel.getPipeline().sendUpstream(); 通知给pipline

 

 

自定义Hander

因为没有hander，所以日志里面会有警告：

 

警告: The pipeline contains no upstream handlers;

所以我们增加一个Hander：

 

    static class StringTesterServerHandler extends SimpleChannelHandler {

        public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
            ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
            String msg = buffer.toString("utf-8");
            System.out.println(msg);
            buffer = ChannelBuffers.copiedBuffer("bye! \r\n", "utf-8");
            ctx.getChannel().write(buffer);
            super.messageReceived(ctx, e);
        }

    }

 

    public static void main(String[] args) {
        ChannelFactory channelFactory = new OioServerSocketChannelFactory(Executors.newCachedThreadPool(),
                                                                          Executors.newCachedThreadPool());
        ServerBootstrap bootStrap = new ServerBootstrap(channelFactory);
        ChannelPipeline pipeline = bootStrap.getPipeline();
        pipeline.addLast("handler", new StringTesterServerHandler());
        bootStrap.bind(new InetSocketAddress(8888));
    }

 

echo 'fcuk' | nc localhost 8888

 

 

 

理解UpStream和DownStream

UpStream = InpuStream

DownStream = OutStream

 

 

SimpleChannelHandler

 

如果直接implements ChannelUpstreamHandler接口,实现其handleUpstream方法是不可行的，因为handleUpstream会被各种情况调用，需要判断各种Channel事件，SimpleChannelHandler提供了完整的实现，包括异常处理，关闭网络连接等：

 public void handleUpstream(
            ChannelHandlerContext ctx, ChannelEvent e) throws Exception {

        if (e instanceof MessageEvent) {
            messageReceived(ctx, (MessageEvent) e);
        } else if (e instanceof WriteCompletionEvent) {
            WriteCompletionEvent evt = (WriteCompletionEvent) e;
            writeComplete(ctx, evt);
        } else if (e instanceof ChildChannelStateEvent) {
            ChildChannelStateEvent evt = (ChildChannelStateEvent) e;
            if (evt.getChildChannel().isOpen()) {
                childChannelOpen(ctx, evt);
            } else {
                childChannelClosed(ctx, evt);
            }
        } else if (e instanceof ChannelStateEvent) {
            ChannelStateEvent evt = (ChannelStateEvent) e;
            switch (evt.getState()) {
            case OPEN:
                if (Boolean.TRUE.equals(evt.getValue())) {
                    channelOpen(ctx, evt);
                } else {
                    channelClosed(ctx, evt);
                }
                break;
            case BOUND:
                if (evt.getValue() != null) {
                    channelBound(ctx, evt);
                } else {
                    channelUnbound(ctx, evt);
                }
                break;
            case CONNECTED:
                if (evt.getValue() != null) {
                    channelConnected(ctx, evt);
                } else {
                    channelDisconnected(ctx, evt);
                }
                break;
            case INTEREST_OPS:
                channelInterestChanged(ctx, evt);
                break;
            default:
                ctx.sendUpstream(e);
            }
        } else if (e instanceof ExceptionEvent) {
            exceptionCaught(ctx, (ExceptionEvent) e);
        } else {
            ctx.sendUpstream(e);
        }
    }

查看上面代码发现，只有ChannelEvent属于MessageEvent时才是我们真正需要关心的，所以只要重写messageReceived方法既可。

        public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
            ChannelBuffer buffer = ChannelBuffers.copiedBuffer("hello", "utf-8");
            ctx.getChannel().write(buffer);
            super.messageReceived(ctx, e);
        }

 

注意，这里需要super.messageReceived(), 因为上层调用了 ctx.sendUpstream(e);查看DefaultChannelHandlerContext.sendUpstream()发现会通知下一个的hander：

        public void sendUpstream(ChannelEvent e) {
            DefaultChannelHandlerContext next = getActualUpstreamContext(this.next);
            if (next != null) {
                DefaultChannelPipeline.this.sendUpstream(next, e);
            }
        }

 

报文frame

 

报文frame的问题，即由于这种状态机模型，你的某个channel每收到一段buffer，虽然顺序是保证的，但是完整性就不一定了。就好像我 们自己写传统socket要循环读一样，你这里同样要循环读，读到完整可解析的一整个frame方能解析，这个问题应该也不是netty独有的，除非你以 前写socket应用都是侥幸没有被网卡被操作系统被网关路由器之类的切过包。。。

netty有dynamicBuffer，就是用来聚集frame的，当然你也可以通过它提供的decoder来保证frame。