java - 不管 CyclicBarrier 有一个线程停止得太早

Question

我知道下面的代码可能看起来很粗俗，但我对这些东西还是陌生的，只是为了让它工作而尝试了一切......

问题:即使我正在使用(可能以错误的方式)一个 CyclicBarrier，一个 - 并且似乎总是相同的 - 线程停止得太快并打印出他的 vector ，留下 11 个“传入连接” "消息不存在。我的循环的最后一次迭代可能存在严重错误，但我似乎无法找到确切的原因。现在程序只是循环等待处理最后一个连接。

public class VectorClockClient implements Runnable {
/*
 * Attributes
 */

/*
 * The client number is store to provide fast
 * array access when, for example, a thread's own
 * clock simply needs to be incremented.
 */
private int clientNumber;
private File configFile, inputFile;
int[] vectorClock;

/*
 * Constructor
 * @param
 * - File config
 * - int line
 * - File input
 * - int clients
 */
public VectorClockClient(File config, int line, File input, int clients) {
    /*
     * Make sure that File handles aren't null and that
     * the line number is valid.
     */
    if (config != null && line >= 0 && input != null) {
        configFile = config;
        inputFile = input;
        clientNumber = line;
        /*
         * Set the array size to the number of lines found in the
         * config file and initialize with zero values.
         */
        vectorClock = new int[clients];
        for (int i = 0; i < vectorClock.length; i++) {
            vectorClock[i] = 0;
        }
    }
}

private int parsePort() {
    int returnable = 0;
    try {
        FileInputStream fstream = new FileInputStream(configFile.getName());
        DataInputStream in = new DataInputStream(fstream);
        BufferedReader br = new BufferedReader(new InputStreamReader(in));
        String strLine = "";
        for (int i = 0; i < clientNumber + 1; i++) {
            strLine = br.readLine();
        }
        String[] tokens = strLine.split(" ");
        returnable = Integer.parseInt(tokens[1]);
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    System.out.println("[" + clientNumber + "] returned with " + returnable + ".");
    return returnable;
}

private int parsePort(int client) {
    int returnable = 0;
    try {
        FileInputStream fstream = new FileInputStream(configFile.getName());
        DataInputStream in = new DataInputStream(fstream);
        BufferedReader br = new BufferedReader(new InputStreamReader(in));
        String strLine = "";
        for (int i = 0; i < client; i++) {
            strLine = br.readLine();
        }
        String[] tokens = strLine.split(" ");
        returnable = Integer.parseInt(tokens[1]);
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    return returnable;
}

private int parseAction(String s) {
    int returnable = -1;
    try {
        FileInputStream fstream = new FileInputStream(configFile.getName());
        DataInputStream in = new DataInputStream(fstream);
        BufferedReader br = new BufferedReader(new InputStreamReader(in));
        String[] tokens = s.split(" ");
        if (!(Integer.parseInt(tokens[0]) == this.clientNumber + 1)) {
            return -1;
        }
        else {
            if (tokens[1].equals("L")) {
                vectorClock[clientNumber] += Integer.parseInt(tokens[2]);
            }
            else {
                returnable = Integer.parseInt(tokens[2]);
            }
        }
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    return returnable;
}

/*
 * Do the actual work.
 */
public void run() {
    try {
        InitClients.barrier.await();
    }
    catch (Exception e) {
        System.out.println(e);
    }
    int port = parsePort();
    String hostname = "localhost";
    String strLine;
    ServerSocketChannel ssc;
    SocketChannel sc;
    FileInputStream fstream;
    DataInputStream in;
    BufferedReader br;
    boolean eof = false;
    try {
        ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(hostname, port));
        ssc.configureBlocking(false);
        fstream = new FileInputStream("input_vector.txt");
        in = new DataInputStream(fstream);
        br = new BufferedReader(new InputStreamReader(in));

        try {
            InitClients.barrier.await();
        }
        catch (Exception e) {
            System.out.println(e);
        }

        while (true && (eof == false)) {
            sc = ssc.accept();

            if (sc == null) {
                if ((strLine = br.readLine()) != null) {
                    int result = parseAction(strLine);
                    if (result >= 0) {
                        //System.out.println("[" + (clientNumber + 1)
                        //+ "] Send a message to " + result + ".");
                        try {
                            SocketChannel client = SocketChannel.open();
                            client.configureBlocking(true);
                            client.connect(
                                    new InetSocketAddress("localhost",
                                    parsePort(result)));
                            //ByteBuffer buf = ByteBuffer.allocateDirect(32);
                            //buf.put((byte)0xFF);
                            //buf.flip();
                            //vectorClock[clientNumber] += 1;
                            //int numBytesWritten = client.write(buf);
                            String obj = Integer.toString(clientNumber+1);
                            ObjectOutputStream oos = new 
                                    ObjectOutputStream(
                                    client.socket().getOutputStream());
                            oos.writeObject(obj);
                            oos.close();
                        }
                        catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
                else {
                    eof = true;
                }
            }
            else {
                ObjectInputStream ois = new 
                        ObjectInputStream(sc.socket().getInputStream());
                String clientNumberString = (String)ois.readObject();
                System.out.println("At {Client[" + (clientNumber + 1)
                        + "]}Incoming connection from: "
                        + sc.socket().getRemoteSocketAddress()
                        + " from {Client[" + clientNumberString + "]}");
                sc.close();
            }
            try {
                InitClients.barrier.await();
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    printVector();
}

private void printVector() {
    System.out.print("{Client[" + (clientNumber + 1) + "]}{");
    for (int i = 0; i < vectorClock.length; i++) {
        System.out.print(vectorClock[i] + "\t");
    }
    System.out.println("}");
}

}
为澄清起见，这里是所用文件的格式。 Config 包含客户端使用的主机名和端口，这些客户端是线程，输入文件的行表示“该客户端向该客户端发送消息”或“该客户端将其逻辑时钟递增某个常量值”。

1 M 2(M表示发送消息)
2米3
3米4
2 L 7 (L表示递增时钟)
2米1
...
127.0.0.1 9000
127.0.0.1 9001
127.0.0.1 9002
127.0.0.1 9003
...

Answer 1

我会查看与您何时期待传入套接字连接相关的逻辑。从您的问题看来，您似乎期望一定数量的传入套接字连接(可能是在每条传出消息之后都有一个传入连接？)。由于您在传入套接字上使用非阻塞 I/O，因此您的 while 语句总是有可能在建立传入套接字之前循环。结果，线程将能够继续并从文件中读取下一行而无需接收连接。由于一旦到达文件末尾就达到了结束状态，因此您可能会错过传入的套接字连接。

我会添加一些简单的打印输出，当您从文件中读取时、发送消息时以及收到传入连接时显示。这应该很快告诉您特定线程是否缺少预期的传入连接。如果事实证明问题是由于非阻塞 I/O 引起的，那么您可能需要在期望传入套接字时禁用非阻塞 I/O，或者实现一个控件来跟踪您期望的传入套接字数量并继续，直到达到该目标。

希望这对您有所帮助。