您的位置: 首页  >  文章  >  Java网络编程-IO多路复用(多线程)

Java网络编程-IO多路复用(多线程)

分类: 文章 • 2024-04-10 20:46:54

1. 简述

IO复用网路模型中的角色Handler，将以状态模式(State Pattern)实现，

并且将Handler接收完网络消息后的业务逻辑处理交由工作线程实现，所有的工作线程以线程池来维护。


2. 网络模型


Java网络编程-IO多路复用(多线程)



3. 代码实现


[TCPReactor.Java]

[java] view plain copy
 Java网络编程-IO多路复用(多线程)Java网络编程-IO多路复用(多线程)
  1. // Reactor線程  
  2. package server;  
  3.   
  4. import java.io.IOException;  
  5. import java.net.InetSocketAddress;  
  6. import java.nio.channels.SelectionKey;  
  7. import java.nio.channels.Selector;  
  8. import java.nio.channels.ServerSocketChannel;  
  9. import java.util.Iterator;  
  10. import java.util.Set;  
  11.   
  12. public class TCPReactor implements Runnable {  
  13.   
  14.     private final ServerSocketChannel ssc;  
  15.     private final Selector selector;  
  16.   
  17.     public TCPReactor(int port) throws IOException {  
  18.         selector = Selector.open();  
  19.         ssc = ServerSocketChannel.open();  
  20.         InetSocketAddress addr = new InetSocketAddress(port);  
  21.         ssc.socket().bind(addr); // 在ServerSocketChannel綁定監聽端口  
  22.         ssc.configureBlocking(false); // 設置ServerSocketChannel為非阻塞  
  23.         SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); // ServerSocketChannel向selector註冊一個OP_ACCEPT事件，然後返回該通道的key  
  24.         sk.attach(new Acceptor(selector, ssc)); // 給定key一個附加的Acceptor對象  
  25.     }  
  26.   
  27.     @Override  
  28.     public void run() {  
  29.         while (!Thread.interrupted()) { // 在線程被中斷前持續運行  
  30.             System.out.println("Waiting for new event on port: " + ssc.socket().getLocalPort() + "...");  
  31.             try {  
  32.                 if (selector.select() == 0) // 若沒有事件就緒則不往下執行  
  33.                     continue;  
  34.             } catch (IOException e) {  
  35.                 // TODO Auto-generated catch block  
  36.                 e.printStackTrace();  
  37.             }  
  38.             Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 取得所有已就緒事件的key集合  
  39.             Iterator<SelectionKey> it = selectedKeys.iterator();  
  40.             while (it.hasNext()) {  
  41.                 dispatch((SelectionKey) (it.next())); // 根據事件的key進行調度  
  42.                 it.remove();  
  43.             }  
  44.         }  
  45.     }  
  46.   
  47.     /* 
  48.      * name: dispatch(SelectionKey key) 
  49.      * description: 調度方法，根據事件綁定的對象開新線程 
  50.      */  
  51.     private void dispatch(SelectionKey key) {  
  52.         Runnable r = (Runnable) (key.attachment()); // 根據事件之key綁定的對象開新線程  
  53.         if (r != null)  
  54.             r.run();  
  55.     }  
  56.   
  57. }  


[Acceptor.java]

[java] view plain copy
 Java网络编程-IO多路复用(多线程)Java网络编程-IO多路复用(多线程)
  1. // 接受連線請求線程  
  2. package server;  
  3.   
  4. import java.io.IOException;  
  5. import java.nio.channels.SelectionKey;  
  6. import java.nio.channels.Selector;  
  7. import java.nio.channels.ServerSocketChannel;  
  8. import java.nio.channels.SocketChannel;  
  9.   
  10. public class Acceptor implements Runnable {  
  11.   
  12.     private final ServerSocketChannel ssc;  
  13.     private final Selector selector;  
  14.       
  15.     public Acceptor(Selector selector, ServerSocketChannel ssc) {  
  16.         this.ssc=ssc;  
  17.         this.selector=selector;  
  18.     }  
  19.       
  20.     @Override  
  21.     public void run() {  
  22.         try {  
  23.             SocketChannel sc= ssc.accept(); // 接受client連線請求  
  24.             System.out.println(sc.socket().getRemoteSocketAddress().toString() + " is connected.");  
  25.               
  26.             if(sc!=null) {  
  27.                 sc.configureBlocking(false); // 設置為非阻塞  
  28.                 SelectionKey sk = sc.register(selector, SelectionKey.OP_READ); // SocketChannel向selector註冊一個OP_READ事件，然後返回該通道的key  
  29.                 selector.wakeup(); // 使一個阻塞住的selector操作立即返回  
  30.                 sk.attach(new TCPHandler(sk, sc)); // 給定key一個附加的TCPHandler對象  
  31.             }  
  32.               
  33.         } catch (IOException e) {  
  34.             // TODO Auto-generated catch block  
  35.             e.printStackTrace();  
  36.         }  
  37.     }  
  38.   
  39.       
  40. }  


