Runnable 中的同步方法答案

【问题标题】：synchronized methods in a RunnableRunnable 中的同步方法
【发布时间】：2012-07-16 22:06:20
【问题描述】：

我正在编写一个 Runnable 类，它将消息打包在一起一段时间或直到达到给定大小，然后再通过网络发送它们。它旨在允许其他线程在通过某些 setter 方法运行时更改某些内部参数（例如数据包大小）。使用内部对象锁使 run() 逻辑的一部分与 setter 互斥是否正确？

类似的东西：

public class Packer implements Runnable {
    private BlockingQueue<byte[]> msgQueue;
    private Object lock = new Object();
    private Packet packet;
    private boolean running = false;

    public synchronized void append(byte[] payload) throws InterruptedException {
        msgQueue.put(payload);
    }

    public synchronized void setPacketCapacity(int size) {
        synchronized (lock) {
            // check to see if we need to flush the current packet first, etc.
            packet.setCapacity(size);
        }
    }
    public void run() {
        running = true;
        while (running) {
            try {
                byte[] msg = msgQueue.take();
                synchronized (lock) {
                    packet.add(msg);
                    // check if we need to flush the packet, etc.
                }
            } catch (InterruptedException ex) {
                logger.warn("interrupted");
                running = false;
            } catch (Exception e) {
                logger.error(e);
            }
        }
        logger.warn("stop");
    }
}

与此相关，另一个线程告诉这个可运行对象停止（和刷新）的正确方法是什么？

由于run() 方法可能正在内部队列msgQueue 上等待，因此仅设置running=false 可能还不够，我可能不得不中断线程。或者，我可以向内部队列发送一个特殊的“流结束”消息，但如果队列已满，我可能需要等待一段时间才能被接受。

【问题讨论】：

我不建议重载interrupt()。那是为了终止一个线程。
当您说“flush”时，您的意思是“清除”还是写入套接字？
@Gray 我尽我所能始终勤奋并接受真正能回答我问题的答案。
@Gray：当我说刷新时，我的意思是发送数据包中可能包含的任何内容，然后将其清除。
@Pierre 的部分问题是问题/答案是为了后代——不仅仅是为了你。通过留下没有答案的问题，您正在做这样的伤害。要么自己回答，要么赏金，要么接受答案（即使不完美），编辑它们以提供更多的特异性或细节，或者删除它们。

标签： java concurrency synchronized runnable

【解决方案1】：

由于您使用一个锁定对象锁定了setter 和run() 内部的逻辑，因此它是正确的。我建议您从 setter 方法签名中删除 synchronized，因为您已经使用锁定对象锁定了其中的所有代码

您可以删除boolean running 并像这样写您的run()：

public void run() {
  while (true) {
    try {
        byte[] msg = msgQueue.take();
        synchronized (lock) {
            packet.add(msg);
            // check if we need to flush the packet, etc.
        }
    } catch (InterruptedException ex) {
        logger.warn("interrupted");
        Thread.currentThread.interrupt();
        return;
    } catch (Exception e) {
        logger.error(e);
    }
  }
  logger.warn("stop");
}

调用thread.interrupt() 将强制run() 方法中的代码转到设置中断标志的InterruptedException 捕获块并从run() 返回

【讨论】：