java - 读者/作家通过引用

Question

以下代码是Java中的Passing the Baton程序的一部分:

主要
P1(作家)
P2(作家)
P3(阅读器)
P4(阅读器)

主要();

package ReadersPreference;

import java.util.concurrent.Semaphore;

/**
  * * @author me
 */
public class Main {

public static void main(String[] args) {


    AnyData x = new AnyData(5.7);//gives writers something to write, 
                                   //readers something to read
    Semaphore e = new Semaphore(1);//control entry
    Semaphore r = new Semaphore(0);//used to delay readers
    Semaphore w = new Semaphore(0);//used to delay writers

    int nr = 0;//readers active
    int nw = 0;//writers active

    int dr = 0;//readers waiting
    int dw = 0;//writers waiting

    P1 r1 = new P1(e, r, w, x, nr, nw, dr, dw); // #reader thread 1
    P2 r2 = new P2(e, r, w, x, nr, nw, dr, dw); // #reader thread 2
    P5 r3 = new P5(e, r, w, x, nr, nw, dr, dw); // #reader thread 3
    P6 r4 = new P6(e, r, w, x, nr, nw, dr, dw); // #reader thread 4

    P3 w1 = new P3(e, r, w, x, nr, nw, dr, dw); // #writer thread 1
    P4 w2 = new P4(e, r, w, x, nr, nw, dr, dw); // #writer thread 2

System.out.println("threads commanded to start");

r1.start();     // calls run() method in Thread
r2.start();
r3.start();     
r4.start();   
w1.start();
w2.start();
}//end of main
}//end of class

读者流程

package ReadersPreference;

import java.util.concurrent.Semaphore;

public class P1 extends Thread {

private Semaphore e;
private Semaphore r;
private Semaphore w;
private AnyData pillarbox;
private int nw;
private int nr;
private int dr;
private int dw;

public P1(Semaphore e, Semaphore r, Semaphore w, AnyData pbox,
        int nw, int nr, int dr, int dw) {

    this.nw = nw;
    this.nr = nr;
    this.dr = dr;
    this.dw = dw;

    this.e = e;
    this.r = r;
    this.w = w;
    pillarbox = pbox;
}// end of constructor

public void run() {

PERFORM OPERATIONS          

}// end of run method
}// end of class

现在，根据我的输出，它似乎可以工作。但是，我的讲师指出了两个主要缺陷。一种是计数器(nr，nw，dr，dw)是通过值而不是通过引用传递的。这意味着每个线程都会检查自己的数据副本，而不是使用共享变量，这会阻止程序按其应有的方式运行。

我一直在阅读有关通过价值和参考传递的内容，起初我感到头肿胀，我想我现在已经了解了大部分内容，但是我仍然没有找到解决方案。我可以理解为什么信号量已经工作，因为它们已经是一个引用(而传递的值是一个引用)。我看不出计数器到底在哪里或如何解决它。

当他说线程时，他是指流程类(构造函数/算法)还是主类中的实例化或两者？

基于阅读作为对象的共享原语，最接近解决方案的是:

public void setCounters(int nr){nr = newNR;}   

再次，尽管我对如何实现它含糊不清

第二个主要的缺陷是我创建了多个进程类文件，并将每个文件作为线程运行，而不是编写两个(读取器/写入器)并根据需要使用了许多对象。我不明白这是什么意思，但是，考虑到我的输出打印语句对于每个过程都是唯一的，因此要在输出中唯一地标识每个输出以进行验证，为什么将这种方法用于某个目的时会认为是缺陷，对读者/作家的解决方案有影响吗？

Answer 1

首先，好的变量名总是比注释更好；遵守这个规则，您就不会出错。想象一下，我在代码中的某个地方遇到了您的e变量，现在我必须滚动到类的顶部，并阅读注释以查看其含义，然后返回到原来的位置。这使得代码几乎不可读...

您的第一个问题是您正在使用的int是原始类型，它是通过值传递的。 @Marcin的解决方案不是线程安全的；如果您执行类似int++的操作，那么当被多个线程调用时，这可能会做各种奇怪的事情(例如不递增，返回错误的值等)。 始终在多线程操作中使用线程安全对象。

正如@Marcin建议的那样，您可以将数据包装在一个类中以减少代码量:

public class SharedData<T> {

    private final T data;
    private final Semaphore entryControl = new Semaphore(1);
    private final Semaphore readerDelay = new Semaphore(0);
    private final Semaphore writerDelay = new Semaphore(0);
    private final AtomicInteger activeReaders = new AtomicInteger(0);
    private final AtomicInteger activeWriters = new AtomicInteger(0);
    private final AtomicInteger waitingReaders = new AtomicInteger(0);
    private final AtomicInteger waitingWriters = new AtomicInteger(0);

    public SharedData(final T data) {
        this.data = data;
    }
    //getters
}

