netty学习十一:NIO客户端服务端通讯demo

package nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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.nio.charset.Charset;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

public class NioServer {
private static Map<String,SocketChannel> clientMap = new HashMap<>();

public static void main(String[] args) throws IOException {
//创建一个服务端的socket channel,但是还未绑定监听地址
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
//非阻塞
serverSocketChannel.configureBlocking(false);

ServerSocket serverSocket = serverSocketChannel.socket();
//通过ServerSocet为ServerSocketChannel指定监听地址
//如果不绑定的话,调用ServerSocketChannel的accept方法会抛出NotYetBoundException
serverSocket.bind(new InetSocketAddress(8899));

Selector selector = Selector.open();
//注册连接建立的事件
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

while (true) {
//一直阻塞,直到有感兴趣的事件发生,select方法才会返回
selector.select();
//注册selector在上的事件可能不止一个,这些事件也可能同时发生,所以这里是SelectionKey Set集合
Set<SelectionKey> selectedKeys = selector.selectedKeys();
for (Iterator<SelectionKey> iterator = selectedKeys.iterator(); iterator.hasNext();) {
SelectionKey selectionKey = (SelectionKey) iterator.next();
final SocketChannel client;
//isAcceptable==true,说明客户端和服务端已经建立好连接了
if(selectionKey.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel)selectionKey.channel();
client = server.accept();

client.configureBlocking(false);
client.register(selector,SelectionKey.OP_READ);

String key = "【"+UUID.randomUUID()+"】";
clientMap.put(key, client);
continue;
}

//isReadable==true,说明可以从SocketChannel中读取数据了
if(selectionKey.isReadable()) {
client = (SocketChannel)selectionKey.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
//从channel中读取数据
int count = client.read(readBuffer);
if(count > 0) {
readBuffer.flip();
Charset charset = Charset.forName("utf-8");
//将bytebuffer转换成字符串
String receiveMessage = String.valueOf(charset.decode(readBuffer).array());
System.out.println(client+":"+receiveMessage);

//找出发送消息的客户端
String sendKey = "";
for (String key : clientMap.keySet()) {
SocketChannel socketChannel = clientMap.get(key);
if (socketChannel == client) {
sendKey = key;
break;
}
}

for (String key : clientMap.keySet()) {
SocketChannel socketChannel = clientMap.get(key);
ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
writeBuffer.put((sendKey + ":" +receiveMessage).getBytes());
writeBuffer.flip();
socketChannel.write(writeBuffer);
}
}
}
}
//清除事件
selectedKeys.clear();
}
}
}

package nio;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
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.time.LocalDateTime;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class NioClient {
public static void main(String[] args) throws IOException {
SocketChannel socketChannel = SocketChannel.open();
//非阻塞
socketChannel.configureBlocking(false);

//客户端同样需要selector去监听channel
Selector selector = Selector.open();
socketChannel.register(selector, SelectionKey.OP_CONNECT);
socketChannel.connect(new InetSocketAddress("127.0.0.1", 8899));

while(true) {
selector.select();
Set<SelectionKey> selectedKeys = selector.selectedKeys();
for (SelectionKey selectionKey : selectedKeys) {
//isConnectable=true,说明已经跟服务端建立连接了
if(selectionKey.isConnectable()) {
SocketChannel client = (SocketChannel)selectionKey.channel();
if (client.isConnectionPending()) {
client.finishConnect();
ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
writeBuffer.put((LocalDateTime.now() + "连接成功").getBytes());
writeBuffer.flip();
client.write(writeBuffer);

//异步给服务端写信息,避免阻塞主线程
ExecutorService executorService = Executors.newSingleThreadExecutor(Executors.defaultThreadFactory());
executorService.submit(new Runnable() {
@Override
public void run() {
while(true) {
writeBuffer.clear();
InputStreamReader inputStreamReader = new InputStreamReader(System.in);
BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
try {
String sendMessage = bufferedReader.readLine();
writeBuffer.put(sendMessage.getBytes());
writeBuffer.flip();
client.write(writeBuffer);
} catch (IOException e) {
e.printStackTrace();
}
}
}
});
}

//一旦客户端和服务端建立连接了,立刻注册读事件
client.register(selector, SelectionKey.OP_READ);
continue;
}

if(selectionKey.isReadable()) {
SocketChannel client = (SocketChannel)selectionKey.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int count = client.read(readBuffer);
if (count > 0) {
String receiveMessage = new String(readBuffer.array(),0,count);
System.out.println(receiveMessage);
}
}
}

//清除事件
selectedKeys.clear();
}
}
}