java - 在 JAVA 中使用多线程(生产者消费者模型)读取和写入文件

Question

我被困在这里，有人可以解释为什么消费者线程在下面的代码中运行在生产者线程之前。当生产者没有放置任何内容时，消费者线程如何运行。是不是程序错了？

实现:-为从给定文件夹中拾取的每个文件运行生成消费者线程。

例如，如果指定的文件夹有 3 个，则每个文件必须启动 2 个线程(生产者/消费者)，这使得线程计数为 6。

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.List;
import java.util.Vector;

class sharedInt {
    private int syncUponInt;
    private boolean available = false;
    private File processingFile;
    private static File[] listOfFile;

    sharedInt(File[] totalList) {
        listOfFile = totalList;
    }

    public int getTotalCount() {
        return listOfFile.length;
    }

    public static File[] getListOfFile() {
        return listOfFile;
    }

    public static void setListOfFile(File[] listOfFile) {
        sharedInt.listOfFile = listOfFile;
    }

    public File getProcessingFile() {
        return processingFile;
    }

    public void setProcessingFile(File processingFile) {
        this.processingFile = processingFile;
    }

    public synchronized int getContents() {
        while (available == false) {
            try {
                wait();
            } catch (InterruptedException e) {
            }
        }
        available = false;
        notify();
        return syncUponInt;
    }

    public synchronized void setContents(int value) {
        while (available == true) {
            try {
                wait();
            } catch (InterruptedException e) {
            }
        }
        syncUponInt = value;
        available = true;
        notify();
    }
}

class Producer1 extends Thread {
    private sharedInt cubbyhole;
    private int number;

    public Producer1(sharedInt c, int number) {
        cubbyhole = c;
        this.number = number;
    }

    public void run() {
        // for (int i = 0; i < cubbyhole.getTotalCount(); i++) {
        cubbyhole.setContents(this.number);
        Vector vectorList = new Vector();
        System.out.println("Producer <current thread>" + this.currentThread() + "put: " + this.number
                + "processing file is" + cubbyhole.getProcessingFile());
        RandomAccessFile raf = null;

        try {
            raf = new RandomAccessFile(cubbyhole.getProcessingFile(), "r");
            StringBuffer sb = new StringBuffer();
            String line = null;
            while ((line = raf.readLine()) != null) {
                sb.append(line);
            }
            vectorList.add(sb.toString());
            System.out.println(sb.toString());
        } catch (FileNotFoundException e) {
        } catch (IOException e) {
        }

        // }
    }
}

class Consumer1 extends Thread {
    private sharedInt cubbyhole;

    public Consumer1(sharedInt c) {
        cubbyhole = c;
    }

    public void run() {
        int value = 0;
        // for (int i = 0; i < cubbyhole.getTotalCount(); i++) {

        System.out.println("Consumer <current thread>" + this.currentThread() + "got: " + cubbyhole.getContents()
                + "processing file is" + cubbyhole.getProcessingFile());
    }
}

public class FileManagementApp {
    public static void main(String[] args) throws InterruptedException {

        System.out.println("1. Please enter the path of the <Directory/Folder>...");
        // Scanner scn = new Scanner(System.in);
        // String folderPath = scn.nextLine();
        File folder = new File("C:\\file\\output");
        File[] fileList = folder.listFiles();
        int countOfFiles = fileList.length;

        sharedInt c = new sharedInt(fileList);
        Producer1 p1 = null;
        List<Producer1> pList = new ArrayList<Producer1>();
        Consumer1 c1 = null;
        List<Consumer1> cList = new ArrayList<Consumer1>();
        for (int i = 0; i < countOfFiles; i++) {
            c = new sharedInt(fileList);
            c.setProcessingFile(fileList[i]);

            p1 = new Producer1(c, i);
            p1.setName("Producer--" + i);
            pList.add(p1);
            c1 = new Consumer1(c);
            c1.setName("Consumer--" + i);
            cList.add(c1);
            pList.get(i).start();
            cList.get(i).start();
        }

    }
}

输出:-

1. Please enter the path of the <Directory/Folder>...
Consumer <current thread>Thread[Consumer--0,5,main]got: 0processing file isC:\file\output\0.A.txt
Producer <current thread>Thread[Producer--0,5,main]put: 0processing file isC:\file\output\0.A.txt
Producer <current thread>Thread[Producer--1,5,main]put: 1processing file isC:\file\output\1.A.txt
Producer <current thread>Thread[Producer--2,5,main]put: 2processing file isC:\file\output\2.A.txt
Consumer <current thread>Thread[Consumer--1,5,main]got: 1processing file isC:\file\output\1.A.txt
fg
abc
Consumer <current thread>Thread[Consumer--2,5,main]got: 2processing file isC:\file\output\2.A.txt
de

编辑:-

将代码修改成这样，可以使用生产者消费者模型实现并发/多线程同时读写文件。

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Vector;

class SharedInteger {
    private boolean available = false;
    public File processingFile;
    public long totalNoOfSplits;
    public Vector<Byte> vectorBytes;
    private File[] listOfFiles;

    SharedInteger(File[] totalList) {
        listOfFiles = totalList;
    }

    public synchronized Vector<Byte> get() {
        while (available == false) {
            try {
                wait();
            } catch (InterruptedException e) {
            }
        }
        available = false;
        notify();
        return vectorBytes;
    }

