javanio--采用Selector实现Socket通信

server:

1 /**

2 * 选择器服务端

3 * Created by ascend on 2017/6/9 9:30. 4 */

5 public class SelectorServer {

6 // public final static String REMOTE_IP = \"192.168.0.44\"; 7 public final static String REMOTE_IP = \"127.0.0.1\"; 8 public final static int PORT = 17531;

9 private static ByteBuffer bb = ByteBuffer.allocate(1024); 10 private static ServerSocketChannel ssc; 11 private static boolean closed = false; 12

13 public static void main(String[] args) throws IOException { 14 //先确定端⼝号 15 int port = PORT;

16 if (args != null && args.length > 0) { 17 port = Integer.parseInt(args[0]); 18 }

19 //打开⼀个ServerSocketChannel 20 ssc = ServerSocketChannel.open();

21 //获取ServerSocketChannel绑定的Socket 22 ServerSocket ss = ssc.socket(); 23 //设置ServerSocket监听的端⼝

24 ss.bind(new InetSocketAddress(port)); 25 //设置ServerSocketChannel为⾮阻塞模式 26 ssc.configureBlocking(false); 27 //打开⼀个选择器

28 Selector selector = Selector.open();

29 //将ServerSocketChannel注册到选择器上去并监听accept事件

30 SelectionKey selectionKey = ssc.register(selector, SelectionKey.OP_ACCEPT); 31 32

33 while (!closed) {

34 //这⾥会发⽣阻塞，等待就绪的通道,但在每次select()⽅法调⽤之间，只有⼀个通道就绪了。 35 int n = selector.select();

36 //没有就绪的通道则什么也不做 37 if (n == 0) { 38 continue; 39 }

40 //获取SelectionKeys上已经就绪的集合

41 Iterator iterator = selector.selectedKeys().iterator(); 42

43 //遍历每⼀个Key

44 while (iterator.hasNext()) {

45 SelectionKey sk = iterator.next(); 46 //通道上是否有可接受的连接 47 if (sk.isAcceptable()) {

48 ServerSocketChannel sscTmp = (ServerSocketChannel) sk.channel();

49 SocketChannel sc = sscTmp.accept(); // accept()⽅法会⼀直阻塞到有新连接到达。 50 sc.configureBlocking(false);

51 sc.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE); 52 } else if (sk.isReadable()) { //通道上是否有数据可读 53 try {

54 readDataFromSocket(sk); 55 } catch (IOException e) { 56 sk.cancel(); 57 continue; 58 } 59 }

60 if (sk.isWritable()) { //测试写⼊数据，若写⼊失败在会⾃动取消注册该键 61 try {

62 writeDataToSocket(sk); 63 } catch (IOException e) { 64 sk.cancel(); 65 continue; 66 } 67 }

68 //必须在处理完通道时⾃⼰移除。下次该通道变成就绪时，Selector会再次将其放⼊已选择键集中。 69 iterator.remove(); 70 }//. end of while 71

72 } 73 74 } 75 76 77

78 /**

79 * 发送测试数据包，若失败则认为该socket失效 80 *

81 * @param sk SelectionKey

82 * @throws IOException IOException 83 */

84 private static void writeDataToSocket(SelectionKey sk) throws IOException { 85 SocketChannel sc = (SocketChannel) sk.channel(); 86 bb.clear();

87 String str = \"server data\"; 88 bb.put(str.getBytes());

89 while (bb.hasRemaining()) { 90 sc.write(bb); 91 } 92 } 93

94 /**

95 * 从通道中读取数据 96 *

97 * @param sk SelectionKey

98 * @throws IOException IOException 99 */

100 private static void readDataFromSocket(SelectionKey sk) throws IOException {101 SocketChannel sc = (SocketChannel) sk.channel();102 bb.clear();

103 List list = new ArrayList<>();104 while (sc.read(bb) > 0) {105 bb.flip();

106 while (bb.hasRemaining()) {107 list.add(bb.get());108 }

109 bb.clear();110 }

111 byte[] bytes = new byte[list.size()];112 for (int i = 0; i < bytes.length; i++) {113 bytes[i] = list.get(i);114 }

115 String s = (new String(bytes)).trim();116 if (!s.isEmpty()) {

117 if (\"exit\".equals(s)){118 ssc.close();119 closed = true;120 }

121 System.out.println(\"服务器收到：\" + s);122 }123 }124 125 }

client:

1 /** 2 *

3 * Created by ascend on 2017/6/13 10:36. 4 */

5 public class Client { 6

7 @org.junit.Test 8 public void test(){

9 Socket socket = new Socket();10 try {

11 socket.connect(new InetSocketAddress(SelectorServer.REMOTE_IP,SelectorServer.PORT));12 DataOutputStream out = new DataOutputStream(socket.getOutputStream());13 out.write(\"exit\".getBytes());14 out.flush();15 out.close();16 socket.close();

17 } catch (IOException e) {18 e.printStackTrace();19 }20 }21

22 public static void main(String[] args) {23 new Thread(new ClientThread()).start();24 }25

26 public void checkStatus(String input){27 if (\"exit\".equals(input.trim())) {

28 System.out.println(\"系统即将退出，bye~~\");29 System.exit(0);30 }31 }32 33 34 }

35

36 class ClientThread implements Runnable {37 private SocketChannel sc;

38 private boolean isConnected = false;39 Client client = new Client();40

41 public ClientThread(){42 try {

43 sc = SocketChannel.open();44 sc.configureBlocking(false);

45 sc.connect(new InetSocketAddress(SelectorServer.REMOTE_IP,SelectorServer.PORT));46 while (!sc.finishConnect()) {

47 System.out.println(\"同\" + SelectorServer.REMOTE_IP + \"的连接正在建⽴，请稍等！\");48 Thread.sleep(10);49 }

50 System.out.println(\"连接已建⽴，待写⼊内容⾄指定ip+端⼝！时间为\" + System.currentTimeMillis());51 } catch (IOException | InterruptedException e) {52 e.printStackTrace();53 }54 }55

56 @Override

57 public void run() {58 try {

59 while (true){

60 Scanner scanner = new Scanner(System.in);61 System.out.print(\"请输⼊要发送的内容：\");62 String writeStr = scanner.nextLine();63 client.checkStatus(writeStr);

64 ByteBuffer bb = ByteBuffer.allocate(writeStr.length());65 bb.put(writeStr.getBytes());

66 bb.flip(); // 写缓冲区的数据之前⼀定要先反转(flip)67 while (bb.hasRemaining()){68 sc.write(bb);69 }

70 bb.clear();71 }

72 } catch (IOException e) {73 e.printStackTrace();

74 if (Objects.nonNull(sc)) {75 try {

76 sc.close();

77 } catch (IOException e1) {78 e1.printStackTrace();79 }80 }

81 }finally {

82 if (Objects.nonNull(sc)) {83 try {

84 sc.close();

85 } catch (IOException e1) {86 e1.printStackTrace();87 }88 }89 }90 }91 }