本文详细介绍组成非阻塞通信的几大类：Buffer、Channel、Selector、SelectionKey

非阻塞通信的流程ServerSocketChannel通过open方法获取ServerSocketChannel，通过ServerSocketChannel设置为非阻塞模式，再通过ServerSocketChannel获取socket，绑定服务进程监听端口。服务启动成功。

然后就是非阻塞通信的精髓了，Selector通过静态的open()方法获取到Selector，然后ServerSocketChannel注册Selection.OP_ACCEPT事件到Selector上。

Selector就会监控事件发生，Selector通过select()监控已发生的SelectionKey对象的数目，通过selectKeys()方法返回对应的selectionKey对象集合。遍历该集合得到相应的selectionKey对象，通过该对象的channel()方法获取关联的ServerSocketChannel对象， 通过selector()方法就可以获取关联的Selector对象。

通过上面获取的ServerSocketChannel执行accept()方法获取SocketChannel，再通过SocketChannel设置为非阻塞模式，在将SocketChannel注册到上面创建的Selector上，注册SelectionKey.OP_READ |SelectionKey.OP_WRITE 事件。

Selector将在监控对应上面绑定的事件，监控到对应的事件的话执行读和写的操作。

示例代码：

上面描述了服务端非阻塞方式通信的一个流程，下面通过具体代码实现：/**

* 非阻塞模式

*

*/public class EchoServer2 {

private Selector selector = null;

private ServerSocketChannel serverSocketChannel = null;

private int port = 8001;

private Charset charset = Charset.forName("UTF-8");

public EchoServer2() throws IOException {

selector = Selector.open();

serverSocketChannel = ServerSocketChannel.open(); //服务器重启的时候，重用端口

serverSocketChannel.socket().setReuseAddress(true); //设置非阻塞模式

serverSocketChannel.configureBlocking(false);

serverSocketChannel.socket().bind(new InetSocketAddress(port));

System.out.println("服务器启动成功");

} /**

* 服务方法

*/

public void service() throws IOException {

serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

while (selector.select() > 0) {

Set readyKes = selector.selectedKeys();

Iterator it = readyKes.iterator();

while (it.hasNext()) {

SelectionKey key = null;

try {

key = (SelectionKey) it.next();

it.remove();

if (key.isAcceptable()) {

System.out.println("连接事件");

//连接事件

ServerSocketChannel ssc = (ServerSocketChannel) key.channel();

SocketChannel socketChannel = ssc.accept();

System.out.println("接收到客户连接，来自：" + socketChannel.socket().getInetAddress() + " : " + socketChannel.socket().getPort());

socketChannel.configureBlocking(false);

ByteBuffer buffer = ByteBuffer.allocate(1024);

socketChannel.register(selector, SelectionKey.OP_READ |

SelectionKey.OP_WRITE, buffer);

} else if (key.isReadable()) {

//接收数据

receive(key);

} else if (key.isWritable()) {

//发送数据

send(key);

}

} catch (IOException e) {

e.printStackTrace();

try {

if (key != null) {

key.cancel();

key.channel().close();

}

}catch (IOException ex){

ex.printStackTrace();

}

}

}

}

} private void send(SelectionKey key) throws IOException {

ByteBuffer buffer = (ByteBuffer) key.attachment();

SocketChannel channel = (SocketChannel) key.channel();

buffer.flip(); //把极限设置为位置，把位置设置为0

String data = decode(buffer);

if (data.indexOf("\r\n") == -1) {

return;

}

String outputData = data.substring(0, data.indexOf("\n") + 1);

System.out.println("请求数据：" + outputData);

ByteBuffer outputBuffer = encode("echo:" + outputData);

while (outputBuffer.hasRemaining()) {

channel.write(outputBuffer);

}

ByteBuffer temp = encode(outputData);

buffer.position(temp.limit());

buffer.compact(); if (outputData.equals("bye\r\n")) {

key.cancel();

channel.close();

System.out.println("关闭与客户的连接");

}

} private String decode(ByteBuffer buffer) {

CharBuffer charBuffer = charset.decode(buffer); return charBuffer.toString();

} private ByteBuffer encode(String s) { return charset.encode(s);

} private void receive(SelectionKey key) throws IOException {

ByteBuffer buffer = (ByteBuffer) key.attachment();

SocketChannel socketChannel = (SocketChannel) key.channel();

ByteBuffer readBuff = ByteBuffer.allocate(32);

socketChannel.read(readBuff);

readBuff.flip();

buffer.limit(buffer.capacity());

buffer.put(readBuff);

} public static void main(String[] args) throws IOException { new EchoServer2().service();

}

}/**

* 创建非阻塞客户端

*

*/public class EchoClient2 {

private SocketChannel socketChannel; private int port = 8001; private Selector selector; private ByteBuffer sendBuffer = ByteBuffer.allocate(1024); private ByteBuffer receiveBuffer = ByteBuffer.allocate(1024); private Charset charset = Charset.forName("UTF-8"); public EchoClient2() throws IOException {

socketChannel = SocketChannel.open();

InetSocketAddress inetSocketAddress = new InetSocketAddress(InetAddress.getLocalHost(), port);

socketChannel.connect(inetSocketAddress);//

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

System.out.println("与服务器连接成功");

selector = Selector.open();

} public static void main(String[] args) throws IOException { final EchoClient2 client = new EchoClient2();

Thread receiver = new Thread(new Runnable() { @Override

public void run() {

client.receiveFromUser();

}

});

receiver.start();

client.talk();

} private void receiveFromUser() { try {

System.out.println("请输入数据：");

BufferedReader localReader = new BufferedReader(new InputStreamReader(System.in));

String msg = null; while ((msg = localReader.readLine()) != null) {

System.out.println("用户输入的数据：" + msg); synchronized (sendBuffer) {

sendBuffer.put(encode(msg + "\r\n"));

} if (msg.equalsIgnoreCase("bye")) { break;

}

}

} catch (IOException e) {

e.printStackTrace();

}

} private ByteBuffer encode(String s) { return charset.encode(s);

} private void talk() throws IOException {

socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE); while (selector.select() > 0) {

Set keys = selector.selectedKeys();

Iterator it = keys.iterator(); while (it.hasNext()) {

SelectionKey key = null; try {

key = it.next();

it.remove(); if (key.isReadable()) { //System.out.println("读事件");

//读事件

receive(key);

} if (key.isWritable()) { // System.out.println("写事件");

//写事件

send(key);

}

} catch (IOException e) {

e.printStackTrace(); if (key != null) {

key.cancel();

key.channel().close();

}

}

}

}

} private void send(SelectionKey key) throws IOException {

SocketChannel channel = (SocketChannel) key.channel(); synchronized (sendBuffer) {

sendBuffer.flip();//把极限设为位置，把位置设为零

channel.write(sendBuffer);

sendBuffer.compact();//删除已经发送的数据。

}

} private void receive(SelectionKey key) throws IOException {

SocketChannel channel = (SocketChannel) key.channel();

channel.read(receiveBuffer);

receiveBuffer.flip();//将limit的值设置为position的值，将position的值设置为0

String receiveData = decode(receiveBuffer); if (receiveData.indexOf("\n") == -1) { return;

}

String outputData = receiveData.substring(0, receiveData.indexOf("\n") + 1);

System.out.println("响应数据：" + outputData); if (outputData.equalsIgnoreCase("echo:bye\r\n")) {

key.cancel();

socketChannel.close();

;

System.out.println("关闭与服务器的连接");

selector.close();

System.exit(0);

}

ByteBuffer temp = encode(outputData);

receiveBuffer.position(temp.limit());

receiveBuffer.compact();//删除已经打印的数据

} private String decode(ByteBuffer receiveBuffer) {

CharBuffer buffer = charset.decode(receiveBuffer); return buffer.toString();

}

}

