server:
/**
* 选择器服务端
* Created by ascend on 2017/6/9 9:30.
*/
public class SelectorServer {
// public final static String REMOTE_IP = "192.168.0.44";
public final static String REMOTE_IP = "127.0.0.1";
public final static int PORT = 17531;
private static ByteBuffer bb = ByteBuffer.allocate(1024);
private static ServerSocketChannel ssc;
private static boolean closed = false; public static void main(String[] args) throws IOException {
//先确定端口号
int port = PORT;
if (args != null && args.length > 0) {
port = Integer.parseInt(args[0]);
}
//打开一个ServerSocketChannel
ssc = ServerSocketChannel.open();
//获取ServerSocketChannel绑定的Socket
ServerSocket ss = ssc.socket();
//设置ServerSocket监听的端口
ss.bind(new InetSocketAddress(port));
//设置ServerSocketChannel为非阻塞模式
ssc.configureBlocking(false);
//打开一个选择器
Selector selector = Selector.open();
//将ServerSocketChannel注册到选择器上去并监听accept事件
SelectionKey selectionKey = ssc.register(selector, SelectionKey.OP_ACCEPT); while (!closed) {
//这里会发生阻塞，等待就绪的通道,但在每次select()方法调用之间，只有一个通道就绪了。
int n = selector.select();
//没有就绪的通道则什么也不做
if (n == 0) {
continue;
}
//获取SelectionKeys上已经就绪的集合
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator(); //遍历每一个Key
while (iterator.hasNext()) {
SelectionKey sk = iterator.next();
//通道上是否有可接受的连接
if (sk.isAcceptable()) {
ServerSocketChannel sscTmp = (ServerSocketChannel) sk.channel();
SocketChannel sc = sscTmp.accept(); // accept()方法会一直阻塞到有新连接到达。
sc.configureBlocking(false);
sc.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
} else if (sk.isReadable()) { //通道上是否有数据可读
try {
readDataFromSocket(sk);
} catch (IOException e) {
sk.cancel();
continue;
}
}
if (sk.isWritable()) { //测试写入数据，若写入失败在会自动取消注册该键
try {
writeDataToSocket(sk);
} catch (IOException e) {
sk.cancel();
continue;
}
}
//必须在处理完通道时自己移除。下次该通道变成就绪时，Selector会再次将其放入已选择键集中。
iterator.remove();
}//. end of while } } /**
* 发送测试数据包，若失败则认为该socket失效
*
* @param sk SelectionKey
* @throws IOException IOException
*/
private static void writeDataToSocket(SelectionKey sk) throws IOException {
SocketChannel sc = (SocketChannel) sk.channel();
bb.clear();
String str = "server data";
bb.put(str.getBytes());
while (bb.hasRemaining()) {
sc.write(bb);
}
} /**
* 从通道中读取数据
*
* @param sk SelectionKey
* @throws IOException IOException
*/
private static void readDataFromSocket(SelectionKey sk) throws IOException {
SocketChannel sc = (SocketChannel) sk.channel();
bb.clear();
List<Byte> list = new ArrayList<>();
while (sc.read(bb) > 0) {
bb.flip();
while (bb.hasRemaining()) {
list.add(bb.get());
}
bb.clear();
}
byte[] bytes = new byte[list.size()];
for (int i = 0; i < bytes.length; i++) {
bytes[i] = list.get(i);
}
String s = (new String(bytes)).trim();
if (!s.isEmpty()) {
if ("exit".equals(s)){
ssc.close();
closed = true;
}
System.out.println("服务器收到：" + s);
}
} }
client:
/**
*
* Created by ascend on 2017/6/13 10:36.
*/
public class Client { @org.junit.Test
public void test(){
Socket socket = new Socket();
try {
socket.connect(new InetSocketAddress(SelectorServer.REMOTE_IP,SelectorServer.PORT));
DataOutputStream out = new DataOutputStream(socket.getOutputStream());
out.write("exit".getBytes());
out.flush();
out.close();
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
} public static void main(String[] args) {
new Thread(new ClientThread()).start();
} public void checkStatus(String input){
if ("exit".equals(input.trim())) {
System.out.println("系统即将退出，bye~~");
System.exit(0);
}
} } class ClientThread implements Runnable {
private SocketChannel sc;
private boolean isConnected = false;
Client client = new Client(); public ClientThread(){
try {
sc = SocketChannel.open();
sc.configureBlocking(false);
sc.connect(new InetSocketAddress(SelectorServer.REMOTE_IP,SelectorServer.PORT));
while (!sc.finishConnect()) {
System.out.println("同" + SelectorServer.REMOTE_IP + "的连接正在建立，请稍等！");
Thread.sleep(10);
}
System.out.println("连接已建立，待写入内容至指定ip+端口！时间为" + System.currentTimeMillis());
} catch (IOException | InterruptedException e) {
e.printStackTrace();
}
} @Override
public void run() {
try {
while (true){
Scanner scanner = new Scanner(System.in);
System.out.print("请输入要发送的内容：");
String writeStr = scanner.nextLine();
client.checkStatus(writeStr);
ByteBuffer bb = ByteBuffer.allocate(writeStr.length());
bb.put(writeStr.getBytes());
bb.flip(); // 写缓冲区的数据之前一定要先反转(flip)
while (bb.hasRemaining()){
sc.write(bb);
}
bb.clear();
}
} catch (IOException e) {
e.printStackTrace();
if (Objects.nonNull(sc)) {
try {
sc.close();
} catch (IOException e1) {
e1.printStackTrace();
}
}
}finally {
if (Objects.nonNull(sc)) {
try {
sc.close();
} catch (IOException e1) {
e1.printStackTrace();
}
}
}
}
}