Запрос на оптимизацию использования RxJava - PullRequest
Новая
       6

Запрос на оптимизацию использования RxJava

0 голосов
/ 08 февраля 2019

Сегодня я попытался решить небольшую задачу:

Вы большая компания с 500 офисами, вы хотите рассчитать глобальный доход (сумма доходов каждого офиса).

Каждыйофис выставляет сервис для получения дохода.Вызов требует определенной задержки (сеть, доступ к базе данных, ...).

Очевидно, вы хотите получить глобальный доход как можно быстрее.

Сначала я попробовал в python довольно неплохие результаты:

import asyncio
import time

DELAYS = (475, 500, 375, 100, 250, 125, 150, 225, 200, 425, 275, 350, 450, 325, 400, 300, 175)


class Office:

    def __init__(self, delay, name, revenue):
        self.delay = delay
        self.name = name
        self.revenue = revenue

    async def compute(self):
        await asyncio.sleep(self.delay / 1000)
        print(f'{self.name} finished in {self.delay}ms')
        return self.revenue


async def main(offices, totest):
    computed = sum(await asyncio.gather(*[o.compute() for o in offices]))
    verdict = ['nok', 'ok'][computed == totest]
    print(f'Sum of revenues = {computed} {verdict}')


if __name__ == "__main__":
    offices = [Office(DELAYS[i % len(DELAYS)], f'Office-{i}', 3 * i + 10) for i in range(500)]
    totest = sum(o.revenue for o in offices)
    start = time.perf_counter()
    asyncio.run(main(offices, totest))
    end = time.perf_counter()
    print(f'Ends in {(end-start)*1000:.3f}ms')

На моем компьютере это занимает около 500 мс, идеальный случай (потому что 500 мс - максимальная задержка)

Далее я пробовал в java с RxJava: 

import java.util.concurrent.TimeUnit;

public class Office {
    private int sleepTime;
    private String name;
    private int revenue;

    public Office(int sleepTime, String name, int revenue) {
        this.sleepTime = sleepTime;
        this.name = name;
        this.revenue = revenue;
    }

    public int getRevenue() {
        return revenue;
    }

    public int compute() {
        try {
            TimeUnit.MILLISECONDS.sleep(this.sleepTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.printf("%s finished in %dms on thread %d%n", this.name, this.sleepTime, Thread.currentThread().getId());
        return this.revenue;
    }
}

import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;

public class Tester {
    private static int[] DELAYS = {475, 500, 375, 100, 250, 125, 150, 225, 200, 425, 275, 350, 450, 325, 400, 300, 175};

    public static void main(String[] args) {
        final ArrayList<Office> offices = new ArrayList<>();

        for (int i = 0; i < 500; i++) {
            offices.add(new Office(DELAYS[i % DELAYS.length], String.format("Office-%d", i), 3 * i + 10));
        }
        int totest = offices.stream().mapToInt(Office::getRevenue).sum();

        final Instant start = Instant.now();
        final Flowable<Office> officeObservable = Flowable.fromIterable(offices);
        int computation = officeObservable.parallel(500).runOn(Schedulers.io()).map(Office::compute).reduce(Integer::sum).blockingSingle();
        boolean verdict = computation == totest;
        System.out.println("" + computation + " " + (verdict ? "ok" : "nok"));
        final Instant end = Instant.now();

        System.out.printf("Ends in %dms%n", Duration.between(start, end).toMillis());

    }
}

На моем компьютере это занимает около 1000 мс (с пулом из 500 потоков!).

Конечно, я пробовал с другим количеством потоков, но результаты худшие или схожие.

Я не хочу сравнивать Python и Java, я просто хочу:

Пояснения, если я допустил ошибки

Лучший подход?

Кроме того, асинхронность Python использует только один потокно в Java я не нашел, как не использовать многопоточность, чтобы получить похожий результат.

Может быть, кто-то может мне помочь?: -)

Ответы [ 2 ]

0 голосов
/ 09 февраля 2019

После многих попыток (спасибо M. T за помощь), наконец, у меня есть хорошая реализация Java! 

public class Office {
    private int sleepTime;
    private int revenue;

