java - java线程生产者-消费者

Question

我必须完成一项练习，我必须使用至少一个生产者线程和 x 个消费者线程的生产者/消费者模式在我的文件夹路径中查找“.java”文件。

生产者消费者级:首先，当生产者完成查找文件时，我尝试通过设置从 true 到 false 的 while 循环来阻止消费者，但这不起作用。它不起作用，因为线程显然仍在运行，只是没有做任何有用的事情。现在我也使用 closePool() 函数。

所以如果我不忍受我的名为locka的锁，这个函数确实可以工作。这基本上是我不明白的事情。

所以如果我有

    loka.lock();
    ende = false;
    loka.unlock();

和

    while(ende){
        loka.lock();
        System.out.println(xy.getQueue());
        loka.unlock();
    }

closePool() 函数永远不会被调用。这是我不明白的事情。如果我把锁放在 while 循环中，它就会工作并且线程会停止。

问题:

1) ende 参数无论如何都会被设置为 false，因此锁最终会被释放。

2)其次我只锁定了方法的一部分而不是对象？!据我了解，同一对象中其他方法中的其他代码仍然会同时工作。或者是像synchronized和i这样的锁在处于锁定状态时同步整个对象？根据我的理解，消费者线程中的 while 循环被锁定，但生产者线程仍然会调用 closePool();

额外说明:也许我什至没有以正确的方式设计我的生产者/消费者模式。

