自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

基于Java NIO的即時(shí)聊天服務(wù)器模型

作者：黎某人
開(kāi)發(fā) 后端
折騰了一個(gè)周，終于搞出來(lái)了一個(gè)雛形，相比于xmpp的xml，本人更喜歡json的簡(jiǎn)潔，為了防止客戶端異常斷開(kāi)等，準(zhǔn)備采用心跳檢測(cè)的機(jī)制來(lái)判斷用戶是否在線，另外還有一種方法是學(xué)習(xí)例如Tomcat等Servlet中間件的方式，設(shè)置Session周期，定時(shí)清除過(guò)期Session。

前不久自己動(dòng)手寫了一個(gè)Android的聊天工具，跟服務(wù)器的交互還是基于HTTP方式的，在一般通訊上還算湊活，但是在即時(shí)聊天的時(shí)候就有點(diǎn)惡心了，客戶端開(kāi)啟Service每隔3秒去詢問(wèn)服務(wù)器是否有自己的新消息（當(dāng)然3秒有點(diǎn)太快了），在心疼性能和流量的前提下，只能自己動(dòng)手寫個(gè)服務(wù)器，傳統(tǒng)的Socket是阻塞的，這樣的話服務(wù)器對(duì)每個(gè)Socket都需要建立一個(gè)線程來(lái)操作，資源開(kāi)銷很大，而且線程多了直接會(huì)影響服務(wù)端的性能（曾經(jīng)測(cè)試開(kāi)了3000多個(gè)線程就不讓創(chuàng)建了，所以并發(fā)數(shù)目也是有限制的），聽(tīng)說(shuō)從JDK1.5就多了個(gè)New IO，灰常不錯(cuò)的樣子，找了找相關(guān)的資料，網(wǎng)上竟然全都是最最最簡(jiǎn)單的一個(gè)demo，然后去CSDN發(fā)帖，基本上都是建議直接使用MINA框架的，這樣一來(lái)根本達(dá)不到學(xué)習(xí)NIO的目的，而且現(xiàn)在的技術(shù)也太快餐了，只知道使用前輩留下的東西，知其然不知其所以然。

折騰了一個(gè)周，終于搞出來(lái)了一個(gè)雛形，相比于xmpp的xml，本人更喜歡json的簡(jiǎn)潔，為了防止客戶端異常斷開(kāi)等，準(zhǔn)備采用心跳檢測(cè)的機(jī)制來(lái)判斷用戶是否在線，另外還有一種方法是學(xué)習(xí)例如Tomcat等Servlet中間件的方式，設(shè)置Session周期，定時(shí)清除過(guò)期Session。本Demo暫時(shí)實(shí)現(xiàn)了Session過(guò)期檢測(cè)，心跳檢測(cè)有空再搞，如果本例子在使用過(guò)程中有性能漏洞或者什么bug請(qǐng)及時(shí)通知我，謝謝。

廢話不多說(shuō)，關(guān)于NIO的SelectionKey、Selector、Channel網(wǎng)上的介紹例子都很多，直接上代碼：

JsonParser

Json的解析類，隨便封裝了下，使用的最近比較火的fastjson

  1. public class JsonParser { 
  2.      
  3.     private static JSONObject mJson; 
  4.      
  5.     public synchronized static String get(String json,String key) { 
  6.         mJson = JSON.parseObject(json); 
  7.         return mJson.getString(key); 
  8.     } 
  9. } 

Main

入口，不解釋

  1. public class Main { 
  2.  
  3.     public static void main(String... args) { 
  4.         new SeekServer().start(); 
  5.     } 
  6. } 

Log

  1. public class Log { 
  2.  
  3.     public static void i(Object obj) { 
  4.         System.out.println(obj); 
  5.     } 
  6.     public static void e(Object e) { 
  7.         System.err.println(e); 
  8.     } 
  9. } 

SeekServer：

