public class SingletonConsumer {
    private final static Logger logger = Logger.getLogger(SingletonConsumer.class);
    private static SingletonConsumer instance;
    private SingletonConsumer() {
    }
    public SingletonConsumer getInstance() {
        if (instance == null) {
            logger.debug("instance is null, trying to instantiate a new one");
            instance = new SingletonConsumer();
        } else {
            logger.debug("instance is not null, return the already-instantiated one");
        }
        return instance;
    }
}

@Test
  public void singletonConsumerTest() throws InterruptedException {
       ExecutorService executors = Executors.newFixedThreadPool(10);
       Set<SingletonConsumer> set = new HashSet<>();
       for(int i = 0; i < 10; i++){
           executors.execute(
                   () -> set.add(SingletonConsumer.getInstance())
           );
       }
       executors.shutdown();
       executors.awaitTermination(1, TimeUnit.HOURS);
       Assert.assertEquals(10, set.size());
   }

public static SingletonConsumer getInstance() {
       synchronized (SingletonConsumer.class) {
           if (instance == null) {
               logger.debug("instance is null, trying to instantiate a new one");
               instance = new SingletonConsumer();
           } else {
               logger.debug("instance is not null, return the already-instantiated one");
           }
           return instance;
       }
   }

if (instance == null) {
            synchronized (SingletonConsumer.class) {
                logger.debug("instance is null, trying to instantiate a new one");
                instance = new SingletonConsumer();
            }
        } else {
            logger.debug("instance is not null, return the already-instantiated one");
        }
        return instance;

public static SingletonConsumer getInstance() {
       if (instance == null) {
           synchronized (SingletonConsumer.class) {
               if (instance == null) {
                   logger.debug("instance is null, trying to instantiate a new one");
                   instance = new SingletonConsumer();
               }
               else {
                   logger.debug("instance is not null, return the already-instantiated one ");
               }
           }
       } else {
           logger.debug("instance is not null, return the already-instantiated one");
       }
       return instance;
   }

public static void startMutiTheads(String inputPath, String outPath) {
       LocalDateTime startTime = LocalDateTime.now();
       List<Thread> threads = new ArrayList<>();
       LinkedBlockingQueue<Optional<String>> queue = new LinkedBlockingQueue<>();
       FileUtil.clearFileContents(outPath);
       Producer producer = new Producer(inputPath, queue);
       threads.add(new Thread(producer));
       Consumer consumer = new Consumer(outPath, queue);
       for(int i = 0; i < Consumer.consumerThreadCount; i++){
           Thread consumerThread = new Thread(consumer);
           threads.add(consumerThread);
           producer.addConsumer(consumer);
       }
       threads.forEach(Thread :: start);
       threads.forEach(thread -> {
           try {
               thread.join();
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       });
       consumer.getFileUtil().flushAndClose();
       // get consumer's totalCount:
       logger.debug("Consumer's totalCount: " + consumer.getTotalCount());
       LocalDateTime endTime = LocalDateTime.now();
       logger.info(String.format("It takes %s seconds to finish", 		LocalDateTime.from(startTime).until(endTime, ChronoUnit.SECONDS)));
       }

LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>();

Producer producer = new Producer(inputPath, queue);
threads.add(new Thread(producer));

Consumer consumer = new Consumer(outPath, queue);
for(int i = 0; i < Consumer.consumerThreadCount; i++){
    Thread consumerThread = new Thread(consumer);
    threads.add(consumerThread);
    producer.addConsumer(consumer);
  }

threads.forEach(Thread :: start);
threads.forEach(thread -> {
    try {
        thread.join();
        } catch (InterruptedException e) {
        e.printStackTrace();
        }
     });

public class Producer implements Runnable {
    private LinkedBlockingQueue<Optional<String>> queue;
    private String inputFile;
    public Producer(LinkedBlockingQueue<Optional<String>> queue, String inputFile) {
        this.queue = queue;
        this.inputFile = inputFile;
    }
    @Override
    public void run() {
        FileUtil.readFileLineByLine(inputFile, line -> {
            queue.add(Optional.of(line));
        });
        for(int i = 0; i < Consumer.consumerThreadCount; i++) {
            queue.add(Optional.empty());
        }
    }
}

public static void readFileLineByLine(String filePath, Consumer<String> consumer) {
       try {
           BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath), "Cp1252"));
           String line;
           while ((line = br.readLine()) != null) {
               consumer.accept(line);
           }
           br.close();
       } catch(IOException e) {
           e.printStackTrace();
       }
   }

public class Consumer implements Runnable {
    private AtomicInteger totalCount = new AtomicInteger(0);
    public static final int consumerThreadCount = 10;
    private FileUtil fileUtil;
    private final static Logger logger = Logger.getLogger(Consumer.class);
    private LinkedBlockingQueue<Optional<String>> queue;
    public Consumer(LinkedBlockingQueue<Optional<String>> queue, String outputFile) {
        this.queue = queue;
        this.fileUtil = new FileUtil(outputFile);
    }
    @Override
    public void run() {
        try {
            while (true) {
                Optional<String> line = queue.take();
                if (!line.isPresent()) break;
                totalCount.incrementAndGet();
                String processedLine = fileUtil.processLine(line.get());
                fileUtil.appendToFile(processedLine);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public FileUtil getFileUtil() {
        return fileUtil;
    }
    public int getTotalCount() {
        return totalCount.get();
    }
}

public String processLine(String line) throws IOException {
        int min = 1;
        int max = 100;
        int randomMillisecconds = min + (int)(Math.random() * ((max - min) + 1));
        try {
            Thread.sleep(randomMillisecconds);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // readLineCount is not accurate in multi-thread mode
        readLineCount++;
        logger.info(String.format("%d lines were read.", readLineCount));
        return line;
    }

if (!line.isPresent()) break;

@Test
   public void multiThreadTest() throws IOException {
       starter.startMultiThreadTask(inputFile, outputFile);
       Assert.assertTrue(isFileDataSame(inputFile, outputFile));
   }

public String processLine(String line) throws IOException {
       synchronized (this) {
           int min = 1;
           int max = 100;
           int randomMillisecconds = min + (int) (Math.random() * ((max - min) + 1));
           try {
               Thread.sleep(randomMillisecconds);
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
           // readLineCount is not accurate in multi-thread mode
           readLineCount++;
           logger.info(String.format("%d lines were read.", readLineCount));
           return line;
       }
   }

private AtomicInteger readLineCount = new AtomicInteger(0);

public String processLine(String line) throws IOException {
       int min = 1;
       int max = 100;
       int randomMillisecconds = min + (int) (Math.random() * ((max - min) + 1));
       try {
           Thread.sleep(randomMillisecconds);
       } catch (InterruptedException e) {
           e.printStackTrace();
       }
       // readLineCount is not accurate in multi-thread mode
       readLineCount.incrementAndGet();
       logger.info(String.format("%d lines were read.", readLineCount.get()));
       return line;
   }