线程生产者-消费者 java

Question

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

生产者消费者-class：首先，当生产者完成查找文件时，我试图通过将 while 循环从 true 设置为 false 来阻止消费者，但这是行不通的。它不起作用，因为线程仍然运行显然只是没有做任何有用的事情。现在我使用 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) 其次我只锁定了方法的一部分而不是对象？！据我了解，同一对象中其他方法中的其他代码仍将同时工作。还是像同步锁一样，我在锁状态下同步整个对象？据我了解，消费者线程中的 while 循环已锁定，但生产者线程仍会调用 closePool();

额外说明：也许我什至没有以正确的方式设计我的 Producer/Consumer 模式。

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

是的，但前提是线程去检查它而不是等待其他事情，这在你的情况下正在发生。在 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 是易失性的，因此用锁保护它不会像 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() andunlock()` 保护块内的代码才会被并发访问锁定。对象本身可以自由地在这些块之外执行任何并发的（对锁定的块）任务。

最后遵循正确的命名约定并为您的变量命名。

你的问题的主要答案 为什么你的线程仍然运行是因为它们正在等待 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);
        }
    }
}

ProducerConsumerMain

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

线程生产者-消费者 java

Thread Producer-Consumer java

java

multithreading

producer-consumer

locks