服務(wù)器端的入口，請(qǐng)求的封裝和接收都在此類，端口暫時(shí)寫死在了代碼里，mSelector.select(TIME_OUT) > 0 目的是為了當(dāng)服務(wù)器空閑的時(shí)候（沒(méi)有任何讀寫甚至請(qǐng)求斷開(kāi)事件），循環(huán)時(shí)有個(gè)間隔時(shí)間，不然基本上相當(dāng)于while(true){//nothing}了，你懂的。

  1. public class SeekServer extends Thread{ 
  2.     private final int ACCPET_PORT = 55555; 
  3.     private final int TIME_OUT = 1000; 
  4.     private Selector mSelector = null; 
  5.     private ServerSocketChannel mSocketChannel = null; 
  6.     private ServerSocket mServerSocket = null; 
  7.     private InetSocketAddress mAddress = null; 
  8.      
  9.     public SeekServer() { 
  10.         long sign = System.currentTimeMillis(); 
  11.         try { 
  12.             mSocketChannel = ServerSocketChannel.open(); 
  13.             if(mSocketChannel == null) { 
  14.                 System.out.println("can't open server socket channel"); 
  15.             } 
  16.             mServerSocket = mSocketChannel.socket(); 
  17.             mAddress = new InetSocketAddress(ACCPET_PORT); 
  18.             mServerSocket.bind(mAddress); 
  19.             Log.i("server bind port is " + ACCPET_PORT); 
  20.             mSelector = Selector.open(); 
  21.             mSocketChannel.configureBlocking(false); 
  22.             SelectionKey key = mSocketChannel.register(mSelector, SelectionKey.OP_ACCEPT); 
  23.             key.attach(new Acceptor()); 
  24.              
  25.             //檢測(cè)Session狀態(tài) 
  26.             Looper.getInstance().loop(); 
  27.              
  28.             //開(kāi)始處理Session 
  29.             SessionProcessor.start(); 
  30.              
  31.             Log.i("Seek server startup in " + (System.currentTimeMillis() - sign) + "ms!"); 
  32.         } catch (ClosedChannelException e) { 
  33.             Log.e(e.getMessage()); 
  34.         } catch (IOException e) { 
  35.             Log.e(e.getMessage()); 
  36.         }  
  37.     } 
  38.      
  39.     public void run() { 
  40.         Log.i("server is listening..."); 
  41.         while(!Thread.interrupted()) { 
  42.             try { 
  43.                 if(mSelector.select(TIME_OUT) > 0) { 
  44.                     Set<SelectionKey> keys = mSelector.selectedKeys(); 
  45.                     Iterator<SelectionKey> iterator = keys.iterator(); 
  46.                     SelectionKey key = null; 
  47.                     while(iterator.hasNext()) { 
  48.                         key = iterator.next(); 
  49.                         Handler at = (Handler) key.attachment(); 
  50.                         if(at != null) { 
  51.                             at.exec(); 
  52.                         } 
  53.                         iterator.remove(); 
  54.                     } 
  55.                 } 
  56.             } catch (IOException e) { 
  57.                 Log.e(e.getMessage()); 
  58.             } 
  59.         } 
  60.     } 
  61.  
  62.     class Acceptor extends Handler{ 
  63.  
  64.         public void exec(){ 
  65.             try { 
  66.                 SocketChannel sc = mSocketChannel.accept(); 
  67.                 new Session(sc, mSelector); 
  68.             } catch (ClosedChannelException e) { 
  69.                 Log.e(e); 
  70.             } catch (IOException e) { 
  71.                 Log.e(e); 
  72.             } 
  73.         } 
  74.     } 
  75. } 

Handler：

只有一個(gè)抽象方法exec，Session將會(huì)繼承它。

  1. public abstract class Handler { 
  2.      
  3.     public abstract void exec(); 
  4. } 

Session：

封裝了用戶的請(qǐng)求和SelectionKey和SocketChannel，每次接收到新的請(qǐng)求時(shí)都重置它的最后活動(dòng)時(shí)間，通過(guò)狀態(tài)mState=READING or SENDING 去執(zhí)行消息的接收與發(fā)送，當(dāng)客戶端異常斷開(kāi)時(shí)則從SessionManager清除該會(huì)話。

  1. public class Session extends Handler{ 
  2.  
  3.     private SocketChannel mChannel; 
  4.     private SelectionKey  mKey; 
  5.     private ByteBuffer mRreceiveBuffer = ByteBuffer.allocate(10240);   
  6.     private Charset charset = Charset.forName("UTF-8"); 
  7.     private CharsetDecoder mDecoder = charset.newDecoder(); 
  8.     private CharsetEncoder mEncoder = charset.newEncoder(); 
  9.     private long lastPant;//最后活動(dòng)時(shí)間 
  10.     private final int TIME_OUT = 1000 * 60 * 5; //Session超時(shí)時(shí)間 
  11.     private String key; 
  12.      
  13.     private String sendData = ""; 
  14.     private String receiveData = null; 
  15.      
  16.     public static final int READING = 0,SENDING = 1; 
  17.     int mState = READING; 
  18.      
  19.     public Session(SocketChannel socket, Selector selector) throws IOException { 
  20.         this.mChannel = socket; 
  21.         mChannel = socket; 
  22.         mChannel.configureBlocking(false); 
  23.         mKey = mChannel.register(selector, 0); 
  24.         mKey.attach(this); 
  25.         mKey.interestOps(SelectionKey.OP_READ); 
  26.         selector.wakeup(); 
  27.         lastPant = Calendar.getInstance().getTimeInMillis(); 
  28.     } 
  29.      
  30.     public String getReceiveData() { 
  31.         return receiveData; 
  32.     } 
  33.      
  34.     public void clear() { 
  35.         receiveData = null; 
  36.     } 
  37.  
  38.     public void setSendData(String sendData) { 
  39.         mState = SENDING; 
  40.         mKey.interestOps(SelectionKey.OP_WRITE); 
  41.         this.sendData = sendData + "\n"; 
  42.     } 
  43.  
  44.     public boolean isKeekAlive() { 
  45.         return lastPant + TIME_OUT > Calendar.getInstance().getTimeInMillis(); 
  46.     } 
  47.      
  48.     public void setAlive() { 
  49.         lastPant = Calendar.getInstance().getTimeInMillis(); 
  50.     } 
  51.      
  52.     /** 
  53.      * 注銷當(dāng)前Session 
  54.      */ 
  55.     public void distroy() { 
  56.         try { 
  57.             mChannel.close(); 
  58.             mKey.cancel(); 
  59.         } catch (IOException e) {} 
  60.     } 
  61.      
  62.     @Override 
  63.     public synchronized void exec() { 
  64.         try { 
  65.             if(mState == READING) { 
  66.                 read(); 
  67.             }else if(mState == SENDING) { 
  68.                 write(); 
  69.             } 
  70.         } catch (IOException e) { 
  71.             SessionManager.remove(key); 
  72.             try { 
  73.                 mChannel.close(); 
  74.             } catch (IOException e1) { 
  75.                 Log.e(e1); 
  76.             } 
  77.             mKey.cancel(); 
  78.         } 
  79.     } 
  80.      
  81.     public void read() throws IOException{ 
  82.         mRreceiveBuffer.clear(); 
  83.         int sign = mChannel.read(mRreceiveBuffer); 
  84.         if(sign == -1) { //客戶端連接關(guān)閉 
  85.             mChannel.close(); 
  86.             mKey.cancel(); 
  87.         } 
  88.         if(sign > 0) { 
  89.             mRreceiveBuffer.flip(); 
  90.             receiveData = mDecoder.decode(mRreceiveBuffer).toString(); 
  91.             setAlive(); 
  92.             setSign(); 
  93.             SessionManager.addSession(key, this); 
  94.         } 
  95.     } 
  96.      
  97.     private void setSign() { 
  98.         //設(shè)置當(dāng)前Session的Key 
  99.         key = JsonParser.get(receiveData,"imei"); 
  100.         //檢測(cè)消息類型是否為心跳包 
  101. //        String type = jo.getString("type"); 
  102. //        if(type.equals("HEART_BEAT")) { 
  103. //            setAlive(); 
  104. //        } 
  105.     } 
  106.      
  107.      
  108.     /** 
  109.      * 寫消息 
  110.      */ 
  111.     public void write() { 
  112.         try { 
  113.             mChannel.write(mEncoder.encode(CharBuffer.wrap(sendData))); 
  114.             sendData = null; 
  115.             mState = READING; 
  116.             mKey.interestOps(SelectionKey.OP_READ); 
  117.         } catch (CharacterCodingException e) { 
  118.             e.printStackTrace(); 
  119.         } catch (IOException e) { 
  120.             try { 
  121.                 mChannel.close(); 
  122.             } catch (IOException e1) { 
  123.                 Log.e(e1); 
  124.             } 
  125.         } 
  126.     } 
  127. } 

SessionManager：

將所有Session存放到ConcurrentHashMap，這里使用手機(jī)用戶的imei做key，ConcurrentHashMap因?yàn)槭蔷€程安全的，所以能很大程度上避免自己去實(shí)現(xiàn)同步的過(guò)程，
封裝了一些操作Session的方法例如get，remove等。

  1. public class SessionManager { 
  2.  
  3.     private static ConcurrentHashMap<String, Session> sessions = new ConcurrentHashMap<String, Session>(); 
  4.      
  5.     public static void addSession(String key,Session session) { 
  6.         sessions.put(key, session); 
  7.     } 
  8.     
  9.     public static Session getSession(String key) { 
  10.         return sessions.get(key); 
  11.     } 
  12.      
  13.     public static Set<String> getSessionKeys() { 
  14.         return sessions.keySet(); 
  15.     } 
  16.      
  17.     public static int getSessionCount() { 
  18.         return sessions.size(); 
  19.     } 
  20.      
  21.     public static void remove(String[] keys) { 
  22.         for(String key:keys) { 
  23.             if(sessions.containsKey(key)) { 
  24.                 sessions.get(key).distroy(); 
  25.                 sessions.remove(key); 
  26.             } 
  27.         } 
  28.     } 
  29.     public static void remove(String key) { 
  30.         if(sessions.containsKey(key)) { 
  31.             sessions.get(key).distroy(); 
  32.             sessions.remove(key); 
  33.         } 
  34.     } 
  35. } 

