NIO的callback调用方式

1.消耗者

1. public class CallbackClient {
2.     public static void main(String[] args) {
3.         try {
4.             SocketChannel socketChannel = SocketChannel.open();
5.             socketChannel.connect(new InetSocketAddress("127.0.0.1", 8000));
7.             ByteBuffer writeBuffer = ByteBuffer.allocate(32);
8.             ByteBuffer readBuffer = ByteBuffer.allocate(32);
10.             getMessage(readBuffer, socketChannel);
11.             sendRandomInt(writeBuffer, socketChannel, 1000);
12.             getMessage(readBuffer, socketChannel);
14.             try {
15.                 Thread.sleep(5000);
16.             } catch (InterruptedException e) {
17.                 e.printStackTrace();
18.             }
20.             sendRandomInt(writeBuffer, socketChannel, 10);
21.             getMessage(readBuffer, socketChannel);
23.             socketChannel.close();
24.         } catch (IOException e) {
25.         }
26.     }
28.     public static void sendRandomInt(ByteBuffer writeBuffer, SocketChannel socketChannel, int bound) {
29.         Random r = new Random();
30.         int d = 0;
32.         d = r.nextInt(bound);
33.         if (d == 0)
34.             d = 1;
35.         System.out.println(d);
36.         writeBuffer.clear();
37.         writeBuffer.put(String.valueOf(d).getBytes());
38.         writeBuffer.flip();
39.         try {
40.             socketChannel.write(writeBuffer);
41.         } catch (IOException e) {
42.             e.printStackTrace();
43.         }
44.     }
46.     public static void getMessage(ByteBuffer readBuffer, SocketChannel socketChannel) {
47.         readBuffer.clear();
48.         byte[] buf = new byte[16];
49.         try {
50.             socketChannel.read(readBuffer);
51.         } catch (IOException e) {
52.         }
53.         readBuffer.flip();
54.         readBuffer.get(buf, 0, readBuffer.remaining());
55.         System.out.println(new String(buf));
56.     }
57. }

复制代码

2.服务提供者

1. package com.example.demo.callback;
3. import java.io.IOException;
4. import java.net.InetSocketAddress;
5. import java.nio.ByteBuffer;
6. import java.nio.channels.SelectionKey;
7. import java.nio.channels.Selector;
8. import java.nio.channels.ServerSocketChannel;
9. import java.nio.channels.SocketChannel;
10. import java.nio.charset.StandardCharsets;
11. import java.util.Iterator;
12. import java.util.Set;
14. public class NioServer {
16.     public static void main(String[] args) throws IOException {
17.         // 打开服务器套接字通道
18.         ServerSocketChannel serverSocket = ServerSocketChannel.open();
19.         serverSocket.configureBlocking(false);
20.         serverSocket.socket().bind(new InetSocketAddress(8000));
22.         // 打开多路复用器
23.         Selector selector = Selector.open();
25.         // 注册服务器通道到多路复用器上，并监听接入事件
26.         serverSocket.register(selector, SelectionKey.OP_ACCEPT);
28.         final ByteBuffer buffer = ByteBuffer.allocate(1024);
30.         while (true) {
31.             // 非阻塞地等待注册的通道事件
32.             selector.select();
34.             // 获取发生事件的selectionKey集合
35.             Set<SelectionKey> selectedKeys = selector.selectedKeys();
36.             Iterator<SelectionKey> it = selectedKeys.iterator();
38.             // 遍历所有发生事件的selectionKey
39.             while (it.hasNext()) {
40.                 SelectionKey key = it.next();
41.                 it.remove();
43.                 // 处理接入请求
44.                 if (key.isAcceptable()) {
45.                        
46.                     ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
47.                     SocketChannel socketChannel = ssc.accept();
48.                     socketChannel.configureBlocking(false);
49.                     SelectionKey newKey = socketChannel.register(selector, SelectionKey.OP_WRITE, ByteBuffer.allocate(1024));
50.                     
51.                     //添加后可使用处理方法2处理
52.                                         CommonClient client = new CommonClient(socketChannel, newKey);
53.                     newKey.attach(client);
54.                 }
56.                 // 处理读事件
57.                 if (key.isReadable()) {
58.                     SocketChannel socketChannel = (SocketChannel) key.channel();
59.                     buffer.clear();
60.                     while (socketChannel.read(buffer) > 0) {
61.                         buffer.flip();
62.                         String receivedMessage = StandardCharsets.UTF_8.decode(buffer).toString();
63.                         handleReceivedMessage(socketChannel, receivedMessage);
64.                         buffer.clear();
65.                     }
67.                                         //处理方法2
68.                                          CommonClient client = (CommonClient) key.attachment();
69.                      client.onRead();
70.                 }
72.                 // 处理写事件
73.                 if (key.isWritable()) {
74.                                                 //处理方法1可以仿照方法2的格式写
75.                                                 //处理方法2
76.                         CommonClient client = (CommonClient) key.attachment();
77.                         client.onWrite();
78.                     }
79.             }
80.         }
81.     }
83.     // 回调函数，处理接收到的数据
84.     private static void handleReceivedMessage(SocketChannel socketChannel, String message) throws IOException {
85.         System.out.println("Received message: " + message);
86.         // 回复客户端
87.         socketChannel.write(ByteBuffer.wrap("Server received the message".getBytes(StandardCharsets.UTF_8)));
88.     }
89. }