实现非阻塞通信的方式缓冲区

通道

Selector

缓冲区作用：减少物理读写次数，减少内存创建和销毁次数。 缓冲区的属性：capacity(最大容量)、limit(实际容量)、position(当前位置)。PS：其他地方是翻译成capacity(容量)、limit(极限)、position位置)，我个人觉得翻译成上面的更好理解，为啥通过下面的方法解析和图解就可明白。当然最好通过英文表达这样最清楚。

三个属性的关系为：capacity≥limit≥position≥0

图解关系如下：

缓冲区类结构：

java.nio.ByteBuffer类是一个抽象类，不能被实例化。但是提供了8个具体的实现类，其中最基本的的缓冲区是ByteBuffer，它存放的数据单元是字节。

常用方法：

clear()：把limit设置为capacity，再把位置设为0

flip()：把limit设置为position，再把位置设置为0。

rewind()：不改变limit，把位置设为0。

allocate():创建一个缓冲中，方法参数指定缓冲区大小

compact():将缓冲区的当前位置和界限之间的字节(如果有)复制到缓冲区的开始处。

测试上述方法：

测试clear()方法@Test

public void testClear() { //创建一个10chars大小的缓冲区，默认情况下limit和capacity是相等的

CharBuffer buffer = CharBuffer.allocate(10);

System.out.println("创建默认情况");

printBufferInfo(buffer);

buffer.limit(8);//修改limit的值

System.out.println("修改limit后");

printBufferInfo(buffer); // clear()：把limit设置为capacity，再把位置设为0

buffer.clear();

System.out.println("执行clear()方法后");

printBufferInfo(buffer);

}