SessionProcessor

里面使用了JDK自帶的線程池，用來(lái)分發(fā)處理所有Session中當(dāng)前需要處理的請(qǐng)求（線程池的初始化參數(shù)不是太熟，望有了解的童鞋能告訴我），內(nèi)部類Process則是將Session再次封裝成SocketRequest和SocketResponse（看到這里是不是有點(diǎn)熟悉的感覺(jué)，對(duì)沒(méi)錯(cuò)，JavaWeb里到處都是request和response）。

  1. public class SessionProcessor implements Runnable{ 
  2.      
  3.     private static Runnable processor = new SessionProcessor(); 
  4.     private static ThreadPoolExecutor pool = new ThreadPoolExecutor(10, 200, 500, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(10),new ThreadPoolExecutor.CallerRunsPolicy()); 
  5.     public static void start() { 
  6.         new Thread(processor).start(); 
  7.     } 
  8.      
  9.     @Override 
  10.     public void run() { 
  11.         while(true) { 
  12.             Session tmp = null; 
  13.             for(String key:SessionManager.getSessionKeys()) { 
  14.                 tmp = SessionManager.getSession(key); 
  15.                 //處理Session未處理的請(qǐng)求 
  16.                 if(tmp.getReceiveData() != null) { 
  17.                     pool.execute(new Process(tmp)); 
  18.                 } 
  19.             } 
  20.             try { 
  21.                 Thread.sleep(10); 
  22.             } catch (InterruptedException e) { 
  23.                 Log.e(e); 
  24.             } 
  25.         } 
  26.     } 
  27.      
  28.     class Process implements Runnable { 
  29.  
  30.         private SocketRequest request; 
  31.         private SocketResponse response; 
  32.          
  33.         public Process(Session session) { 
  34.             //將Session封裝成Request和Response 
  35.             request = new SocketRequest(session); 
  36.             response = new SocketResponse(session); 
  37.         } 
  38.          
  39.         @Override 
  40.         public void run() { 
  41.             new RequestTransform().transfer(request, response); 
  42.         } 
  43.     } 
  44.  
  45. } 

