Synchronization in Multithreading in java

Question

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.google.common.base.Charsets;
import com.google.common.io.Files;


public class Main implements Runnable
{
    static String[] entries;
    static int count;
    static HashMap<String,String> finalMap;

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

        count = 0;

        File f = new File("rd.txt");
        String contents = Files.toString(f, Charsets.UTF_8);
        entries  = contents.split("\n");

        finalMap = new HashMap<String,String>(entries.length);

        ArrayList<Thread> threads = new ArrayList();

        for(int i=0; i<10; i++)
        {
            Runnable temp = new Main();
            Thread t = new Thread(temp);
            t.setName("thread"+i);

            t.start();

            threads.add(t);
        }

        for(int i=0; i<10; i++)
        {
            threads.get(i).join();
        }

        System.out.println(finalMap);
        System.out.println(finalMap.entrySet().size());

        Set<String> set1 = finalMap.keySet();
        System.out.println(set1);
        System.out.println(set1.size());
    }

    @Override
    public void run() {
        while(true)
        {

            String temp ;
            System.out.println("Thread active "+Thread.activeCount());
            System.out.println("Count by thread: "+count + Thread.currentThread().getName());
            if(count <entries.length)
            {
                synchronized(this) 
                {
                    temp = entries[count];
                    count++;
                }

                System.out.println(temp);
                String info[] = temp.split("[|]");
                synchronized (this) {
                    finalMap.put(info[0], info[1]+"written by "+Thread.currentThread().getName());  
                }

            }
            else
            {
                break;
            }

        }

    }

}

There are entries in the rd.txt which follow the following formatting
k1|v1,v2
k2|v3,v4
k3|v5,v6
.
.
.

I have used Google guava library for reading the text file ( just FYI ). So the aim of the program is to read a huge data set using multithreading. but when i run the code i run into arrayoutofboundexception though i have used synchronization on count variable so it must never cross the limit right? so any help regarding this will be helpful.

Also the number of entries read fluctuate on every run..

Answer 1

HashMap is not thread-safe, and you are not protecting it adequately. also, your test of count and entries is outside of a synchronized block, which is not thread-safe (and the synchronized block around count/entries is also incorrect). this is going to be a different for each thread (since you have different Main instances).

Answer 2

Adding to the previous anwsers, you should learn to use the java java.util.concurrent.Executors rather than manage your threads by hand. You should equally read your file line by line within the thread in order to save the JVM memory.

My proposition of code is:

public class Main {
    private static final int NB_THREAD = 10;

    private static Map<String, String> finalMap =
        Collections.synchronizedMap(new HashMap<String, String>()); // Synchronized map
    private static AtomicInteger counter = new AtomicInteger(); // Synchronized counter
    private static BufferedReader buffer; // Line Reader

    public static void main(String[] args) throws IOException, InterruptedException, ExecutionException {
        FileInputStream fstream = new FileInputStream("rd.txt");
      DataInputStream in = new DataInputStream(fstream);
        buffer = new BufferedReader(new InputStreamReader(in));
        // Thread pool executor
        ExecutorService ececutors = Executors.newFixedThreadPool(NB_THREAD);
        List<Future<?>> futures = Lists.newArrayList();
        for (int i = 0; i < NB_THREAD; i++) {
            // Submit line reader to the executor
            futures.add(ececutors.submit(new DataLineReader()));
        }
        for (Future<?> future : futures) {
            // wait the end of the run
            future.get();
        }
        // close the resource
        buffer.close();
        System.out.println();
        System.out.println("Number of lines " + counter.get());
        System.out.println("Map size " + finalMap.entrySet().size());
        System.out.println(finalMap);
        Set<String> set1 = finalMap.keySet();
        System.out.println(set1);
        System.out.println(set1.size());
    }

    static class DataLineReader implements Runnable {
        @Override
        public void run() {
            try {
                String temp = buffer.readLine();
            while (temp != null) {
                    counter.incrementAndGet();    

                    System.out.println("Thread active " + Thread.activeCount());
                    System.out.println("Count by thread: " + Thread.currentThread().getName());
                    System.out.println(temp);

                    String info[] = temp.split("[|]");
                    finalMap.put(info[0], info[1] + "written by " + Thread.currentThread().getName());

                    temp = buffer.readLine();
                }
            } catch (IOException e) {
                Throwables.propagate(e);
            }
        }
    }
}