Reading output from multiple threads and writing to a file

Question

There are 10 threads and I have 15 tasks. Now I have submitted all these tasks to these threads. I need to write all these threads output to a file which I was not successful.

I am getting output by running all the threads.

ThreadPool.java (Creates a Thread pool and adds all the tasks to the Blocking queue and submits)

public class ThreadPool {

    ExecutorService execService = null;
    BlockingQueue<Callable<String[]>> tasks = null;
    BlockingQueue<String> queue = null;

    // Get a list of Employee ID's
    public ThreadPool(List<String> empIDList) {
        try {
            tasks = new ArrayBlockingQueue<Callable<String[]>>(empIDList.size());

            for(String empNum : empIDList) {
                tasks.add(new ThreadTask(empNum));
            }

            performExecution(tasks);

        } catch(Exception e) {
            e.printStackTrace();
        }
    }

    private void performExecution(BlockingQueue<Callable<String[]>> tasks) {
        try {
            execService = Executors.newFixedThreadPool(10);

            queue = new LinkedBlockingQueue<String>();

            Runnable reader = new ReaderEmp(queue);
            Thread readerThread = new Thread(reader);
            readerThread.start();

            for(Callable<String[]> call : tasks) {
                queue.add(executeTask(call, 5));
            }

            execService.shutdown();

        } catch(Exception e) {
            e.printStackTrace();
        }
    }

    private String executeTask(Callable<String[]> task, int seconds) {
        try {
            String[] future = execService.submit(task).get(seconds, TimeUnit.SECONDS);

            System.out.println("Successfully Executed for " + future[0]);

            return future[0];
        } catch(TimeoutException toe) {
            System.out.println("Time Out for " + ((ThreadTask) task).getEmpID());
        } catch (ExecutionException ee) {
            System.out.println("Execution Exception for " + ((ThreadTask) task).getEmpID());
        } catch (InterruptedException ie) {
            System.out.println("Interrupted Exception for " + ((ThreadTask) task).getEmpID());
        }
        return ((ThreadTask) task).getEmpID() + " - Failed";
    }
}

ReaderEmp.Java (Gets each Emp_ID from the queue and writes them to the files.)

public class ReaderEmp implements Runnable {

    private final String EMP_SUCCESS = "Success.txt";
    private final String EMP_FAILURE = "Failure.txt";

    private BlockingQueue<String> queue;

    private FileWriter writerSuccess;
    private FileWriter writerFailure;

    private BufferedWriter bufWriterSuccess;
    private BufferedWriter bufWriterFailure;

    public ReaderEmp(BlockingQueue<String> queue) {
        try {
            writerSuccess = new FileWriter(EMP_SUCCESS);
            writerFailure = new FileWriter(EMP_FAILURE);
        } catch (IOException e) {
            e.printStackTrace();
        }

        bufWriterSuccess = new BufferedWriter(writerSuccess);
        bufWriterFailure = new BufferedWriter(writerFailure);

        this.queue = queue;
    }

    @Override
    public void run() {
    // This is the place which I am not able to reach while running. 
    // But when I keep a debugging point and running in debug mode then I am reaching.
        try {
            while(!queue.isEmpty()) {
                synchronized (this) {
                    String empID = queue.take();
                    if(empID != null) {
                        if(empID.contains("Failed")) {
                            bufWriterFailure.append(empID.split("-")[0].trim() + ";");
                        } else {
                            bufWriterSuccess.append(empID + ";");
                        }
                    }
                }
            }
        } catch(Exception e) {
            e.printStackTrace();
        } finally {
            try {
                synchronized (bufWriterFailure) {
                    bufWriterFailure.flush();
                    bufWriterFailure.close();
                }

                synchronized (bufWriterSuccess) {
                    bufWriterSuccess.flush();
                    bufWriterSuccess.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

Answer 1

Quite a few things are wrong with your code.

• Several threads are trying to write to the same file simultaneously. Either synchronize on a shared FileWriter or simply have each task write to its own file, and merge them afterwards. Given that the computation itself is trivial, the latter is probably preferable. and if the size of the input is small, it's probably not even worth doing this over multiple threads.
• You mix the use of an ExecutorService, with managing threads yourself. Define all the things you want to do on different threads as Runnable or Callable implementations, and submit those to an ExecutorService.
• You do not properly handle InterruptedException.
• tasks is a LinkedBlockingQueue, but could simply be an ArrayList.

Answer 2

I think you've over complicated matters a little.

Some things to be aware of:

1. new FileWriter(fileName) does not append, new FileWriter(fileName, true) does append.
2. If you want multiple threads to write to the same file, you need to synchronize the write.

See the following code which should help you solve your problem. If you really want each thread to write to the same file, then you have options such as use a static synchronized method on a helper class or global ReentrantLock or File locking.

public static void main(String[] args) throws InterruptedException, ExecutionException {

    int amountOfThreads = 10;
    ExecutorService threadPool = Executors.newFixedThreadPool(amountOfThreads);
    ExecutorCompletionService<Long> tasks = new ExecutorCompletionService<Long>(threadPool);

    //Start all of my tasks which will return long values.
    for(int i=0; i < amountOfThreads; i++) {
        tasks.submit(new Callable<Long>() {

            @Override
            public Long call() throws Exception {
                long startTime = System.currentTimeMillis();

                for(int i=0; i < 9999999; i++) {
                    //Do some stuff.
                }

                long endTime = System.currentTimeMillis();
                return endTime-startTime;
            }
        });
    }

    FileWriter out = null;

    //Output the long values of all my tasks in order of completion.
    try {
        out = new FileWriter("Success.txt");
        for(int i=0; i < amountOfThreads; i++) {
            Future<Long> task = tasks.take();
            out.write("Task " + task.toString() + " completed in " + task.get().longValue() + "ms" + System.getProperty("line.separator"));
        }
    } catch (IOException e) {
        e.printStackTrace();
    } finally {
        if(out!=null) {
            try {
                out.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    threadPool.shutdown();

}