    public synchronized void put(Vector<Byte> value) {
        while (available == true) {
            try {
                wait();
            } catch (InterruptedException e) {
            }
        }
        vectorBytes = value;
        available = true;
        notify();
    }
}

class Producer extends Thread {
    private SharedInteger sharedInteger;

    public Producer(SharedInteger c) {
        sharedInteger = c;
    }

    public void run() {
        FileInputStream fis = null;
        Vector<Byte> vectorBytes = new Vector<Byte>();
        try {
            fis = new FileInputStream(sharedInteger.processingFile);
            while (fis.available() != 0) {
                vectorBytes.add((byte) fis.read());
            }
            sharedInteger.put(vectorBytes);
        } catch (Exception e) {

        }
    }
}

class Consumer extends Thread {
    private SharedInteger sharedInteger;
    private FileOutputStream fos;

    public Consumer(SharedInteger c) {
        sharedInteger = c;
    }

    public void run() {
        File newFile = sharedInteger.processingFile;
        try {
            fos = new FileOutputStream(newFile.getParentFile()+"1\\"+newFile.getName());
        } catch (FileNotFoundException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }

        Vector<Byte> v = sharedInteger.get();
        try {
            if (null != v) {
                writeToAFile(v);
            }
        } catch (IOException e) { // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public void writeToAFile(Vector<Byte> contents) throws IOException {
        for (int i = 0; i < contents.size(); i++) {
            System.out.println(Thread.currentThread());
            fos.write(contents.get(i));
            fos.flush();
        }

    }
}

public class ProducerConsumerTest {
    public static void main(String[] args) throws FileNotFoundException {
        File folder = new File("C:\\file\\output");
        File[] fileList = folder.listFiles();
        int countOfFiles = fileList.length;
        SharedInteger c = new SharedInteger(fileList);
        List<Producer> pList = new ArrayList<Producer>();
        List<Consumer> cList = new ArrayList<Consumer>();
        Producer p1 = null;
        Consumer c1 = null;

        for (int i = 0; i < countOfFiles; i++) {
            c = new SharedInteger(fileList);
            c.processingFile = fileList[i];

            p1 = new Producer(c);
            p1.setName("Producer--" + i);
            pList.add(p1);
            pList.get(i).start();

            c1 = new Consumer(c);
            c1.setName("Consumer--" + i);
            cList.add(c1);
            cList.get(i).start();
        }

    }
}

Answer 1

好吧，有几件事很可疑。但是，请查看您的 run() 方法:

// Producer1
public void run() {
    cubbyhole.setContents(this.number);
    Vector vectorList = new Vector();
    System.out.println("Producer <current thread>" + this.currentThread()
            + "put: " + this.number
            + "processing file is" + cubbyhole.getProcessingFile());
    RandomAccessFile raf = null;

    try {
        // ...
    }

// Consumer1
public void run() {
    int value = 0;

    System.out.println("Consumer <current thread>" + this.currentThread()
            + "got: " + cubbyhole.getContents()
            + "processing file is" + cubbyhole.getProcessingFile());
}

一旦您的生产者调用了 setContents(int)(因此也调用了 notify())，您的消费者就可以继续了。仅仅因为您首先看到消费者的控制台输出并不意味着什么。打印是在没有同步的情况下并发完成的，因此您不能依赖执行顺序。

编辑:根据您的要求使用Vector、wait()、notifiy() 和两个每个文件的线程，但请记住，有更好的方法来实现这个(见评论):

public class FileMerger {

    private volatile int currentWriterId = 0;

    public static void main(String[] args) throws Exception {
        // 1st argument: target directory.
        File directory = new File(args[0]);
        // 2nd argument: merge files suffix.
        FilenameFilter filter = (dir, name) -> name.endsWith("." + args[1]);

        new FileMerger().merge(directory, filter);
    }

    public void merge(File directory, FilenameFilter filter) throws IOException, InterruptedException {
        File[] files = directory.listFiles(filter);
        int numberOfFiles = files.length;
        Path mergeFilePath = Paths.get(directory.getPath() + FileSystems.getDefault().getSeparator() + "merge.txt");
        Vector<String> fileContents = new Vector<>(Collections.nCopies(numberOfFiles, null));

        Files.createFile(mergeFilePath);

        for (int i = 0; i < numberOfFiles; i++) {
            final int writerId = i;
            File file = files[i];
            CountDownLatch readWriteLatch = new CountDownLatch(1);

            // Reader.
            new Thread(() -> {
                try {
                    List<String> lines = Files.readAllLines(Paths.get(file.getPath()));
                    String content = String.join("\n", lines);

                    fileContents.set(writerId, content);
                    readWriteLatch.countDown();
                } catch (IOException e) { /* NOP */ }
            }).start();

            // Writer.
            new Thread(() -> {
                try {
                    // Wait for corresponding reader to set content.
                    readWriteLatch.await();

                    // Wait for writer ID.
                    synchronized (this) {
                        while (currentWriterId != writerId) {
                            wait();
                        }
                        Files.write(mergeFilePath, (fileContents.get(writerId) + "\n").getBytes(), StandardOpenOption.APPEND);
                        currentWriterId++;
                        notifyAll();
                    }
                } catch (InterruptedException | IOException e) { /* NOP */ }
            }).start();
        }
    }

}