import java.io.IOException;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class FindJavaVisitorp extends SimpleFileVisitor<Path> {

    private BlockingQueue<String> xxx = new ArrayBlockingQueue<String>(10);

    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {

        if (file.toString().endsWith(".java")) {

            try {
                xxx.put(file.toString());

            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        return FileVisitResult.CONTINUE;
    }

    public String getQueue() throws InterruptedException {
        return xxx.take();

    }
}

public class ProducerConsumer {

    private volatile boolean ende = true;
    private Path path;
    private FindJavaVisitorp xy;
    private Lock loka = new ReentrantLock();
    private ExecutorService pepe;

    public ProducerConsumer(Path path, FindJavaVisitorp xy, ExecutorService xyz) {
        this.path = path;
        this.xy = xy;
        pepe = xyz;
    }

    public void produce() throws IOException, InterruptedException {
        Files.walkFileTree(path, xy);
        loka.lock();
        ende = false;
        loka.unlock();
        closePool();

    }

    public void consume() throws InterruptedException {
        while (ende) {
            loka.lock();
            System.out.println(xy.getQueue());
            loka.unlock();

        }
    }

    public void closePool() {
        pepe.shutdown();
        try {
            if (!pepe.awaitTermination(60, TimeUnit.SECONDS)) {
                pepe.shutdownNow();
                if (!pepe.awaitTermination(60, TimeUnit.SECONDS)) {
                    System.err.println("Pool couldn't be terminated!");
                }

            }

        } catch (InterruptedException e) {
            pepe.shutdownNow();

        }
    }
}

public class Test {
    public static void main(String[] args) {
        Path startingDir = Paths.get("/usr/local/");
        FindJavaVisitorp x = new FindJavaVisitorp();
        ExecutorService exec = Executors.newCachedThreadPool();
        final ProducerConsumer pp = new ProducerConsumer(startingDir, x, exec);

        exec.submit(new Runnable() {

            public void run() {
                try {
                    pp.produce();

                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        });
        // x.printQueue();

        for (int j = 0; j < 5; j++) {

            exec.submit(new Runnable() {

                public void run() {
                    try {
                        pp.consume();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }

                }
            });
        }
        exec.shutdown();
    }
}

Answer 1

1. 是的，但前提是线程去检查它而不是等待其他事情，在你的情况下正在发生。在 while 循环中，线程正在等待 BlockingQueue，如果队列为空，则将没有机会检查标志变量。此外，您不需要锁，因为您已经在使用 BlockingQueue。
   在您的示例中，两个关键部分之间没有关系。

。以下代码将尝试保护 loka=false 免受任何并发访问。

loka.lock();
ende = false;//critical section
loka.unlock();

以下代码将不受并发访问的影响，并且与上述临界区互斥。

while(ende){
    loka.lock();
    System.out.println(xy.getQueue());//critical section
    loka.unlock();
}

由于这两个关键部分之间没有任何共同点，因此互斥不起作用。由于 ende 是 volatile 的，用锁保护它不会像 primitive types already have atomic access 那样做任何事情。 .

Reads and writes are atomic for reference variables and for most primitive variables (all types except long and double).
Reads and writes are atomic for all variables declared volatile (including long and double variables).

只有通过 lock() 和 unlock()` 保护的 block 内的代码才会被锁定，无法进行并发访问。对象本身可以自由地在这些 block 之外执行任何并发(锁定 block )任务。

最后遵循正确的命名约定并为变量指定有意义的名称。

主要答案您的线程仍在运行的原因是因为它们正在等待 blockingQueue.takeItem() 并且它们无法从中释放，除非队列再次被填满，但是由于 Producer 已完成，因此不可能发生这种情况。

如何避免这种行为

BlockingQueue 上没有允许 immediate release of waiting threads 的方法我们可以做的一件事是让生产者放置一个 LAST_ITEM，并让消费者检查他们获得的项目是否是 LAST_ITEM，这样他们就可以释放自己。

以下是工作代码。我对变量和方法名称进行了一些修改，使它们更有意义。

JavaFileVisitor

package filevisitor;

import java.nio.file.FileVisitResult;
import java.nio.file.Path;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class JavaFileVisitor extends SimpleFileVisitor<Path> {
    private BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(10);

    public static String NO_MORE_ITEMS = "### NO MORE ITEMS ###";

    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
        if (file.toString().endsWith(".java")) {
            try {
                blockingQueue.put(file.toString());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return FileVisitResult.CONTINUE;
    }

    public String getQueueItem() throws InterruptedException {
        String item = blockingQueue.take();
        if(NO_MORE_ITEMS.equals(item)) {
            setNoMoreItems();
        }
        return item;
    }

    public void setNoMoreItems() {
        try {
            blockingQueue.put(NO_MORE_ITEMS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

生产者消费者

package filevisitor;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;

public class ProducerConsumer {

    private Path path;
    private JavaFileVisitor fileVisitor;

    public ProducerConsumer(Path path, JavaFileVisitor visitor) {
        this.path = path;
        this.fileVisitor = visitor;
    }

    public void produce() throws IOException, InterruptedException {
        Files.walkFileTree(path, fileVisitor);
        fileVisitor.setNoMoreItems();
    }

    public void consume() throws InterruptedException {
        while (true) {
            String item = fileVisitor.getQueueItem();
            if(JavaFileVisitor.NO_MORE_ITEMS.equals(item)) {
                break;
            }
            System.out.println(item);
        }
    }
}

生产者消费者主要

package filevisitor;

import java.io.IOException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ProducerConsumerMain {
    public static void main(String[] args) {
        Path startingDir = Paths.get("src/filevisitor");
        JavaFileVisitor fileVisitor = new JavaFileVisitor();
        ExecutorService executor = Executors.newCachedThreadPool();
        final ProducerConsumer producerConsumer = new ProducerConsumer(startingDir, fileVisitor);

        executor.submit(new Runnable() {
            public void run() {
                System.out.println("Producer started");
                try {
                    producerConsumer.produce();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("Producer finished");
            }
        });

        for (int j = 0; j < 5; j++) {
            executor.submit(new Runnable() {
                public void run() {
                    String threadName = Thread.currentThread().getName();
                    System.out.println(threadName + " Consumer Started");
                    try {
                        producerConsumer.consume();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(threadName + " Consumer finished");
                }
            });
        }

        executor.shutdown();
        System.out.println("Executor shutdown, waiting for threads to finish");

        try {
            executor.awaitTermination(60, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Exiting main");
    }
}

输出

Producer started
pool-1-thread-3 Consumer Started
pool-1-thread-2 Consumer Started
Executor shutdown, waiting for threads to finish
pool-1-thread-5 Consumer Started
pool-1-thread-6 Consumer Started
pool-1-thread-4 Consumer Started
src\filevisitor\JavaFileVisitor.java
src\filevisitor\ProducerConsumerMain.java
src\filevisitor\ProducerConsumer.java
pool-1-thread-6 Consumer finished
pool-1-thread-4 Consumer finished
pool-1-thread-3 Consumer finished
pool-1-thread-5 Consumer finished
Producer finished
pool-1-thread-2 Consumer finished
Exiting main