[TCPHandler.java]

[java] view plain copy
 Java网络编程-IO多路复用(多线程)Java网络编程-IO多路复用(多线程)
  1. // Handler線程  
  2. package server;  
  3.   
  4. import java.io.IOException;  
  5. import java.nio.channels.SelectionKey;  
  6. import java.nio.channels.SocketChannel;  
  7. import java.util.concurrent.LinkedBlockingQueue;  
  8. import java.util.concurrent.ThreadPoolExecutor;  
  9. import java.util.concurrent.TimeUnit;  
  10.   
  11. public class TCPHandler implements Runnable {  
  12.   
  13.     private final SelectionKey sk;  
  14.     private final SocketChannel sc;  
  15.     private static final int THREAD_COUNTING = 10;  
  16.     private static ThreadPoolExecutor pool = new ThreadPoolExecutor(  
  17.             THREAD_COUNTING, THREAD_COUNTING, 10, TimeUnit.SECONDS,  
  18.             new LinkedBlockingQueue<Runnable>()); // 線程池  
  19.   
  20.     HandlerState state; // 以狀態模式實現Handler  
  21.   
  22.     public TCPHandler(SelectionKey sk, SocketChannel sc) {  
  23.         this.sk = sk;  
  24.         this.sc = sc;  
  25.         state = new ReadState(); // 初始狀態設定為READING  
  26.         pool.setMaximumPoolSize(32); // 設置線程池最大線程數  
  27.     }  
  28.   
  29.     @Override  
  30.     public void run() {  
  31.         try {  
  32.             state.handle(this, sk, sc, pool);  
  33.               
  34.         } catch (IOException e) {  
  35.             System.out.println("[Warning!] A client has been closed.");  
  36.             closeChannel();  
  37.         }  
  38.     }  
  39.       
  40.     public void closeChannel() {  
  41.         try {  
  42.             sk.cancel();  
  43.             sc.close();  
  44.         } catch (IOException e1) {  
  45.             e1.printStackTrace();  
  46.         }  
  47.     }  
  48.   
  49.     public void setState(HandlerState state) {  
  50.         this.state = state;  
  51.     }  
  52. }  



Handler要以状态模式实现还需要定义状态接口(interface)和状态实现类(ConcreteState Class)

[HandlerState.java]

[java] view plain copy
 Java网络编程-IO多路复用(多线程)Java网络编程-IO多路复用(多线程)
  1. package server;  
  2.   
  3. import java.io.IOException;  
  4. import java.nio.channels.SelectionKey;  
  5. import java.nio.channels.SocketChannel;  
  6. import java.util.concurrent.ThreadPoolExecutor;  
  7.   
  8. public interface HandlerState {  
  9.   
  10.     public void changeState(TCPHandler h);  
  11.   
  12.     public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,  
  13.             ThreadPoolExecutor pool) throws IOException ;  
  14. }  


[ReadState.java]

[java] view plain copy
 Java网络编程-IO多路复用(多线程)Java网络编程-IO多路复用(多线程)
  1. package server;  
  2.   
  3. import java.io.IOException;  
  4. import java.nio.ByteBuffer;  
  5. import java.nio.channels.SelectionKey;  
  6. import java.nio.channels.SocketChannel;  
  7. import java.util.concurrent.ThreadPoolExecutor;  
  8.   
  9. public class ReadState implements HandlerState{  
  10.   
  11.     private SelectionKey sk;  
  12.       
  13.     public ReadState() {  
  14.     }  
  15.       
  16.     @Override  
  17.     public void changeState(TCPHandler h) {  
  18.         // TODO Auto-generated method stub  
  19.         h.setState(new WorkState());  
  20.     }  
  21.   
  22.     @Override  
  23.     public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,  
  24.             ThreadPoolExecutor pool) throws IOException { // read()  
  25.         this.sk = sk;  
  26.         // non-blocking下不可用Readers，因為Readers不支援non-blocking  
  27.         byte[] arr = new byte[1024];  
  28.         ByteBuffer buf = ByteBuffer.wrap(arr);  
  29.           
  30.         int numBytes = sc.read(buf); // 讀取字符串  
  31.         if(numBytes == -1)  
  32.         {  
  33.             System.out.println("[Warning!] A client has been closed.");  
  34.             h.closeChannel();  
  35.             return;  
  36.         }  
  37.         String str = new String(arr); // 將讀取到的byte內容轉為字符串型態  
  38.         if ((str != null) && !str.equals(" ")) {  
  39.             h.setState(new WorkState()); // 改變狀態(READING->WORKING)  
  40.             pool.execute(new WorkerThread(h, str)); // do process in worker thread  
  41.             System.out.println(sc.socket().getRemoteSocketAddress().toString()  
  42.                     + " > " + str);  
  43.         }  
  44.           
  45.     }  
  46.       
  47.     /* 
  48.      * 執行邏輯處理之函數 
  49.      */  
  50.     synchronized void process(TCPHandler h, String str) {  
  51.         // do process(decode, logically process, encode)..  
  52.         // ..  
  53.         h.setState(new WriteState()); // 改變狀態(WORKING->SENDING)  
  54.         this.sk.interestOps(SelectionKey.OP_WRITE); // 通過key改變通道註冊的事件  
  55.         this.sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回  
  56.     }  
  57.   
  58.     /* 
  59.      * 工作者線程 
  60.      */  
  61.     class WorkerThread implements Runnable {  
  62.   
  63.         TCPHandler h;  
  64.         String str;  
  65.   
  66.         public WorkerThread(TCPHandler h, String str) {  
  67.             this.h = h;  
  68.             this.str=str;  
  69.         }  
  70.   
  71.         @Override  
  72.         public void run() {  
  73.             process(h, str);  
  74.         }  
  75.   
  76.     }  
  77. }  


