ServerSocketChannel与SocketChannel的使用

TCPServer.java

[html] view
plain copy

print?

package com.xy.nio;

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;

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操作

listenerChannel.register(selector, SelectionKey.OP_ACCEPT);

// 创建一个处理协议的实现类,由它来具体操作

TCPProtocol protocol = new TCPProtocolImpl(BufferSize);

// 反复循环,等待IO

while (true) {

// 等待某信道就绪(或超时)

if (selector.select(TimeOut) == 0) {// 监听注册通道，当其中有注册的 IO

// 操作可以进行时，该函数返回，并将对应的

// SelectionKey 加入 selected-key

// set

System.out.print("独自等待.");

continue;

}

// 取得迭代器.selectedKeys()中包含了每个准备好某一I/O操作的信道的SelectionKey

// Selected-key Iterator 代表了所有通过 select() 方法监测到可以进行 IO 操作的 channel

// ，这个集合可以通过 selectedKeys() 拿到

Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();

while (keyIter.hasNext()) {

SelectionKey key = keyIter.next();

SelectionKey key1;

if (keyIter.hasNext()) {

key1 = keyIter.next();

}

try {

if (key.isAcceptable()) {

// 有客户端连接请求时

protocol.handleAccept(key);

}

if (key.isReadable()) {// 判断是否有数据发送过来

// 从客户端读取数据

protocol.handleRead(key);

}

if (key.isValid() && key.isWritable()) {// 判断是否有效及可以发送给客户端

// 客户端可写时

protocol.handleWrite(key);

}

} catch (IOException ex) {

// 出现IO异常（如客户端断开连接）时移除处理过的键

keyIter.remove();

continue;

}

// 移除处理过的键

keyIter.remove();

}

}

}

}

TCPProtocol.java读写处理接口

[html] view
plain copy

print?

package com.xy.nio;

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.java接口实现

[html] view
plain copy

print?

package com.xy.nio;

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.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 对象

SocketChannel clientChannel = ((ServerSocketChannel) key.channel())

.accept();

// 非阻塞式

clientChannel.configureBlocking(false);

// 注册到selector

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);

// 准备发送的文本

String sendString = "你好,客户端. @" + new Date().toString()

+ "，已经收到你的信息" + 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

}

}

TCPClientReadThread.java客户端：

[html] view
plain copy

print?

package com.xy.nio;

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();

}

@SuppressWarnings("static-access")

public void run() {

try {

while (selector.select() > 0) {//select()方法只能使用一次，用了之后就会自动删除,每个连接到服务器的选择器都是独立的

// 遍历每个有可用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();

}

}

}

TCPClient.java

[html] view
plain copy

print?

package com.xy.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.Scanner;

public class TCPClient {

// 信道选择器

private Selector selector;

// 与服务器通信的信道

SocketChannel socketChannel;

// 要连接的服务器Ip地址

private String hostIp;

// 要连接的远程服务器在监听的端口

private int hostListenningPort;

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-8"));

socketChannel.write(writeBuffer);

}

static TCPClient client;

static boolean mFlag = true;

public static void main(String[] args) throws IOException {

client = new TCPClient("10.81.36.193", 1978);

new Thread() {

@Override

public void run() {

try {

client.sendMsg("你好!Nio!醉里挑灯看剑,梦回吹角连营");

while (mFlag) {

Scanner scan = new Scanner(System.in);//键盘输入数据

String string = scan.next();

client.sendMsg(string);

}

} catch (IOException e) {

mFlag = false;

e.printStackTrace();

} finally {

mFlag = false;

}

super.run();

}

}.start();

}

}