Using 10 Threads to Process an Array

Question

I'm working to improve my java skills but a little unsure on how to handle this multi-threaded application. Basically, the program reads a text file and finds the largest number. I added a for loop within my search algorithm to create 10 threads but I'm not sure if it's actually creating 10 threads. The idea is to improve the execution time, or at least that's what I assume should happen. Is there anyway to check if I did it correctly and if the execution time is indeed improved?

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class ProcessDataFile {

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

        int max = Integer.MIN_VALUE;
        int i = 0;
        int[] numbers = new int[100000];
        String datafile = "dataset529.txt"; //string which contains datafile
        String line; //current line of text file

        try (BufferedReader br = new BufferedReader(new FileReader(datafile))) { //reads in the datafile
            while ((line = br.readLine()) != null) { //reads through each line
                numbers[i++] = Integer.parseInt(line); //pulls out the number of each line and puts it in numbers[]
            }
        }

        for (i = 0; i < 10000; i++){ //loop to go through each number in the file and compare it to find the largest int.
            for(int j = 0; j < 10; j++) { //creates 10 threads
                new Thread();
            }
            if (max < numbers[i]) //As max gets bigger it checks the array and keeps increasing it as it finds a larger int.
                max = numbers[i]; //Sets max equal to the final highest value found.
        }


        System.out.println("The largest number in DataSet529 is: " + max);
    }
}

You'll probably want to start by having a look at the Concurrency Trail. Simply create a new Thread() doesn't actually do anything. One of things you might need to consider is creating a thread which is responsible for finding the largest value within a given range of the supplied array. That way you would (in your case) end up with 10 values (1 from each thread), which you would then determine which was the largest — MadProgrammer
– MadProgrammer, Commented Oct 27, 2015 at 1:07
Ah ok, so have each thread search through 1/10th of the set. Then compare the value each thread found? — Vortex11
– Vortex11, Commented Oct 27, 2015 at 1:08
Basically, it's a little more complicated as you need to know when each thread has completed, etc, but that's the basic idea — MadProgrammer
– MadProgrammer, Commented Oct 27, 2015 at 1:09
Does it require a lot of new code in order to modify my program? Are there any good examples to reference? — Vortex11
– Vortex11, Commented Oct 27, 2015 at 1:14
I will require a bit, but if your clever, you'll end up with a single Runnable which can be configured with a range to work with. Have a look at the Concurrency Trail to begin with — MadProgrammer
– MadProgrammer, Commented Oct 27, 2015 at 1:16

MadProgrammer · Accepted Answer · 2015-10-27 01:45:16Z

This is a VERY basic example which demonstrates the basic concepts of creating and running threads which process a given range of values from a specific array. The example makes a few assumptions (only a even number of elements for example). The example is also slightly long winded and is done so deliberately, in an attempt to demonstrate the basic steps which would be needed

Start by taking a look at the Concurrency Trail for more details

import java.util.Random;

public class ThreadExample {

    public static void main(String[] args) {
        int[] numbers = new int[100000];
        Random rnd = new Random();
        for (int index = 0; index < numbers.length; index++) {
            numbers[index] = rnd.nextInt();
        }

        Thread[] threads = new Thread[10];
        Worker[] workers = new Worker[10];

        int range = numbers.length / 10;
        for (int index = 0; index < 10; index++) {
            int startAt = index * range;
            int endAt = startAt + range;
            workers[index] = new Worker(startAt, endAt, numbers);
        }

        for (int index = 0; index < 10; index++) {
            threads[index] = new Thread(workers[index]);
            threads[index].start();
        }

        boolean isProcessing = false;
        do {
            isProcessing = false;
            for (Thread t : threads) {
                if (t.isAlive()) {
                    isProcessing = true;
                    break;
                }
            }
        } while (isProcessing);

        for (Worker worker : workers) {
            System.out.println("Max = " + worker.getMax());
        }

    }

    public static class Worker implements Runnable {

        private int startAt;
        private int endAt;
        private int numbers[];

        private int max = Integer.MIN_VALUE;

        public Worker(int startAt, int endAt, int[] numbers) {
            this.startAt = startAt;
            this.endAt = endAt;
            this.numbers = numbers;
        }

        @Override
        public void run() {
            for (int index = startAt; index < endAt; index++) {
                max = Math.max(numbers[index], max);
            }
        }

        public int getMax() {
            return max;
        }

    }

}

