标签：oneReactor java selector IOException key import SelectionKey
// Reactor線程
package com.luban.oneReactor;
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;
import java.util.Set;
public class TCPReactor implements Runnable {
private final ServerSocketChannel ssc;
private final Selector selector;
public TCPReactor(int port) throws IOException {
selector = Selector.open();
ssc = ServerSocketChannel.open();
InetSocketAddress addr = new InetSocketAddress(port);
ssc.socket().bind(addr); // 在ServerSocketChannel綁定監聽端口
ssc.configureBlocking(false); // 設置ServerSocketChannel為非阻塞
SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); // ServerSocketChannel向selector註冊一個OP_ACCEPT事件，然後返回該通道的key
sk.attach(new Acceptor(selector, ssc)); // 給定key一個附加的Acceptor對象
}
@Override
public void run() {
while (!Thread.interrupted()) { // 在線程被中斷前持續運行
System.out.println("Waiting for new event on port: " + ssc.socket().getLocalPort() + "...");
try {
if (selector.select() == 0) // 若沒有事件就緒則不往下執行
continue;
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 取得所有已就緒事件的key集合
Iterator<SelectionKey> it = selectedKeys.iterator();
while (it.hasNext()) {
dispatch((it.next())); // 根據事件的key進行調度
it.remove();
}
}
}
/*
* name: dispatch(SelectionKey key)
* description: 調度方法，根據事件綁定的對象開新線程
*/
private void dispatch(SelectionKey key) {
Runnable r = (Runnable) (key.attachment()); // 根據事件之key綁定的對象開新線程
if (r != null)
r.run();
}
}
package com.luban.oneReactor;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
public class Client {
/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
String hostname="127.0.0.1";
int port = 1333;
//String hostname="127.0.0.1";
//int port=1333;
try {
Socket client = new Socket(hostname, port); // 連接至目的地
System.out.println("連接至目的地:"+ hostname);
PrintWriter out = new PrintWriter(client.getOutputStream());
BufferedReader in = new BufferedReader(new InputStreamReader(client.getInputStream()));
BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in));
String input;
while((input=stdIn.readLine()) != null) { // 讀取輸入
out.println(input); // 發送輸入的字符串
out.flush(); // 強制將緩衝區內的數據輸出
if(input.equals("exit"))
{
break;
}
System.out.println("server: "+in.readLine());
}
client.close();
System.out.println("client stop.");
} catch (UnknownHostException e) {
// TODO Auto-generated catch block
System.err.println("Don't know about host: " + hostname);
} catch (IOException e) {
// TODO Auto-generated catch block
System.err.println("Couldn't get I/O for the socket connection");
}
}
}
package com.luban.oneReactor;
import java.io.IOException;
public class Main {
public static void main(String[] args) {
// TODO Auto-generated method stub
try {
TCPReactor reactor = new TCPReactor(1333);
reactor.run();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
// Handler線程
package com.luban.oneReactor;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class TCPHandler implements Runnable {
private final SelectionKey sk;
private final SocketChannel sc;
int state;
public TCPHandler(SelectionKey sk, SocketChannel sc) {
this.sk = sk;
this.sc = sc;
state = 0; // 初始狀態設定為READING
}
@Override
public void run() {
try {
if (state == 0)
read(); // 讀取網絡數據
else
send(); // 發送網絡數據
} catch (IOException e) {
System.out.println("[Warning!] A client has been closed.");
closeChannel();
}
}
private void closeChannel() {
try {
sk.cancel();
sc.close();
} catch (IOException e1) {
e1.printStackTrace();
}
}
private synchronized void read() throws IOException {
// non-blocking下不可用Readers，因為Readers不支援non-blocking
byte[] arr = new byte[1024];
ByteBuffer buf = ByteBuffer.wrap(arr);
int numBytes = sc.read(buf); // 讀取字符串
if(numBytes == -1)
{
System.out.println("[Warning!] A client has been closed.");
closeChannel();
return;
}
String str = new String(arr); // 將讀取到的byte內容轉為字符串型態
if ((str != null) && !str.equals(" ")) {
process(str); // 邏輯處理
System.out.println(sc.socket().getRemoteSocketAddress().toString()
+ " > " + str);
state = 1; // 改變狀態
sk.interestOps(SelectionKey.OP_WRITE); // 通過key改變通道註冊的事件
sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
}
}
private void send() throws IOException {
// get message from message queue
String str = "Your message has sent to "
+ sc.socket().getLocalSocketAddress().toString() + "\r\n";
ByteBuffer buf = ByteBuffer.wrap(str.getBytes()); // wrap自動把buf的position設為0，所以不需要再flip()
while (buf.hasRemaining()) {
sc.write(buf); // 回傳給client回應字符串，發送buf的position位置 到limit位置為止之間的內容
}
state = 0; // 改變狀態
sk.interestOps(SelectionKey.OP_READ); // 通過key改變通道註冊的事件
sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回
}
void process(String str) {
// do process(decode, logically process, encode)..
// ..
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
// Reactor線程
package com.luban.oneReactor;
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;
import java.util.Set;
public class TCPReactor implements Runnable {
private final ServerSocketChannel ssc;
private final Selector selector;
public TCPReactor(int port) throws IOException {
selector = Selector.open();
ssc = ServerSocketChannel.open();
InetSocketAddress addr = new InetSocketAddress(port);
ssc.socket().bind(addr); // 在ServerSocketChannel綁定監聽端口
ssc.configureBlocking(false); // 設置ServerSocketChannel為非阻塞
SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); // ServerSocketChannel向selector註冊一個OP_ACCEPT事件，然後返回該通道的key
sk.attach(new Acceptor(selector, ssc)); // 給定key一個附加的Acceptor對象
}
@Override
public void run() {
while (!Thread.interrupted()) { // 在線程被中斷前持續運行
System.out.println("Waiting for new event on port: " + ssc.socket().getLocalPort() + "...");
try {
if (selector.select() == 0) // 若沒有事件就緒則不往下執行
continue;
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 取得所有已就緒事件的key集合
Iterator<SelectionKey> it = selectedKeys.iterator();
while (it.hasNext()) {
dispatch((it.next())); // 根據事件的key進行調度
it.remove();
}
}
}
/*
* name: dispatch(SelectionKey key)
* description: 調度方法，根據事件綁定的對象開新線程
*/
private void dispatch(SelectionKey key) {
Runnable r = (Runnable) (key.attachment()); // 根據事件之key綁定的對象開新線程
if (r != null)
r.run();
}
}
标签：oneReactor,java,selector,IOException,key,import,SelectionKey 来源： https://www.cnblogs.com/mfk11/p/16251996.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。