像场景这样的生产者消费者模型不起作用

Question

这是场景。我必须按时间间隔轮询 ftp 服务器并获取 csv 文件。然后，必须解析这些 CSV 文件并将其作为输入发送到某些业务逻辑。我是这样做到的。

FTPClientPolling （Producer）

public class FTPClientPolling {
    private static FTPClientPolling instance = null;
    private FTPClientPolling() {
    }
    public synchronized static FTPClientPolling getInstance() {
        if (instance == null) {
            logger.info("Object created for Client Polling");
            instance = new FTPClientPolling();
            initializeFTPClient();
        }
        return instance;
    }
    public static void initializeFTPClient() {
        // initialize the values from properties file
    }
    public void startPolling() {
        FTPClient ftpClient = null;
        try {
            //connecting to ftp server
            //iterating the files in it
            FTPFile[] filesList = ftpClient.listFiles();
            for (FTPFile tmpFile : filesList) {
                //..
                File tempFile = File.createTempFile(tmpFile.getName(), null);
                FileOutputStream fileOut = new FileOutputStream(tempFile);
                ftpClient.retrieveFile(tmpFile.getName(), fileOut);
                //adding the file to the Queue of the file processor
                FileProcessor.getInstance().getFilesToBeProcessedQueue().add(tempFile);
            }
            if (ftpClient.isConnected())
                ftpClient.disconnect();
        } catch (Exception e) {
            //logging
        } finally {
            //closing ftpclient
        }
    }
}

FTPClientPollingTasker （producer Tasker）

public class FTPClientTasker extends TimerTask {
    private static Long timeInterval = 10000l;
    @Override
    public void run() {
        FTPClientPolling.getInstance().startPolling();
    }
    public static void start() {
        TimerTask timerTask = new FTPClientTasker();
        Timer timer = new Timer();
        timer.scheduleAtFixedRate(timerTask, timeInterval, timeInterval);
    }
    public static void main(String[] args) {
        start();
    }
}

文件处理器（消费者）

public final class FileProcessor {
    private static FileProcessor instance = null;
    private Queue<File> filesToBeProcessedQueue = new ArrayBlockingQueue<File>(10);
    private FileProcessor() {
    }
    public synchronized static FileProcessor getInstance() {
        if (instance == null) {
            instance = new FileProcessor();
        }
        return instance;
    }
    public void run() {
        while (!filesToBeProcessedQueue.isEmpty()) {
            processSyncFiles(filesToBeProcessedQueue.poll());
        }
    }
    private void processSyncFiles(File inputFile) {
        try {
            HashMap<String, Boolean> outputConsolidation = new HashMap<String, Boolean>();
            FileReader fileReader = new FileReader(inputFile);
            List<InputBean> csvContentsList = CSVParser.readContentsFromCSV(fileReader, new InputBean());
            for (InputBean inputBean : csvContentsList) {
                boolean output = false;
                // some business logic
                outputConsolidation.put(inputBean.toString(), output);
            }
        } catch (Exception e) {
            //logging
        }
    }
    public synchronized Queue<File> getFilesToBeProcessedQueue() {
        return filesToBeProcessedQueue;
    }
}

文件处理器任务程序（使用者调度程序）此类为 FileProcessor 创建一个 Tasker，并按计划的时间间隔运行它。

public final class FileProcessorTasker extends TimerTask {
    private static Long timeInterval = 5000l;
    @Override
    public void run() {
        FileProcessor.getInstance().run();
    }
    public static void start() {
        TimerTask timerTask = new FileProcessorTasker();
        Timer timer = new Timer();
        timer.schedule(timerTask, timeInterval, timeInterval);
    }
    public static void main(String[] args) {
        FileProcessorTasker.start();
    }
}

这两个程序单独运行良好。但是当通过filesToBeProcessedQueue连接在一起时，它似乎不起作用。问题是FTPClientPolling创建一个FileProcessor的对象并将该文件添加到队列中。但 FileProcessorTasker创建另一个队列大小为零的 FileProcessor 对象。这两个不同的对象是问题所在。当类是singleton时，如何创建两个对象.我在单例实现中是否遗漏了某些内容？

Answer 1

首先，不要使用Timer和TimerTask。使用ExecutorService进行多线程处理。

并在单例类中使用预先初始化。或者仔细检查空的锁定，以使你成为真正的单例。

FTPClientPolling.java

public class FTPClientPolling {
    private static FTPClientPolling instance = new FTPClientPolling();
    private FTPClientPolling() {
        logger.info("Object created for Client Polling");
        initializeFTPClient();
    }
    public static FTPClientPolling getInstance() {
        return instance;
    }
    public static void initializeFTPClient() {
        // initialize the values from properties file
    }
    public void startPolling() {
        FTPClient ftpClient = null;
        try {
            //connecting to ftp server
            //iterating the files in it
            FTPFile[] filesList = ftpClient.listFiles();
            for (FTPFile tmpFile : filesList) {
                //..
                File tempFile = File.createTempFile(tmpFile.getName(), null);
                FileOutputStream fileOut = new FileOutputStream(tempFile);
                ftpClient.retrieveFile(tmpFile.getName(), fileOut);
                //adding the file to the Queue of the file processor
                FileProcessor.getInstance().getFilesToBeProcessedQueue().add(tempFile);
            }
            if (ftpClient.isConnected())
                ftpClient.disconnect();
        } catch (Exception e) {
            //logging
        } finally {
            //closing ftpclient
        }
    }
}

文件处理器.java

public final class FileProcessor {
    private static FileProcessor instance = new FileProcessor();
    private Queue<File> filesToBeProcessedQueue = new ArrayBlockingQueue<File>(10);
    private FileProcessor() {
    }
    public static FileProcessor getInstance() {
        return instance;
    }
    public void run() {
        while (!filesToBeProcessedQueue.isEmpty()) {
            processSyncFiles(filesToBeProcessedQueue.poll());
        }
    }
    private void processSyncFiles(File inputFile) {
        try {
            HashMap<String, Boolean> outputConsolidation = new HashMap<String, Boolean>();
            FileReader fileReader = new FileReader(inputFile);
            List<InputBean> csvContentsList = CSVParser.readContentsFromCSV(fileReader, new InputBean());
            for (InputBean inputBean : csvContentsList) {
                boolean output = false;
                // some business logic
                outputConsolidation.put(inputBean.toString(), output);
            }
        } catch (Exception e) {
            //logging
        }
    }
    public synchronized Queue<File> getFilesToBeProcessedQueue() {
        return filesToBeProcessedQueue;
    }
}

阅读这篇文章以获取更多信息。