Máy chủ không đồng bộ sử dụng Java NIO

Question

Tôi đang sử dụng hướng dẫn this để xây dựng máy chủ nio java mà không có phần có thể ghi.

Tất cả hoạt động tốt, ngoại trừ một điều thú vị:

• Khi khách hàng đang gửi các gói tin quá nhanh, máy chủ không nhận tất cả tin nhắn, máy chủ luôn luôn nhận được là người đầu tiên và các gói thứ hai nhưng không nhiều hơn thế.
• Nếu khách hàng gửi gói chậm, máy chủ sẽ nhận tất cả các gói.

Bất kỳ ý tưởng nào?

Tôi đang thêm mã lớp máy chủ, nếu bạn cần một lớp khác được đề cập trong mã bên dưới, tôi ở đây :).

lớp NIOServer:

package server; 
import java.io.IOException; 
import java.net.InetAddress; 
import java.net.InetSocketAddress; 
import java.net.Socket; 
import java.nio.ByteBuffer; 
import java.nio.channels.SelectionKey; 
import java.nio.channels.Selector; 
import java.nio.channels.ServerSocketChannel; 
import java.nio.channels.SocketChannel; 
import java.nio.channels.spi.SelectorProvider; 
import java.util.*; 

import javax.xml.parsers.ParserConfigurationException; 

import org.xml.sax.SAXException; 

public class NioServer implements Runnable { 



// The host:port combination to listen on 
    private InetAddress hostAddress; 
    private int port; 

    // The channel on which we'll accept connections 
    private ServerSocketChannel serverChannel; 

    // The selector we'll be monitoring 
    private Selector selector; 

    //the cach will hundle the messages that came 
    private Cache cache; 

    // The buffer into which we'll read data when it's available 
    private ByteBuffer readBuffer = ByteBuffer.allocate(8192); 

    public NioServer(InetAddress hostAddress, int port , Cache cache) throws IOException { 
    this.cache = cache; 
    this.hostAddress = hostAddress; 
    this.port = port; 
    this.selector = this.initSelector(); 
    } 


    private Selector initSelector() throws IOException { 
     // Create a new selector 
     Selector socketSelector = SelectorProvider.provider().openSelector(); 

     // Create a new non-blocking server socket channel 
     this.serverChannel = ServerSocketChannel.open(); 
     serverChannel.configureBlocking(false); 

     // Bind the server socket to the specified address and port 
     InetSocketAddress isa = new InetSocketAddress(this.hostAddress, this.port); 
     serverChannel.socket().bind(isa); 

     // Register the server socket channel, indicating an interest in 
     // accepting new connections 
     serverChannel.register(socketSelector, SelectionKey.OP_ACCEPT); 

     return socketSelector; 
     } 

    private void accept(SelectionKey key) throws IOException { 
     // For an accept to be pending the channel must be a server socket channel. 
     ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel(); 

     // Accept the connection and make it non-blocking 
     SocketChannel socketChannel = serverSocketChannel.accept(); 
     Socket socket = socketChannel.socket(); 
     socketChannel.configureBlocking(false); 

     // Register the new SocketChannel with our Selector, indicating 
     // we'd like to be notified when there's data waiting to be read 
     socketChannel.register(this.selector, SelectionKey.OP_READ); 
     } 

    private void read(SelectionKey key) throws IOException { 
     SocketChannel socketChannel = (SocketChannel) key.channel(); 

     // Clear out our read buffer so it's ready for new data 
     this.readBuffer.clear(); 

     // Attempt to read off the channel 
     int numRead; 
     try { 
      numRead = socketChannel.read(this.readBuffer); 
      String test = new String(this.readBuffer.array()); 
      System.out.print(test); 

     } catch (IOException e) { 
      // The remote forcibly closed the connection, cancel 
      // the selection key and close the channel. 
     // key.cancel(); 
     // socketChannel.close(); 
      return; 
     } 

     if (numRead == -1) { 
      // Remote entity shut the socket down cleanly. Do the 
      // same from our end and cancel the channel. 
      key.channel().close(); 
      key.cancel(); 
      return; 
     } 

     // Hand the data off to our worker thread 
     this.cache.processData(this, socketChannel, this.readBuffer.array(), numRead); 
     } 

    public void run() { 
     while (true) { 
      try { 
      // Wait for an event one of the registered channels 

      this.selector.select(); 



      // Iterate over the set of keys for which events are available 
      Iterator selectedKeys = this.selector.selectedKeys().iterator(); 
      while (selectedKeys.hasNext()) { 
       SelectionKey key = (SelectionKey) selectedKeys.next(); 
       selectedKeys.remove(); 

       if (!key.isValid()) { 
       continue; 
       } 

       // Check what event is available and deal with it 
       if (key.isAcceptable()) { 
       this.accept(key); 
       } else if (key.isReadable()) { 
       this.read(key); 
       } 
      } 
      } catch (Exception e) { 
      e.printStackTrace(); 
      } 
     } 
     } 

    public static void main(String[] args) throws ParserConfigurationException, SAXException { 
    try { 
     Cache cache = new Cache(); 
     new Thread(cache).start(); 
     new Thread(new NioServer(null, 9090,cache)).start(); 
    } catch (IOException e) { 
     e.printStackTrace(); 
    } 
    } 

Answer 1

Tôi hy vọng rằng nếu bạn đang đọc UDP. Lưu ý bạn đang xử lý các gói của mình từ từ theo phương pháp read chậm đến mức nào. Bạn đang in chúng vào system.out mà là rất chậm cộng với không chắc chắn làm thế nào nhanh chóng, bạn có thể xử lý dữ liệu đến các chủ đề khác trên phương pháp processData. This library Tôi đã viết có thể giúp bạn thực hiện giao tiếp không chặn liên thông nếu đó là nguồn của sự chậm trễ của bạn. Bạn cũng nên kiểm tra kích thước của bộ đệm đọc cơ bản của bạn. Lớn hơn nó là phòng bạn phải nhanh chóng và bắt kịp trước khi các gói tin sẽ bắt đầu được giảm xuống. Đối với TCP, bạn có thể nhận được một IOException trên kênh nếu bộ đệm ổ cắm bên dưới đầy. Đối với UDP, các gói tin sẽ tự động bị loại bỏ.

Để được tiếp cận với các kích thước ổ cắm đọc đệm cơ bản bạn có thể làm:

final Socket socket = channel.socket(); 
System.out.println(socket.getReceiveBufferSize()); 
socket.setReceiveBufferSize(newSize); 

Lưu ý: AFAIK, Linux có thể yêu cầu một số cấu hình hệ điều hành trong trật tự để bạn có thể thay đổi kích thước bộ đệm phía dưới. Nếu setReceiveBufferSize không có hiệu lực (hãy đọc lại để xem nó có bị thay đổi không), google về nó. :)