大家好，又见面了，我是你们的朋友全栈君。
1、Bio工程结构 maven工程文件结构： │ nettyArticle.iml │ pom.xml │ ├─.idea │ compiler.xml │ misc.xml │ vcs.xml │ workspace.xml │ ├─src │ ├─main │ │ ├─java │ │ │ └─com │ │ │ └─jad │ │ │ └─nettyArticle │ │ │ ├─aio │ │ │ │ AcceptCompletionHandler.java │ │ │ │ AioTimeClient.java │ │ │ │ AioTimeServer.java │ │ │ │ AsyncTimeClientHandler.java │ │ │ │ AysncTimeServerHandler.java │ │ │ │ ReadCompletionHandler.java │ │ │ │ │ │ │ ├─bio │ │ │ │ TimeClient.java │ │ │ │ TimeServer.java │ │ │ │ TimeServerHandler.java │ │ │ │ │ │ │ ├─fakeNio │ │ │ │ FakeNioTimeServer.java │ │ │ │ TimeServerHandlerExecutePool.java │ │ │ │ │ │ │ ├─netty │ │ │ │ NettyTimeClient.java │ │ │ │ NettyTimeClientHandler.java │ │ │ │ NettyTimeServer.java │ │ │ │ NettyTimeServerHandler.java │ │ │ │ │ │ │ └─nio │ │ │ MultiplexerTimeServer.java │ │ │ NioTimeClient.java │ │ │ NioTimeServer.java │ │ │ TimeClientHandler.java │ │ │ │ │ └─resources │ └─test │ └─java └─target ├─classes │ └─com │ └─jad │ └─nettyArticle │ ├─aio │ │ AcceptCompletionHandler.class │ │ AioTimeClient.class │ │ AioTimeServer.class │ │ AsyncTimeClientHandler11.class │ │ AsyncTimeClientHandler1.class │ │ AsyncTimeClientHandler.class │ │ AysncTimeServerHandler.class │ │ ReadCompletionHandler1.class │ │ ReadCompletionHandler.class │ │ │ ├─bio │ │ TimeClient.class │ │ TimeServer.class │ │ TimeServerHandler.class │ │ │ ├─fakeNio │ │ FakeNioTimeServer.class │ │ TimeServerHandlerExecutePool.class │ │ │ ├─netty │ │ NettyTimeClient
2、代码： pom.xml：
代码语言：javascript复制<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.jad</groupId>
<artifactId>nettyArticle</artifactId>
<version>1.0-SNAPSHOT</version>
</project>TimeServer：
代码语言：javascript复制package com.jad.nettyArticle.bio;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
public class TimeServer {
public static void main(String[] args) throws IOException {
int port=8080;
if(args!=null && args.length>0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
ServerSocket server =null;
try {
server=new ServerSocket(port);
System.out.println("The time server is start in port :" port);
Socket socket=null;
while (true) {
socket = server.accept();
new Thread(new TimeServerHandler(socket)).start();
}
} finally {
if (server != null) {
System.out.println("The time server close");
server.close();
server=null;
}
}
}
}TimeServerHandler：
代码语言：javascript复制package com.jad.nettyArticle.bio;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
public class TimeServerHandler implements Runnable {
private Socket socket;
public TimeServerHandler(Socket socket) {
this.socket=socket;
}
public void run() {
BufferedReader in = null;
PrintWriter out = null;
try {
in = new BufferedReader(new InputStreamReader(this.socket.getInputStream()));
out= new PrintWriter(this.socket.getOutputStream(),true);
String currentTime = null;
String body = null;
while (true) {
body=in.readLine();
if (body == null) {
break;
}
System.out.println("The time server receive order :" body);
currentTime="QUERY TIME ORDER".equalsIgnoreCase(body)?new java.util.Date(
System.currentTimeMillis()).toString():"BAD ORDER";
out.println(currentTime);
}
} catch (IOException e) {
if (in != null) {
try {
in.close();
} catch (IOException ex) {
ex.printStackTrace();
}
}
if (out != null) {
out.close();
out = null;
}
if (this.socket != null) {
try {
this.socket.close();
} catch (IOException ex) {
ex.printStackTrace();
}
this.socket= null;
}
}
}
}TimeClient：
代码语言：javascript复制package com.jad.nettyArticle.bio;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
public class TimeClient {
public static void main(String[] args) {
int port = 8080;
if (args!=null && args.length>0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
Socket socket = null;
BufferedReader in = null;
PrintWriter out = null;
try {
socket = new Socket("127.0.0.1",port);
in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
out = new PrintWriter(socket.getOutputStream(),true);
out.println("QUERY TIME ORDER");
System.out.println("Send order 2 server succeed.");
String resp = in.readLine();
System.out.println("Now is :" resp);
} catch (IOException e) {
e.printStackTrace();
} finally {
if (out != null) {
out.close();
out = null;
}
if (in != null) {
try {
in.close();
} catch (IOException e) {
e.printStackTrace();
}
in = null;
}
if (socket != null) {
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
socket = null;
}
}
}
}FakeNioTimeServer：
代码语言：javascript复制package com.jad.nettyArticle.fakeNio;
import com.jad.nettyArticle.bio.TimeServerHandler;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
public class FakeNioTimeServer {
public static void main(String[] args) throws IOException {
int port=8080;
if(args!=null && args.length>0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
ServerSocket server =null;
try {
server=new ServerSocket(port);
System.out.println("The time server is start in port :" port);
Socket socket=null;
TimeServerHandlerExecutePool singleExecutor = new TimeServerHandlerExecutePool(50,100);
while (true) {
socket = server.accept();
//new Thread(new TimeServerHandler(socket)).start();
singleExecutor.execute(new TimeServerHandler(socket));
}
} finally {
if (server != null) {
System.out.println("The time server close");
server.close();
server=null;
}
}
}
}TimeServerHandlerExecutePool：
代码语言：javascript复制package com.jad.nettyArticle.fakeNio;
import com.jad.nettyArticle.bio.TimeServerHandler;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class TimeServerHandlerExecutePool {
private ExecutorService executor;
public TimeServerHandlerExecutePool(int maxPoolSize , int queueSize) {
executor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),maxPoolSize,
120L, TimeUnit.SECONDS,new ArrayBlockingQueue<java.lang.Runnable>(queueSize));
}
public void execute(TimeServerHandler timeServerHandler) {
executor.execute(timeServerHandler);
}
}NioTimeServer：
代码语言：javascript复制package com.jad.nettyArticle.nio;
public class NioTimeServer {
public static void main(String[] args) {
int port=8080;
if(args!=null && args.length>0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
MultiplexerTimeServer timeServer=new MultiplexerTimeServer(port);
new Thread(timeServer,"NIO-MultiplexerTimeServer-001").start();
}
}MultiplexerTimeServer：
代码语言：javascript复制package com.jad.nettyArticle.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class MultiplexerTimeServer implements Runnable{
private Selector selector;
private ServerSocketChannel serverChannel;
private volatile boolean stop = false;
public MultiplexerTimeServer(int port) {
try {
selector=Selector.open();
serverChannel=ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.socket().bind(new InetSocketAddress(port),1024);
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("The time server is start in port:" port);
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
public void stop(){
this.stop = true;
}
public void run() {
while (!stop) {
try {
System.out.println("The server thread is running");
selector.select(1000);
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectionKeys.iterator();
System.out.println("The keys is :" it);
SelectionKey key = null;
while (it.hasNext()) {
key=it.next();
it.remove();
try {
System.out.println("handle key is :" key " start");
handleInput(key);
System.out.println("handle key is :" key " end");
} catch (Exception e) {
e.printStackTrace();
if (key != null) {
key.cancel();
if (key.channel() != null) {
key.channel().close();
}
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
if (selector != null) {
try {
selector.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
private void handleInput(SelectionKey key) throws IOException {
System.out.println("handling key is :" key " start");
if (key.isValid()) {
System.out.println("key :" key " valided");
//SocketChannel sc = null;
if (key.isAcceptable()) {
System.out.println("key :" key " be accepted");
ServerSocketChannel ssc = (ServerSocketChannel)key.channel();
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);
sc.register(selector,SelectionKey.OP_READ);
System.out.println("channel related to key :" key " register read operation!");
}
if (key.isReadable()) {
System.out.println("key :" key " can be read !");
SocketChannel sc = (SocketChannel)key.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(readBuffer);
System.out.println("channel related to key :" key " length is " readBytes);
if (readBytes>0) {
readBuffer.flip();
byte[] bytes = new byte[readBuffer.remaining()];
readBuffer.get(bytes);
String body = new String(bytes,"UTF-8");
System.out.println("The time server receive order :" body);
String currentTime="QUERY TIME ORDER".equalsIgnoreCase(body)?new java.util.Date(
System.currentTimeMillis()).toString():"BAD ORDER";
System.out.println(currentTime);
doWrite(sc,currentTime);
} else if(readBytes<0){
key.cancel();
sc.close();
}else {
;
}
}
}
}
private void doWrite(SocketChannel sc, String response) throws IOException {
if (response != null && response.trim().length()>0) {
byte[] bytes = response.getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
sc.write(writeBuffer);
}
}
}NioTimeClient ：
代码语言：javascript复制package com.jad.nettyArticle.nio;
public class NioTimeClient {
public static void main(String[] args) {
int port = 8080;
if (args!=null && args.length>0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
new Thread(new TimeClientHandler("127.0.0.1",port),"TimeClient-001").start();
}
}TimeClientHandler：
代码语言：javascript复制package com.jad.nettyArticle.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class TimeClientHandler implements Runnable {
private String host;
private int port;
private Selector selector;
private SocketChannel socketChannel;
private volatile boolean stop = false;
public TimeClientHandler(String ip,int port) {
this.host = ip ==null?"127.0.0.1":ip;
this.port = port;
try {
selector=Selector.open();
socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);
System.out.println("Open the time channel succeed ,host :" host ",port :" port);
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
public void run() {
try {
System.out.println("Connnet the time server start!");
doConnect();
System.out.println("Connnet the time server end!");
} catch (Exception e) {
e.printStackTrace();
System.exit(1);
}
while (!stop) {
try {
System.out.println("The client thread is running");
selector.select(1000);
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectionKeys.iterator();
System.out.println("The keys is :" it);
SelectionKey key = null;
while (it.hasNext()) {
key=it.next();
it.remove();
try {
System.out.println("handle key is :" key " start");
handleInput(key);
System.out.println("handle key is :" key " end");
} catch (Exception e) {
if (key != null) {
key.cancel();
if (key.channel() != null) {
key.channel().close();
}
}
e.printStackTrace();
}
}
} catch (IOException e) {
e.printStackTrace();
System.exit(1);
}
}
if (selector != null) {
try {
selector.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
private void doConnect() throws IOException {
System.out.println("Connnet the time server!");
if (socketChannel.connect(new InetSocketAddress(host,port))) {
socketChannel.register(selector,SelectionKey.OP_READ);
doWrite(socketChannel);
System.out.println("channel related to server register read operation!");
} else {
socketChannel.register(selector,SelectionKey.OP_CONNECT);
System.out.println("channel related to server register connect operation!");
}
}
private void handleInput(SelectionKey key) throws IOException {
System.out.println("handling key is :" key " start");
if (key.isValid()) {
System.out.println("key :" key " valided");
SocketChannel sc = (SocketChannel)key.channel();
if (key.isConnectable()) {
if (sc.finishConnect()) {
System.out.println("key :" key " be Connected");
sc.register(selector,SelectionKey.OP_READ);
System.out.println("channel related to key :" key " register read operation!");
doWrite(sc);
System.out.println("channel related to key :" key " execute write action and send request!");
} else {
System.exit(1);
}
}
if (key.isReadable()) {
System.out.println("key :" key " can be read !");
//SocketChannel sc = (SocketChannel)key.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(readBuffer);
System.out.println("channel related to key :" key " length is " readBytes);
if (readBytes>0) {
readBuffer.flip();
byte[] bytes = new byte[readBuffer.remaining()];
readBuffer.get(bytes);
String body = new String(bytes,"UTF-8");
System.out.println("Now is :" body);
this.stop=true;
} else if(readBytes<0){
key.cancel();
sc.close();
}else {
;
}
}
}
}
private void doWrite(SocketChannel sc) throws IOException {
byte[] bytes = "QUERY TIME ORDER".getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
sc.write(writeBuffer);
if (!writeBuffer.hasRemaining()) {
System.out.println("Send order 2 server succeed.");
}
}
}AioTimeServer：
代码语言：javascript复制package com.jad.nettyArticle.aio;
public class AioTimeServer {
public static void main(String[] args) {
int port=8080;
if(args!=null && args.length>0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
AysncTimeServerHandler timeServer=new AysncTimeServerHandler(port);
new Thread(timeServer,"AIO-AysncTimeServerHandler-001").start();
}
}AysncTimeServerHandler：
代码语言：javascript复制package com.jad.nettyArticle.aio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.concurrent.CountDownLatch;
public class AysncTimeServerHandler implements Runnable{
private int port;
public CountDownLatch latch;
public AsynchronousServerSocketChannel asynchronousServerSocketChannel;
public AysncTimeServerHandler(int port) {
this.port=port;
try {
asynchronousServerSocketChannel=AsynchronousServerSocketChannel.open();
asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
System.out.println("The time server is start in port :" port);
} catch (IOException e) {
e.printStackTrace();
}
}
public void run() {
latch=new CountDownLatch(1);
doAccept();
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private void doAccept() {
asynchronousServerSocketChannel.accept(this,new AcceptCompletionHandler());
}
}AcceptCompletionHandler ：
代码语言：javascript复制package com.jad.nettyArticle.aio;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
public class AcceptCompletionHandler implements java.nio.channels.CompletionHandler<java.nio.channels.AsynchronousSocketChannel, AysncTimeServerHandler> {
@Override
public void completed(AsynchronousSocketChannel result, AysncTimeServerHandler attachment) {
attachment.asynchronousServerSocketChannel.accept(attachment,this);
ByteBuffer buffer=ByteBuffer.allocate(1024);
result.read(buffer,buffer,new ReadCompletionHandler(result));
}
@Override
public void failed(Throwable exc, AysncTimeServerHandler attachment) {
exc.printStackTrace();
attachment.latch.countDown();
}
}ReadCompletionHandler：
代码语言：javascript复制package com.jad.nettyArticle.aio;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {
private AsynchronousSocketChannel channel;
public ReadCompletionHandler(AsynchronousSocketChannel result) {
if (this.channel==null) {
this.channel=result;
}
}
@Override
public void completed(Integer result, ByteBuffer attachment) {
attachment.flip();
byte[] body=new byte[attachment.remaining()];
attachment.get(body);
try {
String req = new String(body,"UTF-8");
System.out.println("The time server receive order :" req);
String currentTime="QUERY TIME ORDER".equalsIgnoreCase(req)?new java.util.Date(
System.currentTimeMillis()).toString():"BAD ORDER";
System.out.println(currentTime);
doWrite(currentTime);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
private void doWrite(String currentTime) {
if (currentTime != null && currentTime.trim().length()>0) {
byte[] bytes=currentTime.getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
channel.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
if (attachment.hasRemaining()) {
channel.write(attachment,attachment,this);
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
this.channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}AioTimeClient：
代码语言：javascript复制package com.jad.nettyArticle.aio;
public class AioTimeClient {
public static void main(String[] args) {
int port = 8080;
if (args!=null && args.length>0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
new Thread(new AsyncTimeClientHandler("127.0.0.1",port),"AIO-AsyncTimeClientHandler-001").start();
}
}AsyncTimeClientHandler：
代码语言：javascript复制package com.jad.nettyArticle.aio;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;
public class AsyncTimeClientHandler implements CompletionHandler<Void, AsyncTimeClientHandler>,Runnable {
private AsynchronousSocketChannel client;
private String host;
private int port;
private CountDownLatch latch;
public AsyncTimeClientHandler(String s, int port) {
this.host=s;
this.port=port;
try {
client=AsynchronousSocketChannel.open();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void run() {
latch=new CountDownLatch(1);
client.connect(new InetSocketAddress(host,port),this,this);
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void completed(Void result, AsyncTimeClientHandler attachment) {
byte[] bytes = "QUERY TIME ORDER".getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
writeBuffer.put(bytes);
writeBuffer.flip();
client.write(writeBuffer, writeBuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
if (attachment.hasRemaining()) {
client.write(attachment,attachment,this);
} else {
ByteBuffer readbuffer=ByteBuffer.allocate(1024);
client.read(readbuffer, readbuffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer attachment) {
attachment.flip();
byte[] bytes = new byte[attachment.remaining()];
attachment.get(bytes);
try {
String body = new String(bytes,"UTF-8");
System.out.println("Now is :" body);
latch.countDown();
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
}
@Override
public void failed(Throwable exc, ByteBuffer attachment) {
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
@Override
public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
exc.printStackTrace();
try {
client.close();
latch.countDown();
} catch (IOException e) {
e.printStackTrace();
}
}
}NettyTimeServer：
代码语言：javascript复制package com.jad.nettyArticle.netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class NettyTimeServer {
public static void main(String[] args) throws Exception {
int port = 8080;
if (args!=null && args.length>0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
new NettyTimeServer().bind(port);
}
private void bind(int port) throws Exception {
EventLoopGroup bossGroup=new NioEventLoopGroup();
EventLoopGroup workerGroup=new NioEventLoopGroup();
try {
ServerBootstrap b=new ServerBootstrap();
b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG,1024)
.childHandler(new ChildChannelHandler());
ChannelFuture f=b.bind(port).sync();
f.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new NettyTimeServerHandler());
}
}
}NettyTimeServerHandler：
代码语言：javascript复制package com.jad.nettyArticle.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
public class NettyTimeServerHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf=(ByteBuf)msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body=new String(req,"UTF-8");
System.out.println("The time server receive order :" body);
String currentTime="QUERY TIME ORDER".equalsIgnoreCase(body)?new java.util.Date(
System.currentTimeMillis()).toString():"BAD ORDER";
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
ctx.write(resp);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}NettyTimeClient：
代码语言：javascript复制package com.jad.nettyArticle.netty;
import com.jad.nettyArticle.nio.TimeClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
public class NettyTimeClient {
public static void main(String[] args) throws InterruptedException {
int port = 8080;
if (args!=null && args.length>0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
new NettyTimeClient().connect(port,"127.0.0.1");
}
private void connect(int port, String s) throws InterruptedException {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b=new Bootstrap();
b.group(group).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY,true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new NettyTimeClientHandler());
}
});
ChannelFuture f=b.connect(s,port).sync();
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}NettyTimeClientHandler：
代码语言：javascript复制package com.jad.nettyArticle.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import java.util.logging.Logger;
public class NettyTimeClientHandler extends ChannelHandlerAdapter {
private static final Logger logger = Logger.getLogger(NettyTimeClientHandler.class.getName());
private final ByteBuf firstMessage;
public NettyTimeClientHandler() {
byte[] bytes="QUERY TIME ORDER".getBytes();
this.firstMessage = Unpooled.buffer(bytes.length);
this.firstMessage.writeBytes(bytes);
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(firstMessage);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf=(ByteBuf)msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body=new String(req,"UTF-8");
System.out.println("Now is :" body);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
logger.warning("Unexpected exception from downstream :" cause.getMessage());
ctx.close();
}
}参考资料： https://blog.csdn.net/m0_38125278/article/details/85610025 https://blog.csdn.net/g1l2y3/article/details/53114765 https://blog.csdn.net/wireless_com/article/details/45935591 https://blog.csdn.net/buyueliuying/article/details/53572066 https://blog.csdn.net/qq_34781336/article/details/84834801
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/186679.html原文链接：https://javaforall.cn