执行结果如下：

测试flip()方法：@Test

public void testFlip() {

CharBuffer buffer = CharBuffer.allocate(10);

System.out.println("创建默认情况");

printBufferInfo(buffer); //put的方法会修改position的值

buffer.put('H');

buffer.put('E');

buffer.put('L');

buffer.put('L');

buffer.put('O');

System.out.println("调用put方法后：");

printBufferInfo(buffer); //flip()：把limit设置为position，再把位置设置为0。

buffer.flip();

System.out.println("调用flip方法后：");

printBufferInfo(buffer);

}

执行结果如下：

测试rewind()方法@Test

public void testRewind() {

CharBuffer buffer = CharBuffer.allocate(10);

System.out.println("创建默认情况");

printBufferInfo(buffer); //put的方法会修改position的值

buffer.put('H');

buffer.put('E');

buffer.put('L');

buffer.put('L');

buffer.put('O');

buffer.limit(8);

System.out.println("调用put、limit方法后：");

printBufferInfo(buffer); //rewind()：不改变limit，把位置设为0。

buffer.rewind();

System.out.println("调用rewind方法后：");

printBufferInfo(buffer);

}

执行结果如下：

测试compact()方法@Test

public void testCompact(){

CharBuffer buffer = CharBuffer.allocate(10);

System.out.println("创建默认情况");

printBufferInfo(buffer); //put的方法会修改position的值

buffer.put('H');

buffer.put('E');

buffer.put('L');

buffer.put('L');

buffer.put('O');

buffer.limit(8);//修改limit的值

System.out.println("调用put和limit方法后：");

printBufferInfo(buffer);

System.out.println("调用compact方法后："); //将缓冲区的当前位置和界限之间的字节(如果有)复制到缓冲区的开始处。

buffer.compact();

printBufferInfo(buffer);

}

这是JDK中介绍该方法的作用：

将缓冲区的当前位置和界限之间的字节(如果有)复制到缓冲区的开始处。即将索引 p = position() 处的字节复制到索引 0 处，将索引 p + 1 处的字节复制到索引 1 处，依此类推，直到将索引 limit() - 1 处的字节复制到索引 n = limit() - 1 - p 处。然后将缓冲区的位置设置为 n+1，并将其界限设置为其容量。如果已定义了标记，则丢弃它。

官方表示的太难理解了：

将缓冲区的当前位置和界限之间的字节(如果有)复制到缓冲区的开始处。并将limit(实际容量)设置为 capacity(最大容量)。执行compact()方法前，limit的值是：8，position的值是：5。按照上面描述的执行完compact()后，position的值计算方式是：n+1;n=limit-1-p;所有n=8-1-5=2,最后position的值为：2+1=3。和程序运行的结果一致。

