使用非阻塞套接字的多线程

Question

我正在尝试使用 nio 在 Java 中实现 TCP 服务器。 它只是使用 Selector 的 select 方法来获取准备好的键。然后处理这些密钥（如果它们是可接受的、可读的等）。服务器工作得很好，直到我使用单线程。但是当我尝试使用更多线程来处理密钥时，服务器的响应会变慢并最终停止响应，比如在 4-5 个请求之后。 这就是我所做的一切：（伪）

Iterator<SelectionKey> keyIterator =  selector.selectedKeys().iterator();
while (keyIterator.hasNext()) {
                SelectionKey readyKey = keyIterator.next();
                if (readyKey.isAcceptable()) {
                    //A new connection attempt, registering socket channel with selector

                } else {
                    Worker.add( readyKey );
                }

Worker 是从通道执行输入/输出的线程类。 这是我的 Worker 类的代码：

private static List<SelectionKey> keyPool = Collections.synchronizedList(new LinkedList());

public static void add(SelectionKey key) {
    synchronized (keyPool) {
        keyPool.add(key);
        keyPool.notifyAll();
    }
}


public void run() {
    while ( true ) {

        SelectionKey myKey = null;
        synchronized (keyPool) {
            try {
                while (keyPool.isEmpty()) {
                    keyPool.wait();
                }
            } catch (InterruptedException ex) {                    
            }
            myKey = keyPool.remove(0);
            keyPool.notifyAll();
        }

        if (myKey != null && myKey.isValid() ) {

            if (myKey.isReadable()) {
                //Performing reading
            } else if (myKey.isWritable()) {
                //performing writing
                myKey.cancel();  
            }
        }
    }

我的基本想法是将密钥添加到 keyPool 中，各个线程都可以从中获取密钥，一次一个。 我的 BaseServer 类本身作为线程运行。它在事件循环开始之前创建 10 个工作线程。我还尝试增加 BaseServer 线程的优先级，以便它有更多机会接受可接受的密钥。尽管如此，它在大约 8 个请求后停止响应。请帮忙，我是不是错了。提前致谢。 :)

Answer 1

第三，您不会从选定键集中删除任何内容。您必须在每次循环时都这样做，例如在调用 next() 之后调用 keyIterator.remove()。

您需要阅读 NIO 教程。

Answer 2

首先，您不应该再使用 wait() 和 notify() 调用，因为 java.util.concurrent 中存在良好的标准 Java (1.5+) 包装类，例如 BlockingQueue。

其次，建议在选择线程中进行 IO，而不是在工作线程中。工作线程应该只是排队读取/写入选择器线程。

这个页面解释得很好，甚至提供了一个简单的 TCP/IP 服务器的工作代码示例：http://rox-xmlrpc.sourceforge.net/niotut/

抱歉，我还没有时间查看您的具体示例。

Answer 3

尝试使用 xsocket 库。它为我节省了很多在论坛上阅读的时间。

下载：http://xsocket.org/

教程：http://xsocket.sourceforge.net/core/tutorial/V2/TutorialCore.htm

服务器代码：

import org.xsocket.connection.*;

/**
 *
 * @author wsserver
 */
public class XServer {

    protected static IServer server;

    public static void main(String[] args) {
        try {
            server = new Server(9905, new XServerHandler());
            server.start();
        } catch (Exception ex) {
            System.out.println(ex.getMessage());
        }
    }
     protected static void shutdownServer(){
        try{
            server.close();
        }catch(Exception ex){
            System.out.println(ex.getMessage());
        }        
    }
}

服务器处理程序：

import java.io.IOException;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import java.util.*;
import org.xsocket.*;
import org.xsocket.connection.*;

public class XServerHandler implements IConnectHandler, IDisconnectHandler, IDataHandler {

    private Set<ConnectedClients> sessions = Collections.synchronizedSet(new HashSet<ConnectedClients>());

    Charset charset = Charset.forName("ISO-8859-1");
    CharsetEncoder encoder = charset.newEncoder();
    CharsetDecoder decoder = charset.newDecoder();
    ByteBuffer buffer = ByteBuffer.allocate(1024);

    @Override
    public boolean onConnect(INonBlockingConnection inbc) throws IOException, BufferUnderflowException, MaxReadSizeExceededException {
        try {
            synchronized (sessions) {
                sessions.add(new ConnectedClients(inbc, inbc.getRemoteAddress()));
            }
            System.out.println("onConnect"+" IP:"+inbc.getRemoteAddress().getHostAddress()+" Port:"+inbc.getRemotePort());
        } catch (Exception ex) {
            System.out.println("onConnect: " + ex.getMessage());
        }
        return true;
    }

    @Override
    public boolean onDisconnect(INonBlockingConnection inbc) throws IOException {
        try {
            synchronized (sessions) {
                sessions.remove(inbc);
            }
            System.out.println("onDisconnect");
        } catch (Exception ex) {
            System.out.println("onDisconnect: " + ex.getMessage());
        }
        return true;
    }

    @Override
    public boolean onData(INonBlockingConnection inbc) throws IOException, BufferUnderflowException, ClosedChannelException, MaxReadSizeExceededException {
        inbc.read(buffer);
        buffer.flip();
        String request = decoder.decode(buffer).toString();
        System.out.println("request:"+request);
        buffer.clear();
        return true;
    }
}

连接的客户：

import java.net.InetAddress;
import org.xsocket.connection.INonBlockingConnection;

/**
 *
 * @author wsserver
 */
public class ConnectedClients {

    private INonBlockingConnection inbc;
    private InetAddress address;

    //CONSTRUCTOR
    public ConnectedClients(INonBlockingConnection inbc, InetAddress address) {
        this.inbc = inbc;
        this.address = address;
    }

    //GETERS AND SETTERS
    public INonBlockingConnection getInbc() {
        return inbc;
    }

    public void setInbc(INonBlockingConnection inbc) {
        this.inbc = inbc;
    }

    public InetAddress getAddress() {
        return address;
    }

    public void setAddress(InetAddress address) {
        this.address = address;
    }
}

客户代码：

import java.net.InetAddress;
import org.xsocket.connection.INonBlockingConnection;
import org.xsocket.connection.NonBlockingConnection;

/**
 *
 * @author wsserver
 */
public class XClient {

    protected static INonBlockingConnection inbc;
    public static void main(String[] args) {
        try {
            inbc = new NonBlockingConnection(InetAddress.getByName("localhost"), 9905, new XClientHandler());

            while(true){

            }
        } catch (Exception ex) {
            System.out.println(ex.getMessage());
        }
    }
}

客户端处理程序：

import java.io.IOException;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import org.xsocket.MaxReadSizeExceededException;
import org.xsocket.connection.IConnectExceptionHandler;
import org.xsocket.connection.IConnectHandler;
import org.xsocket.connection.IDataHandler;
import org.xsocket.connection.IDisconnectHandler;
import org.xsocket.connection.INonBlockingConnection;

/**
 *
 * @author wsserver
 */
public class XClientHandler implements IConnectHandler, IDataHandler,IDisconnectHandler, IConnectExceptionHandler {

    Charset charset = Charset.forName("ISO-8859-1");
    CharsetEncoder encoder = charset.newEncoder();
    CharsetDecoder decoder = charset.newDecoder();
    ByteBuffer buffer = ByteBuffer.allocate(1024);

    @Override
    public boolean onConnect(INonBlockingConnection nbc) throws IOException {
        System.out.println("Connected to server");
        nbc.write("hello server\r\n");
        return true;
    }

    @Override
    public boolean onConnectException(INonBlockingConnection nbc, IOException ioe) throws IOException {

        System.out.println("On connect exception:"+ioe.getMessage());
        return true;
    }

    @Override
    public boolean onDisconnect(INonBlockingConnection nbc) throws IOException {

        System.out.println("Dissconected from server");
        return true;
    }

    @Override
    public boolean onData(INonBlockingConnection inbc) throws IOException, BufferUnderflowException, ClosedChannelException, MaxReadSizeExceededException {

        inbc.read(buffer);
        buffer.flip();
        String request = decoder.decode(buffer).toString();
        System.out.println(request);
        buffer.clear();
        return true;
    }
}

我花了很多时间在论坛上阅读这方面的内容，我希望我的代码可以帮助你。