RequestTransform里的transfer方法利用反射對(duì)請(qǐng)求參數(shù)中的請(qǐng)求類別和請(qǐng)求動(dòng)作來(lái)調(diào)用不同類的不同方法（UserHandler和MessageHandler）

  1. public class RequestTransform { 
  2.  
  3.     public void transfer(SocketRequest request,SocketResponse response) { 
  4.         String action = request.getValue("action"); 
  5.         String handlerName = request.getValue("handler"); 
  6.         //根據(jù)Session的請(qǐng)求類型，讓不同的類方法去處理 
  7.         try { 
  8.             Class<?> c= Class.forName("com.seek.server.handler." + handlerName); 
  9.             Class<?>[] arg=new Class[]{SocketRequest.class,SocketResponse.class}; 
  10.             Method method=c.getMethod(action,arg); 
  11.             method.invoke(c.newInstance(), new Object[]{request,response}); 
  12.         } catch (Exception e) { 
  13.             e.printStackTrace(); 
  14.         } 
  15.     } 
  16. } 

SocketRequest和SocketResponse

  1. public class SocketRequest { 
  2.  
  3.     private Session mSession; 
  4.     private String  mReceive; 
  5.      
  6.     public SocketRequest(Session session) { 
  7.         mSession = session; 
  8.         mReceive = session.getReceiveData(); 
  9.         mSession.clear(); 
  10.     } 
  11.      
  12.     public String getValue(String key) { 
  13.         return JsonParser.get(mReceive, key); 
  14.     } 
  15.      
  16.     public String getQueryString() { 
  17.         return mReceive; 
  18.     } 
  19. } 
  1. public class SocketResponse {  
  2.   
  3.     private Session mSession;  
  4.     public SocketResponse(Session session) {  
  5.         mSession = session;  
  6.     }  
  7.       
  8.     public void write(String msg) {  
  9.         mSession.setSendData(msg);  
  10.     }  
  11. }  

