7.Java NIO系列教程之Server/Client完整示例

时间：2022-09-05 13:11:56

TCPServer类：

package com.gw.demo;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;
import java.util.Set;

public class TCPServer {
// 缓冲区大小
private static final int BufferSize = 1024;

// 超时时间，单位毫秒
private static final int TimeOut = 3000;

// 本地监听端口
private static final int ListenPort = 1978;

public static void main(String[] args) throws IOException {
// 创建选择器
Selector selector = Selector.open();

// 打开监听信道
ServerSocketChannel listenerChannel = ServerSocketChannel.open();

// 与本地端口绑定
listenerChannel.socket().bind(new InetSocketAddress(ListenPort));

// 设置为非阻塞模式
listenerChannel.configureBlocking(false);

// 将选择器绑定到监听信道,只有非阻塞信道才可以注册选择器.并在注册过程中指出该信道可以进行Accept操作
//一个server socket channel准备好接收新进入的连接称为“接收就绪”
listenerChannel.register(selector, SelectionKey.OP_ACCEPT);

// 创建一个处理协议的实现类,由它来具体操作
TCPProtocol protocol = new TCPProtocolImpl(BufferSize);

// 反复循环,等待IO
while (true) {
// 等待某信道就绪(或超时)
int keys = selector.select(TimeOut);
//刚启动时连续输出0，client连接后一直输出1
//System.out.print(keys);
if (keys == 0) {
System.out.println("独自等待.");
continue;
}

/*if (selector.select(TimeOut) == 0) {
System.out.println("独自等待.");
continue;
}*/

// 取得迭代器.selectedKeys()中包含了每个准备好某一I/O操作的信道的SelectionKey
Set<SelectionKey> set = selector.selectedKeys();
//输出为1
//System.out.println("selectedKeysSize:" + set.size());
Iterator<SelectionKey> keyIter  = set.iterator();

while (keyIter.hasNext()) {
SelectionKey key = keyIter.next();

try {
if (key.isAcceptable()) {
System.out.println("acceptable");
//该方法在内部，会将interest由OP_ACCEPT改为OP_READ
//如果不执行下面的语句，则会一直是accept状态（初始时设置为了accept），无法进入后面的两个if语句
//console一直打印上面的语句
protocol.handleAccept(key);
}

if (key.isReadable()) {
// 从客户端读取数据
System.out.println("readable");
protocol.handleRead(key);
}

if (key.isValid() && key.isWritable()) {
//客户端连接一次后，N次连续进入该方法
//System.out.println("writable");//连续输出
protocol.handleWrite(key);
}
} catch (IOException ex) {
// 出现IO异常（如客户端断开连接）时移除处理过的键
keyIter.remove();
continue;
}

// 移除处理过的键
keyIter.remove();
}
}
}
}

TCPProtocol接口： 

package com.gw.demo;

import java.io.IOException;
import java.nio.channels.SelectionKey;

public interface TCPProtocol {
/**
 * 接收一个SocketChannel的处理
 * 
 * @param key
 * @throws IOException
 */
void handleAccept(SelectionKey key) throws IOException;

/**
 * 从一个SocketChannel读取信息的处理
 * 
 * @param key
 * @throws IOException
 */
void handleRead(SelectionKey key) throws IOException;

/**
 * 向一个SocketChannel写入信息的处理
 * 
 * @param key
 * @throws IOException
 */
void handleWrite(SelectionKey key) throws IOException;
}


TCPProtocolImpl实现类：

package com.gw.demo;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.text.SimpleDateFormat;
import java.util.Date;

public class TCPProtocolImpl implements TCPProtocol {
private int bufferSize;

public TCPProtocolImpl(int bufferSize) {
this.bufferSize = bufferSize;
}

/**
 * 将可连接 调整为 可读取
 */
public void handleAccept(SelectionKey key) throws IOException {
SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
clientChannel.configureBlocking(false);
clientChannel.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocate(bufferSize));
}

public void handleRead(SelectionKey key) throws IOException {
// 获得与客户端通信的信道
SocketChannel clientChannel = (SocketChannel) key.channel();

// 得到并清空缓冲区
ByteBuffer buffer = (ByteBuffer) key.attachment();
buffer.clear();

// 读取信息获得读取的字节数
long bytesRead = clientChannel.read(buffer);

if (bytesRead == -1) {
// 没有读取到内容的情况
clientChannel.close();
} else {
// 将缓冲区准备为数据传出状态
buffer.flip();

// 将字节转化为为UTF-16的字符串
String receivedString = Charset.forName("UTF-16").newDecoder().decode(buffer).toString();

// 控制台打印出来
System.out.println("接收到来自" + clientChannel.socket().getRemoteSocketAddress() + "的信息:" + receivedString);

SimpleDateFormat format = new SimpleDateFormat("yyyy-mm-dd HH:mm:ss");
String f = format.format(new Date());
// 准备发送的文本
String sendString = "你好,客户端. @" + f + "，已经收到你的信息:" + receivedString;
buffer = ByteBuffer.wrap(sendString.getBytes("UTF-16"));
clientChannel.write(buffer);

// 设置为下一次读取或是写入做准备
key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
}
}

public void handleWrite(SelectionKey key) throws IOException {
// do nothing
}
}


TCPClient类：

package com.gw.demo;

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;