[WorkState.java]

[java] view plain copy
 Java网络编程-IO多路复用(多线程)Java网络编程-IO多路复用(多线程)
  1. package server;  
  2.   
  3. import java.io.IOException;  
  4. import java.nio.channels.SelectionKey;  
  5. import java.nio.channels.SocketChannel;  
  6. import java.util.concurrent.ThreadPoolExecutor;  
  7.   
  8. public class WorkState implements HandlerState {  
  9.   
  10.     public WorkState() {  
  11.     }  
  12.       
  13.     @Override  
  14.     public void changeState(TCPHandler h) {  
  15.         // TODO Auto-generated method stub  
  16.         h.setState(new WriteState());  
  17.     }  
  18.   
  19.     @Override  
  20.     public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,  
  21.             ThreadPoolExecutor pool) throws IOException {  
  22.         // TODO Auto-generated method stub  
  23.           
  24.     }  
  25.   
  26. }  


[WriteState.java]

[java] view plain copy
 Java网络编程-IO多路复用(多线程)Java网络编程-IO多路复用(多线程)
  1. package server;  
  2.   
  3. import java.io.IOException;  
  4. import java.nio.ByteBuffer;  
  5. import java.nio.channels.SelectionKey;  
  6. import java.nio.channels.SocketChannel;  
  7. import java.util.concurrent.ThreadPoolExecutor;  
  8.   
  9. public class WriteState implements HandlerState{  
  10.   
  11.     public WriteState() {  
  12.     }  
  13.       
  14.     @Override  
  15.     public void changeState(TCPHandler h) {  
  16.         // TODO Auto-generated method stub  
  17.         h.setState(new ReadState());  
  18.     }  
  19.   
  20.     @Override  
  21.     public void handle(TCPHandler h, SelectionKey sk, SocketChannel sc,  
  22.             ThreadPoolExecutor pool) throws IOException { // send()  
  23.         // get message from message queue  
  24.           
  25.         String str = "Your message has sent to "  
  26.                 + sc.socket().getLocalSocketAddress().toString() + "\r\n";  
  27.         ByteBuffer buf = ByteBuffer.wrap(str.getBytes()); // wrap自動把buf的position設為0，所以不需要再flip()  
  28.   
  29.         while (buf.hasRemaining()) {  
  30.             sc.write(buf); // 回傳給client回應字符串，發送buf的position位置 到limit位置為止之間的內容  
  31.         }  
  32.           
  33.         h.setState(new ReadState()); // 改變狀態(SENDING->READING)  
  34.         sk.interestOps(SelectionKey.OP_READ); // 通過key改變通道註冊的事件  
  35.         sk.selector().wakeup(); // 使一個阻塞住的selector操作立即返回  
  36.     }  
  37. }  



最後是主程序

[Main.java]

[java] view plain copy
 Java网络编程-IO多路复用(多线程)Java网络编程-IO多路复用(多线程)
  1. package server;  
  2.   
  3. import java.io.IOException;  
  4.   
  5. public class Main {  
  6.   
  7.       
  8.     public static void main(String[] args) {  
  9.         // TODO Auto-generated method stub  
  10.         try {  
  11.             TCPReactor reactor = new TCPReactor(1333);  
  12.             reactor.run();  
  13.         } catch (IOException e) {  
  14.             // TODO Auto-generated catch block  
  15.             e.printStackTrace();  
  16.         }  
  17.     }  
  18.   
  19. }  

相关推荐