最后則是兩個(gè)處理請(qǐng)求的Handler

  1. public class UserHandler { 
  2.  
  3.     public void login(SocketRequest request,SocketResponse response) { 
  4.         System.out.println(request.getQueryString()); 
  5.         //TODO: 處理用戶登錄 
  6.         response.write("你肯定收到消息了"); 
  7.     } 
  8. } 
  1. public class MessageHandler { 
  2.     public void send(SocketRequest request,SocketResponse response) { 
  3.         System.out.println(request.getQueryString()); 
  4.         //消息發(fā)送 
  5.         String key = request.getValue("imei"); 
  6.         Session session = SessionManager.getSession(key); 
  7.         new SocketResponse(session).write(request.getValue("sms")); 
  8.     } 
  9. } 

還有個(gè)監(jiān)測(cè)是否超時(shí)的類Looper，定期去刪除Session

  1. public class Looper extends Thread{ 
  2.     private static Looper looper = new Looper(); 
  3.     private static boolean isStart = false; 
  4.     private final int INTERVAL = 1000 * 60 * 5; 
  5.     private Looper(){} 
  6.     public static Looper getInstance() { 
  7.         return looper; 
  8.     } 
  9.      
  10.     public void loop() { 
  11.         if(!isStart) { 
  12.             isStart = true; 
  13.             this.start(); 
  14.         } 
  15.     } 
  16.      
  17.     public void run() { 
  18.         Task task = new Task(); 
  19.         while(true) { 
  20.             //Session過(guò)期檢測(cè) 
  21.             task.checkState(); 
  22.             //心跳包檢測(cè) 
  23.             //task.sendAck(); 
  24.             try { 
  25.                 Thread.sleep(INTERVAL); 
  26.             } catch (InterruptedException e) { 
  27.                 Log.e(e); 
  28.             } 
  29.         } 
  30.     } 
  31. } 
  1. public class Task { 
  2.     public void checkState() { 
  3.         Set<String> keys = SessionManager.getSessionKeys(); 
  4.         if(keys.size() == 0) { 
  5.             return; 
  6.         } 
  7.         List<String> removes = new ArrayList<String>(); 
  8.         Iterator<String> iterator = keys.iterator(); 
  9.         String key = null; 
  10.         while(iterator.hasNext()) { 
  11.             key = iterator.next(); 
  12.             if(!SessionManager.getSession(key).isKeekAlive()) { 
  13.                 removes.add(key); 
  14.             } 
  15.        } 
  16.         if(removes.size() > 0) { 
  17.             Log.i("sessions is time out,remove " + removes.size() + "session"); 
  18.         } 
  19.         SessionManager.remove(removes.toArray(new String[removes.size()])); 
  20.     } 
  21.      
  22.     public void sendAck() { 
  23.         Set<String> keys = SessionManager.getSessionKeys(); 
  24.         if(keys.size() == 0) { 
  25.             return; 
  26.         } 
  27.         Iterator<String> iterator = keys.iterator(); 
  28.         while(iterator.hasNext()) { 
  29.             iterator.next(); 
  30.             //TODO 發(fā)送心跳包 
  31.         } 
  32.     } 
  33. } 

