客户端:
public static void main(String[] args) throws Exception { final SslContext sslCtx; if (SSL) { sslCtx = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build(); } else { sslCtx = null; } EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline p = ch.pipeline(); if (sslCtx != null) { p.addLast(sslCtx.newHandler(ch.alloc(),HOST,PORT)); } p.addLast(new DiscardClientHandler()); } }); //make the connection attempt. ChannelFuture f = b.connect(HOST,PORT).sync(); //wait until the connection is closed. f.channel().closeFuture().sync(); log.info("connection is closed"); } finally { group.shutdownGracefully(); } }
或者
public static void main(String[] args) throws Exception { EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .handler(new TelnetClientInitializer()); Channel ch = b.connect(HOST, PORT).channel(); //read commands from the stdbin. ChannelFuture lastWriteFuture = null; BufferedReader in = new BufferedReader(new InputStreamReader(System.in)); for(;;) { String line = in.readLine(); if (line == null) { break; } //send the received line to the server lastWriteFuture = ch.writeAndFlush(line +" "); //if user typed the 'bye' command,wait unitl the server closes the connection. if ("bye".equals(line.toLowerCase())) { ch.closeFuture().sync(); break; } } //wait unitl all messages are flushed before closing the channel. if (lastWriteFuture != null) { lastWriteFuture.sync(); } } finally { group.shutdownGracefully(); } }
或者
@Override protected void channelRead0(ChannelHandlerContext ctx, BigInteger msg) throws Exception { receivedMessages ++; if (receivedMessages == FactorialClient.COUNT-start+1) { //offer(放入)the answer after closing the connection ctx.channel().close().addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { boolean offered = answer.offer(msg); log.info("offer the answer result:{}",offered); } }); } }
服务端:
1、继承SimpleChannelInboundHandler或ChannelInboundHandlerAdapter的server端
@Override protected void channelRead0(ChannelHandlerContext ctx, String request) throws Exception { //Generate and write a response String response; boolean close = false; if (request.isEmpty()) { response = "Please type something. "; } else if("bye".equals(request.toLowerCase())) { response = "Have a good day! "; close = true; } else { response = "Did you say '"+request+"'? "; } //不需要write ByteBuf,只需write string,因为传递给StringEncoder ChannelFuture future = ctx.write(response); if (close) { future.addListener(ChannelFutureListener.CLOSE); } }
如果是短链接,必须在服务端关闭该channel。此时,才能通知到客户端的chanel.future.close()方法。
2、需要解码器(decoder)的server端
在serverHandler类中的channelRead方法中,无需加入future.addListener(ChannelFutureListener.CLOSE);如下,
@Override protected void channelRead0(ChannelHandlerContext ctx, BigInteger msg) throws Exception { //计算阶乘并发送到客户端 lastMultiplier = msg; factorial = factorial.multiply(msg); ctx.writeAndFlush(factorial); }
因为,ByteToMessageDecoder类中,已执行了关闭,如下
@Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { channelInputClosed(ctx, true); }