复制代码

1. public class CommonClient {
2.         private SocketChannel clientSocket;
3.         private ByteBuffer recvBuffer;
4.         private SelectionKey key;
5.         private Callback callback;
7.         private String msg;
10.         public CommonClient(SocketChannel clientSocket, SelectionKey key) {
11.             this.clientSocket = clientSocket;
12.             this.key = key;
13.             recvBuffer = ByteBuffer.allocate(8);
15.             try {
16.                 this.clientSocket.configureBlocking(false);
17.                 key.interestOps(SelectionKey.OP_WRITE);
18.             } catch (IOException e) {
19.             }
20.         }
22.         public void close() {
23.             try {
24.                 clientSocket.close();
25.                 key.cancel();
26.             }
27.             catch (IOException e){};
28.         }
30.         // an rpc to notify client to send a number
31.         public void sendMessage(String msg, Callback cback)  {
32.             this.callback = cback;
34.             try {
35.                 try {
36.                     recvBuffer.clear();
37.                     recvBuffer.put(msg.getBytes());
38.                     recvBuffer.flip();
39.                     clientSocket.write(recvBuffer);
41.                     key.interestOps(SelectionKey.OP_READ);
42.                 } catch (IOException e) {
43.                     e.printStackTrace();
44.                 }
45.             }
46.             catch (Exception e) {
47.             }
48.         }
50.         // when key is writable, resume the fiber to continue
51.         // to write.
52.         public void onWrite() {
53.             sendMessage("divident", new Callback() {
54.                 @Override
55.                 public void onSucceed(int data) {
56.                     int a = data;
57.                     sendMessage("divisor", new Callback() {
58.                         @Override
59.                         public void onSucceed(int data) {
60.                             int b = data;
62.                             sendMessage(String.valueOf(a / b), null);
63.                         }
64.                     });
65.                 }
66.             });
67.         }
69.         public void onRead() {
70.             int res = 0;
71.             try {
72.                 try {
73.                     recvBuffer.clear();
75.                     // read may fail even SelectionKey is readable
76.                     // when read fails, the fiber should suspend, waiting for next
77.                     // time the key is ready.
78.                     int n = clientSocket.read(recvBuffer);
79.                     while (n == 0) {
80.                         n = clientSocket.read(recvBuffer);
81.                     }
83.                     if (n == -1) {
84.                         close();
85.                         return;
86.                     }
88.                     System.out.println("received " + n + " bytes from client");
89.                 } catch (IOException e) {
90.                     e.printStackTrace();
91.                 }
93.                 recvBuffer.flip();
94.                 res = getInt(recvBuffer);
96.                 // when read ends, we are no longer interested in reading,
97.                 // but in writing.
98.                 key.interestOps(SelectionKey.OP_WRITE);
99.             } catch (Exception e) {
100.             }
102.             this.callback.onSucceed(res);
103.         }
105.         public int getInt(ByteBuffer buf) {
106.             int r = 0;
107.             while (buf.hasRemaining()) {
108.                 r *= 10;
109.                 r += buf.get() - '0';
110.             }
111.             return r;
112.         }
113.     }

复制代码

1.     public interface Callback {
2.         public void onSucceed(int data);
3.     }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！qidao123.com:ToB企服之家，中国第一个企服评测及软件市场,开放入驻,技术点评得现金