public class TCPClient {
//信道选择器
private Selector selector;

// 与服务器通信的信道
SocketChannel socketChannel;

// 要连接的服务器Ip地址
private String hostIp;

// 要连接的远程服务器在监听的端口
private int hostListenningPort;

/**
 * 构造函数
 * 
 * @param HostIp
 * @param HostListenningPort
 * @throws IOException
 */
public TCPClient(String HostIp, int HostListenningPort) throws IOException {
this.hostIp = HostIp;
this.hostListenningPort = HostListenningPort;

initialize();
}

/**
 * 初始化
 * 
 * @throws IOException
 */
private void initialize() throws IOException {
// 打开监听信道并设置为非阻塞模式
socketChannel = SocketChannel.open(new InetSocketAddress(hostIp, hostListenningPort));
socketChannel.configureBlocking(false);

// 打开并注册选择器到信道
selector = Selector.open();
socketChannel.register(selector, SelectionKey.OP_READ);

// 启动读取线程
new TCPClientReadThread(selector);
}

/**
 * 发送字符串到服务器
 * 
 * @param message
 * @throws IOException
 */
public void sendMsg(String message) throws IOException {
ByteBuffer writeBuffer = ByteBuffer.wrap(message.getBytes("UTF-16"));

int r = socketChannel.write(writeBuffer);
System.out.println("write return:" + r);
//socketChannel.
}

public static void main(String[] args) throws IOException {
TCPClient client = new TCPClient("10.10.24.67", 1978);
for(int i=0; i<10; i++){
client.sendMsg("Nio" + i);
/*try {
Thread.sleep(20);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}*/
}

}
}

TCPClientReadThread类： 

package com.gw.demo;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class TCPClientReadThread implements Runnable {
private Selector selector;

public TCPClientReadThread(Selector selector) {
this.selector = selector;

new Thread(this).start();
}

@Override
public void run() {
try {
while (selector.select() > 0) {
// 遍历每个有可用IO操作Channel对应的SelectionKey
for (SelectionKey sk : selector.selectedKeys()) {

// 如果该SelectionKey对应的Channel中有可读的数据
if (sk.isReadable()) {
// 使用NIO读取Channel中的数据
SocketChannel sc = (SocketChannel) sk.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
sc.read(buffer);
buffer.flip();

// 将字节转化为为UTF-16的字符串
String receivedString = Charset.forName("UTF-16").newDecoder().decode(buffer).toString();

// 控制台打印出来
System.out.println("接收到来自服务器" + sc.socket().getRemoteSocketAddress() + "的信息:" + receivedString);

// 为下一次读取作准备
sk.interestOps(SelectionKey.OP_READ);
}

// 删除正在处理的SelectionKey
selector.selectedKeys().remove(sk);
}
}
} catch (IOException ex) {
ex.printStackTrace();
}
}
}


服务端输出如下：

独自等待.
独自等待.
acceptable
readable
接收到来自/10.10.24.67:57863的信息:Nio0Nio1Nio2Nio3Nio4Nio5Nio6
readable
接收到来自/10.10.24.67:57863的信息:Nio7Nio8Nio9


客户端输出如下：

接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-18-23 15:18:06，已经收到你的信息:Nio0Nio1Nio2Nio3Nio4Nio5Nio6你好,客户端. @2015-18-23 15:18:06，已经收到你的信息:Nio7Nio8Nio9


说明：客户端循环十次，调用sendMsg，但是服务器端只接收到了两次消息（不固定，多次运行，甚至只会接收到一次消息），客户端只收到了一次消息。这是为什么呢？


在client的买方法中，增加Thread.sleep(20);语句后。

服务端输出如下：

独自等待.
acceptable
readable
接收到来自/10.10.24.67:58343的信息:Nio0
readable
接收到来自/10.10.24.67:58343的信息:Nio1
readable
接收到来自/10.10.24.67:58343的信息:Nio2
readable
接收到来自/10.10.24.67:58343的信息:Nio3
readable
接收到来自/10.10.24.67:58343的信息:Nio4
readable
接收到来自/10.10.24.67:58343的信息:Nio5
readable
接收到来自/10.10.24.67:58343的信息:Nio6
readable
接收到来自/10.10.24.67:58343的信息:Nio7
readable
接收到来自/10.10.24.67:58343的信息:Nio8
readable
接收到来自/10.10.24.67:58343的信息:Nio9


客户端输出如下：

接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio0你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio1
write return:10
接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio2
write return:10
接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio3
write return:10
接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio4
write return:10
接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio5
write return:10
接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio6
write return:10
接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio7
write return:10
接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio8
write return:10
接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-31-23 15:31:56，已经收到你的信息:Nio9



将TCPClient类的main方法改成如下形式：

public static void main(String[] args) throws IOException {
class MyThread extends Thread{
int i;
TCPClient client ;

public MyThread(TCPClient client, int index) {
i = index;
this.client = client;
}

@Override
public void run() {
try {
client.sendMsg("Nio" + i);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

TCPClient client = new TCPClient("10.10.24.67", 1978);
for(int i=0; i<10; i++){
new MyThread(client, i).start();

}

}

服务器端输出如下：

独自等待.
acceptable
readable
接收到来自/10.10.24.67:58503的信息:Nio1
readable
接收到来自/10.10.24.67:58503的信息:Nio0Nio3Nio2Nio4Nio6Nio8Nio5Nio7Nio9


客户端输出如下：

接收到来自服务器/10.10.24.67:1978的信息:你好,客户端. @2015-38-23 15:38:52，已经收到你的信息:Nio1你好,客户端. @2015-38-23 15:38:52，已经收到你的信息:Nio0Nio3Nio2Nio4Nio6Nio8Nio5Nio7Nio9



标签：

相关文章