可以在这种情况：从缓冲区写入数据之后调用此方法，以防写入不完整。buf.clear(); // Prepare buffer for use

while (in.read(buf) >= 0 || buf.position != 0) {

buf.flip();

out.write(buf);

buf.compact(); // In case of partial write

}

如果out.write()方法没有将缓存中的数据读取完，这个时候的position位置指向的是剩余数据的位置。达到防止写入不完整。

通道作用： 连接缓冲区与数据源或数据目的地。

常用类：

Channel

接口有下面两个子接口ReadableByteChannel和WritableByteChannel和一个抽象实现类SelectableChannel。

在ReadableByteChannel接口中申明了read(ByteBuffer

dst)方法。在WritableByteChannel接口中申明了write(ByteBuffer[]

srcs)：方法。SelectableChannel抽象类中主要方法，configureBlocking(boolean

block)、register();方法。 ByteChannel

接口继承了ReadableChannel和WritableChannel。所以ByteChannel具有读和写的功能。

ServerSocketChannel继承了SelectableChannel类抽象类，所以SocketChannel具有设置是否是阻塞模式、向selector注册事件功能。

SocketChannel也继承了SelectableChannel类还实现ByteChannel接口，所以SocketChannel具有设置是否是阻塞模式、向selector注册事件、从缓冲区读写数据的功能。

通过类图展现：

Selector类：作用：只要ServerSocketChannel及SocketChannel向Selector注册了特定的事件，Selector就会监听这些事件的发生。

流程：

Selector通过静态的open()方法创建一个Selector对象，SelectableChannel类向Selector注册了特定的事件。Selector就会监控这些事件发生，Selector通过select()监控已发生的SelectionKey对象的数目，通过selectKeys()方法返回对应的selectionKey对象集合。遍历该集合得到相应的selectionKey对象，通过该对象的channel()方法获取关联的SelectableChannel对象，

通过selector()方法就可以获取关联的Selector对象。

Note：

当Selector的select()方法还有一个重载方式:select(long timeout)。并且该方法采用阻塞的工作方式，如果相关事件的selectionKey对象的数目一个也没有，就进入阻塞状态。知道出现以下情况之一，才从select()方法中返回。至少有一个SelectionKey的相关事件已经发生。

其他线程调用了Selector的wakeup()方法，导致执行select()方法的线程立即返回。

当前执行的select()方法的线程被中断。

超出了等待时间。仅限调用select(long timeout)方法时出现。如果没有设置超时时间，则永远不会超时。Selector类有两个非常重要的方法： 静态方法open()，这是Selector的静态工厂方法，创建一个Selector对象。

selectedKeys()方法返回被Selector捕获的SelectionKey的集合。

SelectionKey类作用：

ServerSocketChannel或SocketChannel通过register()方法向Selector注册事件时，register()方法会创建一个SelectionKey对象，该对象是用来跟踪注册事件的句柄。在SelectionKey对象的有效期间，Selector会一直监控与SelectionKey对象相关的事件，如果事件发生，就会把SelectionKey对象添加到Selected-keys集合中。

SelectionKey中定义的事件： 定义了4种事件：

1、SelectionKey.OP_ACCEPT：接收连接就绪事件，表示服务器监听到了客户连接，服务器可以接收这个连接了。常量值为16.

2、SelectionKey.OP_CONNECT：连接就绪事件，表示客户与服务器的连接已经建立成功。常量值为8.

3、SelectionKey.OP_READ：读就绪事件，表示通道中已经有了可读数据可以执行读操作。常量值为1.

4、SelectionKey.OP_WRITE：写就绪事件，表示已经可以向通道写数据了。常量值为4.

常用方法：

channel()方法：返回与它关联的SelectedChannel(包括ServerSocketChannel和SocketChannel)。

selector()方法：返回与它关联的Selector对象。

它们之间的关系如下：

以上就是Java网络编程由浅入深三 一文了解非阻塞通信的图文代码示例详解的内容，更多相关内容请关注PHP中文网(www.php.cn)！

本文原创发布php中文网，转载请注明出处，感谢您的尊重！