锁是否正确

Question

我正在读取 UDP 提要，然后对其进行解码并写入 MSMQ（消息队列）。

我创建了一个调用 UDPReader 的新线程。 UDPReader 依次创建一个线程池并调用 class ipaddrConnection。运行内部 ipaddrConnection 包含一个 while 循环，不断从多播套接字读取数据包并将其推送到 class parseUDP。从 parseUDP 它被解码并最终推送到写入 MSMQ 的 class。我相信当谈到 ipaddrConnection 中的 while 循环时我没有正确锁定线程，因为线程正试图写入 MSMQ 中的相同内存位置。我认为通过将我的锁放在 while 循环中，池中的每个线程都会在 "Critical Section" 1.receive 数据包中有自己的时间，然后 2.decode 并写入 MSMQ。我仍在学习并发并寻求帮助。我提供了一个我不知道如何正确读取的故障转储文件以及我的 UDPReader 和 ipaddrConnection classes。 parseUDP 调用 class 来解码数据包，而 class 调用 MSMQ class 来写入内存。所有这些都在我的关键部分。

class UDPReader implements Runnable
{
    private final String ip, socket, queue, threadName;
    private final JTextArea screen;

    UDPReader(String ip, String socket, String queue, String threadName, JTextArea screen) 
    {
        this.ip = ip;
        this.socket = socket;
        this.queue = queue;
        this.threadName = threadName;
        this.screen = screen;
    }

    public void run()
    {
        screen.append("Thread " + threadName + " running\n\n");
        ExecutorService executor = Executors.newFixedThreadPool(5);
        Runnable reader = new ipaddrConnection(ip, socket, queue);
        executor.execute(reader);
    }

}

public final class ipaddrConnection implements Runnable
{
    private final ReentrantLock lock = new ReentrantLock();
    byte[] bytes = new byte[(int)100000];
    InetAddress group; 
    MulticastSocket s;
    DatagramPacket packet = new DatagramPacket(bytes, bytes.length);
    private String queue;

    public ipaddrConnection(String ip, String socket, String queue) {
        try {
            this.s = new MulticastSocket(Integer.parseInt(socket));
            this.group = InetAddress.getByName(ip);
            this.queue = queue;
        } catch (IOException ex) {
            Logger.getLogger(ipaddrConnection.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

        @Override
        public void run() {
            try {
                parseUDP p = new parseUDP(queue);
                s.joinGroup(group);
                s.setSoTimeout(95000);

                try{
                    while(true){
                        lock.lock();
                        s.receive(packet);
                        p.parseUDP(packet.getData());
                    } 
                }finally {
                    lock.unlock();
                }


             } catch (SocketException ex) {
                Logger.getLogger(ipaddrConnection.class.getName()).log(Level.SEVERE, null, ex);
            } catch (IOException ex) {
                Logger.getLogger(ipaddrConnection.class.getName()).log(Level.SEVERE, null, ex);
            }
        }

}

崩溃报告 https://drive.google.com/file/d/0B4GWNCU6_CBlM2tJNGJqNzRVazg/view?usp=sharing

Answer 1

在您的代码中，您的锁没有做任何有用的事情。

每个线程都有自己的锁，所以一次可以有多个线程使用队列（因为线程 1 锁定了 Lock 1，线程 2 锁定了 Lock 2，并且没有什么可以阻止他们使用队列同时）。

如果您在代码中创建 lock 字段 static，那么所有线程都将使用相同的锁。

你可能还有问题，因为线程永远不会释放锁（除非它们遇到异常）所以只允许一个线程做工作：

try{
    while(true){
        lock.lock();
        s.receive(packet);
        p.parseUDP(packet.getData());
    } 
}finally {
    lock.unlock();
}

注意线程解锁锁的唯一方法是在出现异常时吗？

您可能想要更像这样的东西：

while(true) {
    s.receive(packet);
    try {
        lock.lock();
        s.parseUDP(packet.getData());
    } finally {
        lock.unlock();
    }
}

- 使用这种结构，线程只会在解析数据包时持有锁，而不是在接收数据包时持有锁。（不知道是不是你想要的）

Answer 2

ExecutorService executor = Executors.newFixedThreadPool(5);
Runnable reader = new ipaddrConnection(ip, socket, queue);
executor.execute(reader);

这段代码实际上是单线程的，因为尽管池有五个线程，但您只用完了一个。

Answer 3

让 UDPReader 实现 Runnable 并且它的 run() 实现至少不是惯用的。
正如 immibis 所提到的，您的锁对象不会在线程之间共享，并且它们不提供您正在寻找的保护。

只有退出 while (true) { ... } 才能解锁，也就是说永远不会。考虑到这一点，您可能需要考虑以下内容：

    public class UDPReader {
        ...

        UDPReader(String ip, String socket, String queue, String threadName, JTextArea screen, numberOfThreads) {
            ...
            this.numberOfThreads = numberOfThreads;
            this.lock = new ReentrantLock();
        }

        public void run() {
            ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
            for (int i = 0; i < numberOfThreads; i++){
                executor.execute(new ipaddrConnection(ip, socket, queue, lock));
            }
        }
     }


     public final class ipaddrConnection implements Runnable {
        private lock ;
        ...

        public ipaddrConnection(String ip, String socket, String queue, ReentrantLock lock) {
            ...
            this.lock = lock;
        }

        @Override
        public void run() {
                ...
                while (true) {
                    try {
                        lock.lock();
                        s.receive(packet);
                        p.parseUDP(packet.getData());
                    } finally {
                        lock.unlock();
                    }
                }
             ....
        }
    }
}

锁是否正确

Is the lock being done correctly

java

multithreading

locks