netty 学习记录二

netty 最新版本是netty-5.0.0.Alpha1,去年10月份发布的,至今没有发新版本,估计这个版本还是比较稳定. 整包下载,里面包含一个 netty-example-5.0.0.Alpha1-sources.jar文件,提供了比较丰富的example例子,多看几遍还是非常有收获的,这里记录下.

先来看下channelHandler的两个不同继承:

方式一:直接从ChannelHandlerAdapter类里继承,读取操作从channelRead方法里执行

@Sharable
public class EchoServerHandler extends ChannelHandlerAdapter {

    private static final Logger logger = Logger.getLogger(
            EchoServerHandler.class.getName());

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.write(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        logger.log(Level.WARNING, "Unexpected exception from downstream.", cause);
        ctx.close();
    }
}

方式二:继承至SimpleChannelInboundHandler类,读取操作从方法messageReceived()里执行

public class FactorialServerHandler extends
		SimpleChannelInboundHandler<BigInteger> {

	private static final Logger logger = Logger
			.getLogger(FactorialServerHandler.class.getName());

	private BigInteger lastMultiplier = new BigInteger("1");
	private BigInteger factorial = new BigInteger("1");

	@Override
	public void messageReceived(ChannelHandlerContext ctx, BigInteger msg)
			throws Exception {
		// Calculate the cumulative factorial and send it to the client.
		System.out.println("server msg:" + msg);
		lastMultiplier = msg;
		factorial = factorial.multiply(msg);
		ctx.writeAndFlush(factorial);
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		Formatter fmt = new Formatter();
		logger.info(fmt.format("Factorial of %,d is: %,d", lastMultiplier,
				factorial).toString());
		fmt.close();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		logger.log(Level.WARNING, "Unexpected exception from downstream.",
				cause);
		ctx.close();
	}
}

区别可以从SimpleChannelInboundHandler类的说明文字看出,

SimpleChannelInboundHandler : ChannelHandler which allows to explicit only handle a specific type of messages. 也就是可以操作指定类型的信息.

接着看下 io.netty.example.factorial  这个包, 内容主要描述客户端向服务端发数字,服务端返回数字的阶乘给客户端,业务比较简单.

编解码操作涉及到的类,BigIntegerDecoder与NumberEncoder都是自定义的

pipeline.addLast("decoder", new BigIntegerDecoder());

pipeline.addLast("encoder", new NumberEncoder());

客户端有这么一段代码:

ChannelFuture future = null;
		for (int i = 0; i < 4096 && next <= count; i++) {
			future = ctx.write(Integer.valueOf(next));
			next++;
		}
		if (next <= count) {
			assert future != null;
			future.addListener(numberSender);
		}
		ctx.flush();

服务端代码:

@Override
	public void messageReceived(ChannelHandlerContext ctx, BigInteger msg)
			throws Exception {
		// Calculate the cumulative factorial and send it to the client.
		System.out.println("server msg:" + msg);
		lastMultiplier = msg;
		factorial = factorial.multiply(msg);
		ctx.writeAndFlush(factorial);
	}

看了客户端代码觉得代码在最后才flush的,所以服务端应该只收到一条消息,messageReceived方法应该只调用一次,结果是messageReceived方法调用次数与客户端发送次数一致. 有点奇怪. 看了下BigIntegerDecoder解码类操作才发现服务端的确只收到了一条消息,但是解码器解码成了多个对象(BigIntegerDecoder 类的decode方法被其父类ByteToMessageDecoder的callDecode方法反复读取同一条消息的ByteBuf,直到读完.),最后多次调用messageReceived方法.

以后如果自己写编解码类完全可以参考BigIntegerDecoder与NumberEncoder这两个类来.

最后来看下http的几个ChannelHandler

服务端:

 pipeline.addLast("decoder", new HttpRequestDecoder());
 pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
 pipeline.addLast("encoder", new HttpResponseEncoder());
 pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());

客户端:

p.addLast("codec", new HttpClientCodec());
p.addLast("aggregator", new HttpObjectAggregator(1048576));

刚看到时候觉得奇怪怎么服务端与客户端编解码方式不一样, 看下httpClientCoder类的说明(A combination of HttpRequestEncoder and HttpResponseDecoder which enables easier client side HTTP implementation),自然就懂了.

HttpObjectAggregator 了解这个handler先看段代码:

@Override
    protected void messageReceived(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof HttpRequest) {

        }

        if (msg instanceof HttpContent) {

        }
    }

一般http请求或者响应,解码器都将其解码成为多个消息对象,主要是httpRequest/httpResponse, httpcontent, lastHttpContent.然后反复调用messageReceive这个方法,

HttpObjectAggregator 这个handler就是将同一个http请求或响应的多个消息对象变成一个 fullHttpRequest完整的消息对象.