    public Office(int sleepTime, int revenue) {
        this.sleepTime = sleepTime;
        this.revenue = revenue;
    }

    public int getRevenue() {
        return revenue;
    }

    public Single<Integer> compute() {
        return Single.timer(sleepTime, TimeUnit.MILLISECONDS).map(l -> this.revenue);
    }
}


public class Tester {
    private static int[] DELAYS = {475, 500, 375, 100, 250, 125, 150, 225, 200, 425, 275, 350, 450, 325, 400, 300, 175};

    public static void main(String[] args) {
        final ArrayList<Office> offices = new ArrayList<>();

        for (int i = 0; i < 1_000_000; i++) {
            offices.add(new Office(DELAYS[i % DELAYS.length], 1));
        }
        int totest = offices.stream().mapToInt(Office::getRevenue).sum();

        final Instant start = Instant.now();
        final Flowable<Office> officeObservable = Flowable.fromIterable(offices);
        int computation = officeObservable.flatMapSingle(Office::compute).reduce(Integer::sum).blockingGet();
        boolean verdict = computation == totest;
        System.out.println("" + computation + " " + (verdict ? "ok" : "nok"));
        final Instant end = Instant.now();

        System.out.printf("Ends in %dms%n", Duration.between(start, end).toMillis());
    }
}

Этот код работает быстро!2с для 1_000_000 офисов!

0 голосов
/ 08 февраля 2019

Это довольно просто.На стороне Python вы ждете в асинхронном режиме (не блокируя) на стороне Java, вы ждете с кодом блокировки, отсюда и разница.

Правильный код в Java должен быть: 

package com.test;

import io.reactivex.Flowable;
import io.reactivex.Single;
import io.reactivex.schedulers.Schedulers;
import org.reactivestreams.Publisher;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;


public class TestReactive {

    public static class Office {
        private int sleepTime;
        private String name;
        private int revenue;

        public Office(int sleepTime, String name, int revenue) {
            this.sleepTime = sleepTime;
            this.name = name;
            this.revenue = revenue;
        }

        public int getRevenue() {
            return revenue;
        }

        public Publisher<Integer> compute() {
            return Single.just("")
                    .delay(this.sleepTime, TimeUnit.MILLISECONDS)
                    .map(x-> {
                        System.out.printf("%s finished in %dms on thread %d%n", this.name, this.sleepTime, Thread.currentThread().getId());
                        return this.revenue;
                    }).toFlowable();
        }
    }

    private static int[] DELAYS = {475, 500, 375, 100, 250, 125, 150, 225, 200, 425, 275, 350, 450, 325, 400, 300, 175};

    public static void main(String[] args) {
        final ArrayList<Office> offices = new ArrayList<>();

        for (int i = 0; i < 500; i++) {
            offices.add(new Office(DELAYS[i % DELAYS.length], String.format("Office-%d", i), 3 * i + 10));
        }
        int totest = offices.stream().mapToInt(Office::getRevenue).sum();

        final Instant start = Instant.now();

        final Flowable<Office> officeObservable = Flowable.fromIterable(offices);
        int computation = officeObservable.parallel(2).runOn(Schedulers.io()).flatMap(Office::compute).reduce(Integer::sum).blockingSingle();
        boolean verdict = computation == totest;
        System.out.println("" + computation + " " + (verdict ? "ok" : "nok"));
        final Instant end = Instant.now();

        System.out.printf("Ends in %dms%n", Duration.between(start, end).toMillis());

    }

}

Редактировать: я установил параллель 2, но кого это волнует, вы можете поставить один поток, так как здесь не проблема с ограничением процессора.

...