Java - 同步但允许不同线程访问一种方法

Question

在下面的例子中:

public class MsLunch {
    private long c1 = 0;
    private long c2 = 0;
    private Object lock1 = new Object();
    private Object lock2 = new Object();

    public void inc1() {
        synchronized(lock1) {
            c1++;
        }
    }      

    public void inc2() {
        synchronized(lock2) {
            c2++;      
        }
    }
}

inc1和inc2可以同时访问，但都不能被多线程同时访问。

怎么可能只允许访问 inc1 或 inc2 而另一个就像常规同步一样，但是允许被访问的那个被尽可能多的线程完成。

Answer 1

我认为一个有用的类比是通过十字路口的交通，在那里您可以有多辆汽车共享一条道路，只要它们平行行驶即可。挑战在于为交叉路口找到一种协调策略。

如果流量是间歇性的，@Greg 提出的解决方案可以工作，我们可以等待一个流停止，然后再允许相交的流继续进行。但我怀疑这不太现实。如果一条道路上交通稳定，其余车辆将永远等待，也就是线程饥饿。

另一种策略是让汽车按照先到先得的原则通过，就像在 parking 标志处一样。我们可以使用专用的 semaphore 来实现它对于每个用户获得许可的每个“道路”或路段，首先确保其他路段都没有使用许可:

public class StopSign {
    private final Semaphore[] locks;
    private volatile int current = 0;

    public StopSign(int segments) {
        // create and populate lock array, leaving
        // all segments drained besides the first
        locks = new Semaphore[segments];
        Arrays.setAll(locks, i -> new Semaphore(i == 0 ? Integer.MAX_VALUE : 0, true));
    }

    public void enter(int segment) {
        // synchronization is necessary to guard `current`,
        // with the added benefit of holding up new threads
        // in the active segment while we're gathering permits
        synchronized (locks) {
            if (segment == current) {
                // if our segment is active, acquire a permit
                locks[segment].acquireUninterruptibly();
            } else {
                // otherwise, gather all permits from the active segment
                // as they become available and then reclaim our own permits
                locks[current].acquireUninterruptibly(Integer.MAX_VALUE);
                current = segment;
                locks[segment].release(Integer.MAX_VALUE - 1);
            }
        }
    }

    public void exit(int segment) {
        if (segment != current) {
            // we don't own the lock!
            throw new IllegalMonitorStateException();
        }
        locks[segment].release();
    }
}

要使用该类，我们只需调用enter(i) 和exit(i)，其中i 标识道路/路段/我们要使用的方法。这是一个使用 3 个片段的演示:

public static void main(String args[]) {
    int segments = 3;
    StopSign lock = new StopSign(segments);
    IntStream.range(0, segments).parallel().forEach(i -> {
        for (int j = 0; j < 10; j++) {
            lock.enter(i);
            System.out.print(i);
            lock.exit(i);
            sleepUninterruptibly(20, TimeUnit.MILLISECONDS);
        }
    });
}

在我的机器上运行测试会产生这种交替模式:

120201210012012210102120021021

如果交通相对较少，此策略可能有意义，但在交通繁忙时，协调每个交叉路口的开销会显着限制吞吐量。对于繁忙的十字路口，您通常需要一个红绿灯，或者可以以合理的频率转移控制权的第三方。这是一个这样的概念的实现，使用管理一组 read/write locks 的后台线程, 确保一次只有一个段有可用的写锁:

public class TrafficLight {
    private final ReadWriteLock[] locks;
    private final Thread changer;

    public TrafficLight(int segments, long changeFrequency, TimeUnit unit) {
        // create and populate lock array
        locks = new ReadWriteLock[segments];
        Arrays.setAll(locks, i -> new ReentrantReadWriteLock(true));

        CountDownLatch initialized = new CountDownLatch(1);
        changer = new Thread(() -> {
            // lock every segment besides the first
            for (int i = 1; i < locks.length; i++) {
                locks[i].writeLock().lock();
            }
            initialized.countDown();

            int current = 0;
            try {
                while (true) {
                    unit.sleep(changeFrequency);
                    // lock the current segment and cycle to the next
                    locks[current].writeLock().lock();
                    current = (current + 1) % locks.length;
                    locks[current].writeLock().unlock();
                }
            } catch (InterruptedException e) {}
        });
        changer.setDaemon(true);
        changer.start();

        // wait for the locks to be initialized
        awaitUninterruptibly(initialized);
    }

    public void enter(int segment) {
        locks[segment].readLock().lock();
    }

    public void exit(int segment) {
        locks[segment].readLock().unlock();
    }

    public void shutdown() {
        changer.interrupt();
    }
}

现在让我们调整测试代码:

TrafficLight lock = new TrafficLight(segments, 100, TimeUnit.MILLISECONDS);

结果是一个有序的模式:

000111112222200000111112222200

注意事项:

• awaitUninterruptibly() 和 sleepUninterruptibly() 是 Guava helper methods以避免处理 InterruptedException。随意复制 implementation如果您不想导入库。
• TrafficLight 可以通过将状态管理委托(delegate)给访问线程来实现，而不是依赖后台线程。这个实现更简单(我认为)，但它确实有一些额外的开销，它需要一个 shutdown() 来进行垃圾收集。
• 为方便起见，测试代码使用并行流，但根据您的环境，它可能不会很好地交错。您始终可以改用适当的线程。