我已经将此类设置为泛型类，但是您可以根据需要删除泛型并将“数据”作为“对象”-泛型更好，因为它提供了类型安全性，请阅读 here。

该类使用 AtomicInteger对象，这是一个线程安全的整数，允许进行诸如 getAndSet(int newValue)之类的原子操作-这样就消除了访问单个值时出现线程安全问题的可能性，但看起来您可能想要访问两个值，但这仍然不是线程安全的因此，您可能需要按照以下步骤向数据类添加一些方法:

public synchronized void makeReaderActive() {
    //perform checks etc
    waitingReaders.decrementAndGet();
    activeReaders.incrementAndGet();
}

否则，读者可能会递减 waitingReaders，然后在递增之前先读取 activeReaders。

我认为，这可以解决您的第一个查询；现在到您的第二个。老师说您创建了多个流程类文件，而不是创建实例。这是因为您复制粘贴的同一文件后会调用不同的内容( P1， P2等)，这不是Java的工作方式。考虑在创建 Semaphore的地方的代码:

Semaphore e = new Semaphore(1);//control entry
Semaphore r = new Semaphore(0);//used to delay readers

您有一个类文件(Semaphore.class)，并已创建了两个实例。您不必将JDK的 Semaphore类复制到另一个文件中并创建该文件。您不必这样做:

Semaphore1 e = new Semaphore1(1);//control entry
Semaphore2 r = new Semaphore2(0);//used to delay readers

所以;在您的示例中，我假设您有一些 Reader进程和一些 Writer进程，然后需要两个类:

public class MyReader implements Callable<Void> {

    private final String name;
    private final SharedData sharedData;

    public MyReader(final String name, final SharedData sharedData) {
        this.name = name;
        this.sharedData = sharedData;
    }

    @Override
    public Void call() {
        //do stuff
        return null;
    }
}

public class MyWriter implements Callable<Void> {

    private final String name;
    private final SharedData sharedData;

    public MyWriter(final String name, final SharedData sharedData) {
        this.name = name;
        this.sharedData = sharedData;
    }

    @Override
    public Void call() {
        //do stuff
        return null;
    }
}

一类代表您所有作家的所作所为，另一类代表您所有读者的所作所为。我使它们成为 Callable而不是线程；这使我想到了下一个要点。

您不应该使用 Thread对象，这些对象级别很低并且很难正确管理。您应该使用新的 ExecutorService。因此，您的 main方法现在看起来像:

public static void main(String[] args) {
    final SharedData<Double> sharedData = new SharedData<Double>(5.7);
    final List<Callable<Void>> myCallables = new LinkedList<Callable<Void>>();

    for (int i = 0; i < 4; ++i) {
        myCallables.add(new MyReader("reader" + i, sharedData));
    }
    for (int i = 0; i < 2; ++i) {
        myCallables.add(new MyWriter("writer" + i, sharedData));
    }

    final ExecutorService executorService = Executors.newFixedThreadPool(myCallables.size());
    final List<Future<Void>> futures;
    try {
        futures = executorService.invokeAll(myCallables);
    } catch (InterruptedException ex) {
        throw new RuntimeException(ex);
    }
    for (final Future<Void> future : futures) {
        try {
            future.get();
        } catch (InterruptedException ex) {
            throw new RuntimeException(ex);
        } catch (ExecutionException ex) {
            throw new RuntimeException(ex);
        }
    }
}

让我带您看一下这段代码，以便您了解此处的情况。在第一行，我们创建一个新的 SharedData类-在这种情况下，它是 SharedData<Double>；这意味着它包含的 data是 Double类型；这可能是任何东西。

接下来，我们创建一个 List的 Callable，这是我们的工作 class 去的地方。然后，我们将工作类循环放入 List中。请注意，我们为每个 MyReader和 MyWriter创建了相同类的实例-我们不需要每个类都有一个类文件。

然后，我们创建一个新的 ExecutorService，其线程池的大小与我们创建的工作类的数量相同-注意，我们可以拥有更少的线程。在这种情况下，每个线程将分配一个工作类，然后在完成工作后将为其分配一个新的工作类，直到完成所有工作。在我们的例子中，有足够的线程，因此每个简单的线程都会分配一个工作类。

现在，我们在工作类的 invokeAll中传递 List，这是我们要求 ExecutorService对所有 call进行 Callable的地方。该方法一直阻塞，直到一切都完成为止，它可能会像其他任何等待方法一样抛出 InterrupedException-在这种情况下，我们抛出异常并退出。

最后，这就是 ExecutorService的亮点，我们遍历返回的 Future类列表并调用 get-如果在与该将来相关的工作中遇到任何问题，这将抛出 ExecutionException。在这种情况下，我们抛出异常。

注意 Callable的类型为 Void(即声明为 Callable<Void>)，然后过滤器过滤到 Future的类型为 Future<Void>。如果要从每个进程返回一些数据，则可以将类型更改为 Callable<MyData>，然后 get的 Future方法将向您返回此数据。