本文参考至：http://ifeve.com/socket-channel/
在NIO系列4中，采用了SocketChannel作为案例讲解Selector，当时我确实看不太懂。现在写一下SocketChannel的理解：
Java NIO中的SocketChannel是一个连接到TCP网络套接字的通道。可以通过以下2种方式创建SocketChannel：
1、打开一个SocketChannel并连接到互联网上的某台服务器。
2、一个新连接到达ServerSocketChannel时，会创建一个SocketChannel。
这里简要的介绍一下Channel的读写数据的方法，其实对于所有的Channel读写数据的方法都几乎一样，都是从Buffer中读或者写到Buffer中，下面举FileChannel和SocketChannel两个例子：
Reading from a FileChannel：
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = inChannel.read(buf);
Reading from a SocketChannel：
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead = socketChannel.read(buf);
Writing to a SocketChannel：
String newData = "New String to write to file..." + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining()) {
socketchannel.write(buf);
}
Writing Data to a FileChannel:
String newData = "New String to write to file..." + System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining()) {
fileChannel.write(buf);
}
以下代码模拟了服务器和客户端：
服务器：
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.nio.charset.Charset;
import java.util.Iterator;
public class TCPServer {
private static final int bufferSize = 1024;
private static final long timeOut = 3000;// 超时时间
private static final int listenPort = 1993;// 本地监听端口
public static void main(String[] args) throws Exception {
Selector selector = Selector.open();
ServerSocketChannel listenerChannel = ServerSocketChannel.open();// 创建监听通道，专门用来监听指定的本地端口
listenerChannel.socket().bind(new InetSocketAddress(listenPort));// 将listenerChannel的socket绑定为本地服务器（IP+prot）绑定
listenerChannel.configureBlocking(false);
// 将选择器绑定到监听信道,只有非阻塞信道才可以注册选择器.并在注册过程中指出该信道可以进行Accept操作
listenerChannel.register(selector, SelectionKey.OP_ACCEPT);
TCPProtocolImpl protocol = new TCPProtocolImpl(bufferSize);
while (true) {
if (selector.select(timeOut) == 0) {// 监听注册的通道，当其中有注册的IO时该函数返回（3000ms没有反应返回0），操作可以进行，并添加对应的SelectorKey
System.out.println("It haven't I/O now, please wait!");
continue;
}
Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
while (keyIter.hasNext()) {
try {
SelectionKey key = keyIter.next();
if (key.isAcceptable()) {
protocol.handleAccept(key);
}
if (key.isReadable()) {
protocol.handleRead(key);
}
} catch (IOException e) {
keyIter.remove();
continue;
}
keyIter.remove();
}
}
}
}
class TCPProtocolImpl{
private int bufferSize;
public TCPProtocolImpl() {
super();
}
public TCPProtocolImpl(int bufferSize) {
super();
this.bufferSize = bufferSize;
}
public void handleAccept(SelectionKey key) throws IOException {
// 返回创建此键的通道，接受客户端建立连接的请求，并返回SocketChannel对象
SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
clientChannel.configureBlocking(false);
// 将clientChannel注册到服务端的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();
// 从客户端通道读取信息到buffer缓冲区中(并返回读到信息的字节数)
long bytesRead = clientChannel.read(buffer);
if (bytesRead == -1) {
clientChannel.close();
} else {
buffer.flip();
// 将字节转化为为UTF-8的字符串
String receivedString = Charset.forName("UTF-8").newDecoder().decode(buffer).toString();
System.out.println("接收到来自：" + clientChannel.socket().getRemoteSocketAddress() + "发来的信息：" + receivedString);
String msgSendToClient = "已接收到你的信息：" + receivedString + "正在处理中";
buffer = ByteBuffer.wrap(msgSendToClient.getBytes("UTF-8"));
clientChannel.write(buffer);
// 设置为下一次读取或是写入做准备
key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
}
}
}
客户端：
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.nio.charset.Charset;
import java.util.Scanner;
public class TCPClient {
// 通道选择器，用于管理客户端的通道
private Selector selector;
// 与服务器通信的通道
SocketChannel socketChannel;
// 要连接的服务器的IP
private String hostIp;
// 要连接的远程服务器在监听的端口
private int hostListenningPort;
static TCPClient client;
static boolean mFlag = true;
public TCPClient(String hostIp, int hostPort) throws IOException {
this.hostIp = hostIp;
this.hostListenningPort = hostPort;
init();
}
private void init() 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);
}
public static void main(String[] args) throws IOException {
client = new TCPClient("127.0.0.1", 1993);
new Thread(){
@Override
public void run(){
try{
client.sendMsg("test----~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
while(mFlag){
Scanner scan = new Scanner(System.in);
String string = scan.next();
client.sendMsg(string);
}
}catch (Exception e) {
mFlag = false;
}finally{
mFlag = false;
}
super.run();
}
}.start();
}
}
class TCPClientReadThread implements Runnable {
private Selector selector;
public TCPClientReadThread(Selector selector) {
super();
this.selector = selector;
new Thread(this).start();
}
@Override
public void run() {
try {
while (selector.select() > 0) {// select()方法只能使用一次，用了之后就会自动删除,每个连接到服务器的选择器都是独立的
// 遍历每个有IO操作Channel对应的SelectionKey
for (SelectionKey sk : selector.selectedKeys()) {
if (sk.isReadable()) {
// 使用NIO读取Channel中的数据
SocketChannel sc = (SocketChannel) sk.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
sc.read(buffer);
buffer.flip();
String receivedString = Charset.forName("UTF-8").newDecoder().decode(buffer).toString();
System.out.println("接收到来自服务器：" + sc.socket().getRemoteSocketAddress() + "的信息：" + receivedString);
sk.interestOps(SelectionKey.OP_READ);
}
selector.selectedKeys().remove(sk);
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
网友评论