-2

这是我的代码。它应该实现一种在分布式系统中同步调用的方式。这是我发送消息并等待确认的部分。网络和对等点被认为是可靠的。

我的问题虽然与本地同步有关，因为它不断引发此异常，我不明白为什么。

Exception in thread "Thread-7" java.lang.IllegalMonitorStateException
at java.lang.Object.wait(Native Method)
at java.lang.Object.wait(Object.java:503)
at distributed.PeerManager.sendAllWithAck(PeerManager.java:212)
at distributed.TokenManager.onTokenReceived(TokenManager.java:67)
at communication.TokenMessage.execute(TokenMessage.java:21)
at distributed.ListenThread.run(ListenThread.java:38)
Exception in thread "Thread-8" java.lang.IllegalMonitorStateException
    at java.lang.Object.notify(Native Method)
    at distributed.AckWaiter.run(AckWaiter.java:74)

这是 AckWaiter 类：

public class AckWaiter extends Thread {
    public int counter;
    PeerManager pm;
    Message m;
    public Object waiter;
    public int parentPort;

    public AckWaiter(PeerManager pm, Message m, int n,int parentPort) {
        counter = n;
        this.pm = pm;
        this.m = m;
        this.parentPort=parentPort;
        waiter=new Object();
    }

    public synchronized void notifyAck() {
        counter--;
            notify();

    }

    @Override
    public synchronized void run(){
        pm.sendAllExceptMe(m);

        try {
            BufferedReader inFromClient=new BufferedReader(new
                    InputStreamReader(pm.listener.socketMap.get(parentPort).getInputStream()));
            while(counter>0){
                    wait(100);
                    Message m=CustomMarshaller.getCustomMarshaller().unmarshal(inFromClient.readLine());
                    if(m==null){
                        continue;
                    }
                    m.execute(pm);
                }
                } catch (IOException | InterruptedException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        } 


        waiter.notify();
        return;
    }
}

这是应该阻塞线程的调用

public synchronized void sendAllWithAck(Message m){
    if(aw!=null){
        throw new RuntimeException();
    }
    aw=new AckWaiter(this,m,connectionList.size()-1,m.sender.getPort());
    aw.start();
    try {
        aw.waiter.wait();
    } catch (InterruptedException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

    aw=null;
}

'sendAllWithAck' 应该启动一个 AckWaiter 线程，等待当我收到 'n' 个 ACK​​ 时将通知的对象 'waiter'。

当收到 ACK 消息时，通信层会调用“notifyAck”方法。

AckWaiter 是一个独立的线程，因为“sendAllWithAck”可以由负责从套接字读取的线程调用。我有 n 个对等点，每个对等点都有一个开放的套接字（带有处理入站消息的线程）用于每个其他对等点，所以如果我等待 ACK 以响应收到的消息，我将无法从那个中读取 ACK对等点（parentPort 是该对等点的标识符，因此我可以定期检查该线程内的 ACK）。

如果这个异常是由于结构问题引起的，我愿意修改我的架构，但我不知道在不修改我的通信层的情况下以不同的方式处理它，那将是一个痛苦。

4

1 回答 1

1

方法wait()，notify()并且notifyAll()必须在synchronized块中。并且run方法只被一个线程访问（所以它不需要同步）并且异常在ListenThread中。

于 2013-05-26T11:22:09.473 回答