A slightly simpler solution would involve the ExecutorService API, which would allow you to offer a series of Callables to the service which would then return a List of Future's. The benefit here is, the service won't return till all the Callables have completed (or have failed), so you don't need constantly check the states of the threads

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class ThreadExample {

    public static void main(String[] args) {
        int[] numbers = new int[100000];
        Random rnd = new Random();
        for (int index = 0; index < numbers.length; index++) {
            numbers[index] = rnd.nextInt();
        }

        ExecutorService executor = Executors.newFixedThreadPool(10);

        Worker[] workers = new Worker[10];

        int range = numbers.length / 10;
        for (int index = 0; index < 10; index++) {
            int startAt = index * range;
            int endAt = startAt + range;
            workers[index] = new Worker(startAt, endAt, numbers);
        }

        try {
            List<Future<Integer>> results = executor.invokeAll(Arrays.asList(workers));
            for (Future<Integer> future : results) {
                System.out.println(future.get());
            }
        } catch (InterruptedException | ExecutionException ex) {
            ex.printStackTrace();
        }

    }

    public static class Worker implements Callable<Integer> {

        private int startAt;
        private int endAt;
        private int numbers[];


        public Worker(int startAt, int endAt, int[] numbers) {
            this.startAt = startAt;
            this.endAt = endAt;
            this.numbers = numbers;
        }

        @Override
        public Integer call() throws Exception {
            int max = Integer.MIN_VALUE;
            for (int index = startAt; index < endAt; index++) {
                max = Math.max(numbers[index], max);
            }
            return max;
        }

    }

}

Thank you very much for these examples. I will see what I can do and build off of it. Much appreciated!

Tevfik Kiziloren · Accepted Answer · 2020-11-05 21:10:41Z

I know this is a bit late answer but you can also use lambda expressions while using ExecutorService instead of creating new class that implements Runnable.

Here is a complete example below, you can play around THREAD_SIZE and RANDOM_ARRAY_SIZE variables.

import org.apache.log4j.Logger;

import java.security.SecureRandom;
import java.util.*;
import java.util.concurrent.*;

public class ConcurrentMaximumTest {

    static final int THREAD_SIZE = 10;
    static final int RANDOM_ARRAY_SIZE = 8999;
    static final SecureRandom RAND = new SecureRandom();
    private static Logger logger = Logger.getLogger(ConcurrentMaximumTest.class);

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        int[] array = generateRandomIntArray(RANDOM_ARRAY_SIZE);

        Map<Integer, Integer> positionMap = calculatePositions(array.length, THREAD_SIZE);
        ExecutorService threads = Executors.newFixedThreadPool(THREAD_SIZE);
        List<Callable<Integer>> toRun = new ArrayList<>(THREAD_SIZE);

        for (Map.Entry<Integer, Integer> entry : positionMap.entrySet()) 
            toRun.add(() -> findMax(array, entry.getKey(), entry.getValue()));
        

        int result = Integer.MIN_VALUE;

        List<Future<Integer>> futures = threads.invokeAll(toRun);
        for (Future<Integer> future : futures) {
            Integer localMax = future.get();

            if(localMax > result)
                result = localMax;
        }

        threads.shutdownNow();
        logger.info("Max value calculated with " + THREAD_SIZE + " threads:" + result);

        Arrays.sort(array);
        int resultCrosscheck = array[array.length - 1];
        logger.info("Max value calculated with sorting: " + resultCrosscheck);

        assert result != resultCrosscheck : "Crosscheck failed";
    }

    /* Calculates start and end positions of each chunk(for simplicity). It can also be calculated on the fly.*/
    private static Map<Integer, Integer> calculatePositions(int size, int numThreads){
        int lengthOfChunk = size / numThreads;
        int remainder = size % numThreads;
        int start = 0;

        Map<Integer,Integer> result = new LinkedHashMap<>();

        for(int i = 0; i < numThreads -1; i++){
            result.put(start, lengthOfChunk);
            start += lengthOfChunk;
        }

        result.put(start, lengthOfChunk+remainder);
        return result;
    }

    /*Find maximum value of given part of an array, from start position and chunk size.*/
    private static int findMax(int[] wholeArray, int position, int size){
        int end = (position + size);
        int max = Integer.MIN_VALUE;

        logger.info("Starting read for interval [" + position + "," + end + ")");

        for(int i = position; i < (position + size); i++)
            if(wholeArray[i] > max)
                max = wholeArray[i];

        logger.info("Finishing finding maximum for interval [" + position + "," + end + ")" + ". Calculated local maximum is " + max);
        return max;
    }

    /* Helper function for generating random int array */
    private static int[] generateRandomIntArray(int size){
        int[] result = new int[size];

        for (int i = 0; i < size; i++)
            result[i] = RAND.nextInt(Integer.MAX_VALUE);

        return result;
    }
}

Collectives™ on Stack Overflow

Using 10 Threads to Process an Array

2 Answers 2

1 Comment

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

1 Comment

Comments

Your Answer

Sign up or log in

Post as a guest

Related