Многопоточность и магия

[gattsu]

Реализовал многопоточный исполнитель, присутсвует только атомарная синхронизация с помощью volatile.
Все потоки общаются с помшью каналов передавая друг другу задачи для исполнения. Задачи передаются через специальный списки, очередь с возможность изьятия за O(1) , в данный момент только в LinkdedQueue, в ArrayQueue O. Разница между ними, то что LinkedQueue активно потребляет память, в то в время как ArrayQueue массив, который расширяется по необходимость, в общем может кому будет надо, все довольно просто, реализация не сильно объемна в коде.

Цель. Понадобилось ручное управление потока исполениния сложных задач, так как ForkJoinPool не представляется средства ручной балансировки нагрузки, там используется алгоритм "кражи" нагрузки. Очень интересная реализация ForkJoinPool, когда не можешь понять как работает код. Танцы с бубном, забавно, когда ForkJoinPool реализован с помощью магии.
ПС sun.misc.Contended решает проблемы

 

[gattsu]

Вы не можете просматривать ссылку пожалуйста воспользуйтесь следующими ссылками Вход или Регистрация

 

[gattsu]

Добавил поток исполнения, теперь можно программировать поток исполнения с назначенным контекстом.

Спойлер

package concurrent.samples;

import concurrent.WorkerService;
import concurrent.subroutine.Flow;
/*
 * long count = 0;
 * long time = System.currentTimeMillis() + 1000;
 * while(true) {
 *  count++;
 *  if(time <= System.currentTimeMillis()) {
 *   System.out.println(count);
 *   count = 0;
 *   time = System.curretnTimeMillis() + 1000;
 *  }
 * }
 */
public class FlowCounter {
    public long count = 0;
   
    public long time = System.currentTimeMillis() + 1000;
   
    public static void main(String...args) {
        WorkerService.execute(
            Flow.context(new FlowCounter())
            .then(FlowCounter::inc)
            .predicate(FlowCounter::isTime)
            .then(FlowCounter::print)
             .and(FlowCounter::reset)
             .and(FlowCounter::newTime)
            .loop()
        );
    }
   
    public static void print(FlowCounter ctx) {
        System.out.println(ctx.count);
    }
   
    public static void inc(FlowCounter ctx) {
        ctx.count++;
    }
   
    public static boolean isTime(FlowCounter ctx) {
        return ctx.time <= System.currentTimeMillis();
    }
   
    public static void reset(FlowCounter ctx) {
        ctx.count = 0;
    }
   
    public static void newTime(FlowCounter ctx) {
        ctx.time = System.currentTimeMillis() + 1000;
    }
}

Зацыкленный поток исполнения, который отображает, каждую секунду, количество сделанных итерация.