all
I'm trying to check multithread processing of the some data set that contain number from 1 to N. For example, I want to sum all this number:
1) Hold the sum (result).
public class ResultHolder {
public static AtomicLong total_time = new AtomicLong(0);
public static Long sum = 0l;
public Long getSum() {
return sum;
} // END: getSum()
@PostConstruct
public void init() {
} // END: init()
public void setSum(Long sum) {
this.sum = sum;
} // END: setSum()
public void printSum() {
System.out.println("Sum is " + sum);
}
public void clearSum() {
sum = 0l;
}
} // ENDC: ResultHolder
2) Process part of number's set:
public class SumProcessor {
private static int global_id = 0;
final public int processor_id;
private final ArrayList<Long> numbers;
private Long processor_sum = 0l;
@Autowired
private final ResultHolder sumHoldder = null;
public SumProcessor(ArrayList<Long> numbers) {
this.numbers = numbers;
processor_id = ++global_id;
} // END: constructor
public void work() throws Exception {
long t1 = new java.util.Date().getTime();
int i = 0;
try {
if (numbers == null) throw new Exception("Не удалось получить массив чисел.");
for (i = 0; i < numbers.size(); i++) {
Long o = null;
try {
o = numbers.get(i);
if (o == null) throw new Exception("no number");
} catch (Exception e) {
throw new Exception("Ошибка извлечения числа из массива: " + e);
}
processor_sum += o;
} // END: for
if (sumHoldder == null) throw new Exception("No sum holder");
synchronized (sumHoldder) {
sumHoldder.setSum(sumHoldder.getSum() + processor_sum);
}
long t2 = new java.util.Date().getTime();
this.sumHoldder.total_time.addAndGet(t2 - t1);
} catch (Exception e) {
System.out.println("Work() error (" + i + ") " + e);
}
return;
} //END: method1
@PostConstruct
public void init() {
System.out.println("Initializated B: " + this);
} //END: method2
@PreDestroy
public void destroy() {
System.out.println("Destroy B: " + this);
} //END: method3
@Override
public String toString() {
return "" +
"Processor " + processor_id + " " +
"contain " + numbers.size() + " " +
"numbers from " + numbers.get(0) +
" to " + numbers.get(numbers.size() - 1);
} //END: toString()
} //END: class SumProcessor
3) Very simple profiler (calcs processing time)
@Aspect
public class MethodLoggerBasic {
@Pointcut("execution(* *.work(..))")
void around_work() {};
@Around("around_work()")
public void logMethodName(ProceedingJoinPoint joinPoint) throws Throwable {
long starttime = new Date().getTime();
joinPoint.proceed();
long endtime = new Date().getTime();
long time = endtime - starttime;
MainApp.time += time;
} // END:
} // ENDC
4) Main program (can start processing linear or in parallel)
public class MainApp {
static AnnotationConfigApplicationContext context;
public static long time = 0l;
public final static int SIZE = 40_000_000;
public final static int DIVIDE_FACTOR = 4;
public static ArrayList<Long>[] numbers = new ArrayList[DIVIDE_FACTOR];
public static ArrayList<SumProcessor> processors = new ArrayList<>();
public static void main(String[] args) throws Exception {
context = new AnnotationConfigApplicationContext(myConfig.class);
// form 4 datasets
int part_size = SIZE / DIVIDE_FACTOR;
int i;
int j;
for (j = 0; j < DIVIDE_FACTOR; j++) {
numbers[j] = new ArrayList<>();
for (i = 0; i < (int) part_size; i++) {
numbers[j].add(((j * part_size) + i + 1l));
}
}
// create 4 processors (bean)
for (i = 0; i < DIVIDE_FACTOR; i++) {
SumProcessor bean = context.getBean(SumProcessor.class, numbers[i]);
if (bean == null) throw new Exception("Error recive bean SumProcessor.class");
processors.add(bean);
}
// creates 4 threads fro processors
thread_process thread1 = new thread_process();
thread_process thread2 = new thread_process();
thread_process thread3 = new thread_process();
thread_process thread4 = new thread_process();
ResultHolder a;
a = context.getBean(ResultHolder.class);
try {
boolean isByPool = true; // flag
time = 0;
if (isByPool) {
System.out.println("-------------------");
System.out.println("Multithread compute");
System.out.println("-------------------");
ExecutorService pool = new ThreadPoolExecutor(
4,
4,
0,
TimeUnit.MICROSECONDS,
new ArrayBlockingQueue<>(4)
);
List<Callable<Boolean>> tasks = new ArrayList();
tasks.add(thread1);
tasks.add(thread2);
tasks.add(thread3);
tasks.add(thread4);
pool.invokeAll(tasks);
pool.shutdown();
pool.awaitTermination(60, TimeUnit.SECONDS);
} else {
thread1.start();
thread2.start();
thread3.start();
thread4.start();
thread1.join();
thread2.join();
thread3.join();
thread4.join();
}
a.printSum();
a.clearSum();
System.out.println("total time is " + a.total_time);
System.out.println("basic time is " + MainApp.time);
System.out.println("-------------");
System.out.println("Single thread");
System.out.println("-------------");
ArrayList<Long> numbers_tolal = new ArrayList<>();
for (i = 0; i < SIZE; i++) {
numbers_tolal.add((i + 1l));
}
SumProcessor sumProcessor = context.getBean(SumProcessor.class, numbers_tolal);
a.total_time.set(0l);
time = 0l;
sumProcessor.work();
a.printSum();
System.out.println("total time is " + a.total_time);
System.out.println("basic time is " + MainApp.time);
} catch (Exception e) {
throw new Exception("MainApp error: " + e);
}
context.close();
} // END: main
} // END: class MainApp
5) Thread process:
public class thread_process extends Thread implements Callable, Runnable {
static int index = 0;
@Override
public void run() {
try {
SumProcessor next = MainApp.processors.get(index++);
if (next == null) {
System.out.println("Нет процессора");
System.exit(-1);
}
next.work();
System.out.println("Thread " + this + " complete!");
} catch (Exception e) {
System.out.println("Error in thread " + this + ": " + e);
}
} //END: run()
@Override
public Boolean call() throws Exception {
run();
return true;
} //END: call()
}; //END: class thread_process
The output is:
Initializated B: Processor 1 contain 10000000 numbers from 1 to 10000000
Initializated B: Processor 2 contain 10000000 numbers from 10000001 to 20000000
Initializated B: Processor 3 contain 10000000 numbers from 20000001 to 30000000
Initializated B: Processor 4 contain 10000000 numbers from 30000001 to 40000000
-------------------
Multithread compute
-------------------
Thread Thread[Thread-3,5,main] complete!
Thread Thread[Thread-4,5,main] complete!
Thread Thread[Thread-2,5,main] complete!
Thread Thread[Thread-1,5,main] complete!
Sum is 800000020000000
total time is 11254
basic time is 11254
-------------
Single thread
-------------
Initializated B: Processor 5 contain 40000000 numbers from 1 to 40000000
Sum is 800000020000000
total time is 6995
basic time is 6995
Is there a method to make it faster in parallel than linear? Or do I perhaps not need to fork this task? Or maybe my profiler is not so good...
