0

Я лишь начал изучать RxJava и нуждаюсь в помощи в нескольких вопросах, чтобы выполнить задание.

1.

У меня есть сервис, который красит фигурки и должен возвращать фигурку после задержки — то есть через секунду (для кружков) или две(для квадратиков) после того, как его об этом попросили. Блокировать поток нельзя, нужно использовать таймеры.

Пока я смог добиться задержки таким способом, но она занимает заметно больше, чем секунду. Я добавил Scheduler, поскольку без него результат не возвращается вообще:

public Observable<PaintedCircle> paint(Shape shape) {
        return Observable
                .timer(1, TimeUnit.SECONDS, Schedulers.immediate())
                .flatMap(x -> Observable.just(new PaintedCircle(shape.getSize())));
    }

Этот сервис покраски я вызываю из кода ниже (код пока черновой). Там же фигурки раскладываются по коробкам по 5 штук и выводятся в консоль:

public class Consumer {

    private static final int CIRCLE_MIN_SIZE = 30;

    public static void main(String[] args) {

        Producer producer = new Producer();
        while (!producer.doesShouldStop()) {
            Observable.from(producer.produceShapes(40))
                    .filter(shape -> shape instanceof Square || shape instanceof Circle && shape.getSize() > CIRCLE_MIN_SIZE)
                    .flatMap(shape -> shape.getPaintingService().paint(shape))
                    .buffer(5)
                    .map(shapes -> new Box(shapes))
                    .forEach(System.out::println);
        }
    }
}

Как улучшить реализацию задержки?

2.

Второй вопрос касается создания самих самих фигурок. Квадратики и кружки должны создаваться случайным образом и случайным размером. Пока что я использую для этого обычную Джаву, но, вероятно, это тоже можно делать с помощью реактивности. Пока мой код выглядит так:

public class Producer {

    private int MAX_SHAPE_SIZE = 100;

    private int counter;
    private boolean shouldStop;

    List<Shape> produceShapes(int amount) {

        List<Shape> shapes = new ArrayList<>();
        Random random = new Random();

        System.out.println("Produced following shapes:");
        for (int i = 0; i < amount; i++) {
            int coin = random.nextInt(2);
            int size = Math.abs(random.nextInt(MAX_SHAPE_SIZE));
            Shape shape = coin == 0 ? new Circle(size) : new Square(size);
            shapes.add(shape);
            System.out.print(shape);
        }
        System.out.println();
        if (++counter == 3) {
            shouldStop = true;
        }
        return shapes;
    }

    boolean doesShouldStop() {
        return shouldStop;
    }
}

Можно ли это переписать с помощью RxJava?

Кроме того, по условию задачи, Producer после трех циклов должен сообщить Consumer'у, что фигурок больше нет. Как видно на коде Consumer'а, пока это делается с помощью цикла while, но это нужно тоже изменить — вероятно, с помощью отписки.

Прошу помочь.

1 ответ 1

3

Блокировать поток нельзя, нужно использовать таймеры. Пока я смог добиться задержки таким способом, но она занимает заметно больше, чем секунду. Я добавил Scheduler, поскольку без него результат не возвращается вообще

Во первых ваш Scheduler.immediate() по сути как раз и блокирует поток ибо заставляет таймер работать не в отдельном потоке а в текущем.

Почему у вас задержка занимает значительно больше секунды из приведенного кода не ясно, возможно генерация или вывод фигур отжирают много времени?

Далее не понятно зачем вам понадобился flatMap в методе paint. В общем надо бы так:

public Observable<PaintedCircle> paint(Shape shape) {
    return Observable
            .timer(1, TimeUnit.SECONDS)
            .map(x -> new PaintedCircle(shape.getSize()));
}

Но тут есть 2 проблемки:

  1. Таймер у вас будет срабатывать через секунду после получения очередного шейпа, мэпить в какой то ваш отрисованный шейп и отсылать дальше. Но следующий шейп будет браться не после этого, а сразу после запуска таймера от предыдущего. В результате у вас Выводиться шейпы будут не каждую секунду, а все скопом через секунду после их генерации так как сгенерировались они скопом. У вас оно работало вроде как нормлаьно из за использования Scheduler.immediate() который блочил поток на таймере и не давал заэмитится слудующему шейпу пока не отработает таймер текущего.
  2. Слудующая проблемка в том что без использования Scheduler.immediate() в main потоке у вас запустятся таймеры и назначатся действия что делать когда истекут таймеры, но ожидать завершения работы тамеров никто не собирается, метод main завершится сразу после запуска всех таймеров, а с ним и приложенька, не выдав никаких результатов.

По поводу первой проблемки я не знаю, может так вам и надо, но если нет, то скорее всего вам надо при передаче шейпа в сервис на отрисовку делать не flatMap а concatMap. Если я правильно понял доки, то эта штука должна в отличии от flatMap ждать завершения предыдущего действия прежде чем приступить к новому.

Со второй же проблемой легко справиться добавив перед .forEach(System.out::println) переход в блокирущее состояние: toBlocking()

В итоге второй ваш фрагмент станет таким:

public class Consumer {

    private static final int CIRCLE_MIN_SIZE = 30;

    public static void main(String[] args) {

        Producer producer = new Producer();
        while (!producer.doesShouldStop()) {
            Observable.from(producer.produceShapes(40))
                    .filter(shape -> shape instanceof Square || shape instanceof Circle && shape.getSize() > CIRCLE_MIN_SIZE)
                    .concatMap(shape -> shape.getPaintingService().paint(shape))
                    .buffer(5)
                    .map(shapes -> new Box(shapes))
                    .toBlocking()
                    .forEach(System.out::println);
        }
    }
}

Кстати а зачем вам тут buffer(5)? Из за него вывод в консоль будет откладываться до получения каждого 5го шейпа.

По поводу переписывания Producer на Rx - я в этом смысла не вижу...

9
  • спасибо большое за ответ! По поводу последнего момента — я использую buffer(5), поскольку мне нужно разложить все фигурки в коробки по 5 штук. Да, они действительно раскладываются одновременно после получения пятой. Есть ли другой способ делить поток на элементы по пять штук? 13 дек 2016 в 6:20
  • что значит разложить в коробки и как этому способствует buffer?
    – xkor
    13 дек 2016 в 8:12
  • зачем буфер кажется понял, но не понял тогда как они у вас должны раскладываться с учетом задержек
    – xkor
    13 дек 2016 в 8:24
  • По условию все фигурки после покраски должны раскладаться в "коробки" (списки или любая другая группировка) в том порядке в котором мы получили их из покраски, а содержимое коробок должно выводиться на экран. buffer, кажется работает, но я не уверен, верно ли. 13 дек 2016 в 9:00
  • Еще я не указал важный момент: есть два вида фигурок, кружки и квадратики. Сервис покраски кружков должен работать со скоростью один кружок в секунду, сервис покраски квадратиков — один квадратик в две секунды. Это также, как я понимаю, должно влиять на порядок получения фигурок из покраски 13 дек 2016 в 9:02

Ваш ответ

Нажимая на кнопку «Отправить ответ», вы соглашаетесь с нашими пользовательским соглашением, политикой конфиденциальности и политикой о куки

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.