像场景这样的生产者消费者模型不起作用
producer consumer model like scenario is not working
场景是这样的。我必须在定时间隔内轮询 ftp 服务器并获取 csv 文件。然后必须解析这些 CSV 文件并将其作为输入发送到某些业务逻辑。我是这样做的。
FTPClientPolling(生产者)
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(生产者 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(消费者调度程序)
class 为 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
的对象,它的队列大小为零。这两个不同的对象就是问题所在。当 class 是 singleton
时,它如何创建两个对象。我在单例实现中遗漏了什么吗?
首先,不要使用Timer
和TimerTask
。使用 ExecutorService
进行多线程处理。
并在你的单例中使用 Eager Initialization 类。或双重检查 null 锁定以使您的 Singleton 真正成为单身人士。
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
}
}
}
FileProcessor.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;
}
}
阅读此 post 了解更多信息。
场景是这样的。我必须在定时间隔内轮询 ftp 服务器并获取 csv 文件。然后必须解析这些 CSV 文件并将其作为输入发送到某些业务逻辑。我是这样做的。
FTPClientPolling(生产者)
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(生产者 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(消费者调度程序) class 为 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
的对象,它的队列大小为零。这两个不同的对象就是问题所在。当 class 是 singleton
时,它如何创建两个对象。我在单例实现中遗漏了什么吗?
首先,不要使用Timer
和TimerTask
。使用 ExecutorService
进行多线程处理。
并在你的单例中使用 Eager Initialization 类。或双重检查 null 锁定以使您的 Singleton 真正成为单身人士。
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
}
}
}
FileProcessor.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;
}
}
阅读此 post 了解更多信息。