注意，在Task和SessionProcessor類里都有對(duì)SessionManager的sessions做遍歷，文中使用的方法并不是很好，主要是效率問(wèn)題，推薦使用遍歷Entry的方式來(lái)獲取Key和Value，因?yàn)橐恢痹贘avaWeb上折騰，所以會(huì)的童鞋看到Request和Response會(huì)挺親切，這個(gè)例子沒(méi)有經(jīng)過(guò)任何安全和性能測(cè)試，如果需要放到生產(chǎn)環(huán)境上得話請(qǐng)先自行做測(cè)試- -！

客戶端請(qǐng)求時(shí)的數(shù)據(jù)內(nèi)容例如{handler:"UserHandler",action:"login",imei:"2364656512636".......},這些約定就自己來(lái)定了。

原文鏈接：http://www.cnblogs.com/freedom-elf/archive/2011/08/11/2135015.html

【編輯推薦】

  1. Java解讀NIO Socket非阻塞模式
  2. 利用NIO建立Socket服務(wù)器
  3. Java NIO的wakeup剖析
  4. Java NIO類庫(kù)關(guān)系圖解
  5. 淺析Tomcat NIO 配置
責(zé)任編輯：林師授 來(lái)源： 黎某人的博客
JavaNIO
相關(guān)推薦

2011-12-08 13:04:06

JavaNIO

2011-12-08 10:12:34

JavaNIO

2011-03-11 09:51:47

Java NIO

2011-12-07 17:05:45

JavaNIO

2009-08-18 12:51:19

服務(wù)器+客戶端

2011-12-15 11:11:51

JavaNIO

2011-06-07 08:03:30

2019-11-17 22:31:18

TCP連接流量Java

2011-12-15 10:19:55

JavaNIO

2009-12-14 10:44:51

Java 7NIO2

2009-04-26 22:27:54

觸發(fā)器密碼修改數(shù)據(jù)庫(kù)

2018-01-15 09:32:59

即時(shí)通信服務(wù)器架構(gòu)

2010-03-19 14:01:55

Java Socket

2019-01-27 20:00:14

Linux靜態(tài)文件服務(wù)器

2011-02-22 10:01:13

2019-04-26 10:57:26

無(wú)服務(wù)器應(yīng)用程序HTTP

2017-03-13 14:02:10

分布式聊天服務(wù)器

2021-08-03 15:33:05

gRPC服務(wù)器

2014-04-23 16:43:05

Linux服務(wù)器性能分析

2011-11-04 16:50:33

UbuntuARM服務(wù)器
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)