Шаблон дизайна наблюдателя

Question

В шаблоне Observer Design субъект уведомляет всех наблюдателей, вызывая операцию update() каждого наблюдателя. Один из способов сделать это -

void notify() {
   for (observer: observers) {
      observer.update(this);
   }
}

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

Вопрос:

1. Есть ли способ обойти эту проблему?
2. Если так, что было бы хорошим примером?

Answer 1

Проблема в бесконечном цикле, а не в уведомлениях «один за другим».

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

Жаловаться на то, что один бесконечный цикл останавливает другие обновления, - все равно, что жаловаться на то, что взятие блокировки, а затем переход в бесконечный цикл не дает другим получить доступ к заблокированному объекту - проблема в бесконечном цикле, а не в диспетчере блокировки.

Answer 2

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

Answer 3

Вы можете использовать метод java.utils.concurrent.Executors.newFixedThreadPool (int nThreads), а затем вызвать метод invokeAll (также можно использовать метод с тайм-аутом, чтобы избежать бесконечного цикла).

Вы должны изменить свой цикл, чтобы добавить класс Callable, который принимает «наблюдатель» и «this», а затем вызвать метод update в методе «call».

Посмотрите на этот пакет для получения дополнительной информации .

Это быстрая и грязная реализация того, о чем я говорил:

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Main
{
    private Main()
    {
    }

    public static void main(final String[] argv)
    {
        final Watched       watched;
        final List<Watcher> watchers;

        watched = new Watched();
        watchers = makeWatchers(watched, 10);
        watched.notifyWatchers(9);
    }

    private static List<Watcher> makeWatchers(final Watched watched,
                                              final int     count)
    {
        final List<Watcher> watchers;

        watchers = new ArrayList<Watcher>(count);

        for(int i = 0; i < count; i++)
        {
            final Watcher watcher;

            watcher = new Watcher(i + 1);
            watched.addWatcher(watcher);
            watchers.add(watcher);
        }

        return (watchers);
    }
}

class Watched
{
    private final List<Watcher> watchers;

    {
        watchers = new ArrayList<Watcher>();
    }

    public void addWatcher(final Watcher watcher)
    {
        watchers.add(watcher);
    }

    public void notifyWatchers(final int seconds)
    {
        final List<Watcher>         currentWatchers;
        final List<WatcherCallable> callables;
        final ExecutorService       service;

        currentWatchers = new CopyOnWriteArrayList<Watcher>(watchers);
        callables       = new ArrayList<WatcherCallable>(currentWatchers.size());

        for(final Watcher watcher : currentWatchers)
        {
            final WatcherCallable callable;

            callable = new WatcherCallable(watcher);
            callables.add(callable);
        }

        service = Executors.newFixedThreadPool(callables.size());

        try
        {
            final boolean value;

            service.invokeAll(callables, seconds, TimeUnit.SECONDS);
            value = service.awaitTermination(seconds, TimeUnit.SECONDS);
            System.out.println("done: " + value);
        }
        catch (InterruptedException ex)
        {
        }

        service.shutdown();
        System.out.println("leaving");
    }

    private class WatcherCallable
        implements Callable<Void>
    {
        private final Watcher watcher;

        WatcherCallable(final Watcher w)
        {
            watcher = w;
        }

        public Void call()
        {
            watcher.update(Watched.this);
            return (null);
        }
    }
}

class Watcher
{
    private final int value;

    Watcher(final int val)
    {
        value = val;
    }

    public void update(final Watched watched)
    {
        try
        {
            Thread.sleep(value * 1000);
        }
        catch (InterruptedException ex)
        {
            System.out.println(value + "interupted");
        }

        System.out.println(value + " done");
    }
}

Answer 4

Меня больше беспокоит наблюдатель, генерирующий исключение, чем бесконечный цикл. Ваша текущая реализация не уведомит остальных наблюдателей о таком событии.

Answer 5

1. Есть ли способ обойти эту проблему?

Да, убедитесь, что наблюдатель работает нормально и своевременно возвращайтесь.

2. Может кто-нибудь, пожалуйста, объясните это на примере.

Sure:

class ObserverImpl implements Observer {
     public void update( Object state ) {
            // remove the infinite loop.
            //while( true ) {
            //   doSomething();
            //}

            // and use some kind of control:
            int iterationControl = 100;
            int currentIteration = 0;
            while( curentIteration++ < iterationControl ) {
                 doSomething();
            }
     }
     private void doSomething(){}
}

Это предотвращает бесконечный цикл (если это имеет смысл, он должен выполняться не более 100 раз)

Другим механизмом является запуск новой задачи во втором потоке, но если он входит в бесконечный цикл, он в конечном итоге потребляет всю системную память:

class ObserverImpl implements Observer {
     public void update( Object state ) {
         new Thread( new Runnable(){ 
             public void run() {
                 while( true ) {
                     doSomething();
                 }
             }
          }).start();
     }
     private void doSomething(){}
}

Это заставит тот экземпляр наблюдателя немедленно вернуться, но это будет всего лишь иллюзия, что вам на самом деле нужно будет избежать бесконечного цикла.

Наконец, если ваши наблюдатели работают нормально, но вы просто хотите уведомить их всех раньше, вы можете взглянуть на этот связанный вопрос: Вызвать код после выполнения всех прослушивателей событий мыши. *

Answer 6

Если у вас есть наблюдатель с «бесконечным циклом», он больше не является паттерном наблюдателя.

Вы можете запустить разные потоки для каждого наблюдателя, но наблюдателям ДОЛЖНО быть запрещено изменять состояние наблюдаемого объекта.

Самый простой (и глупый) метод - просто взять ваш пример и сделать его многопоточным.

void notify() {
   for (observer: observers) {
      new Thread(){
          public static void run() {
              observer.update(this);
          } 
      }.start();
   }
}

(это было написано вручную, не проверено и, возможно, содержит ошибку или пять - в любом случае, это плохая идея)

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

Таким образом, чтобы решить проблему с одновременным запуском всех протекторов, используйте ThreadPoolExecutor, поскольку он A) будет перерабатывать потоки, а B) может ограничивать максимальное количество работающих потоков.

Это не является детерминированным в вашем случае "Цикл навсегда", поскольку каждый цикл навсегда будет постоянно поглощать один из потоков из вашего пула.

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

Если вам нужно поддерживать классы, которые не могут измениться, но вы можете определить, какие из них будут работать быстро, а какие - «Навсегда» (в компьютерном смысле я думаю, что это больше, чем секунда или две), тогда вы МОЖЕТЕ использовать такой цикл:

void notify() {
   for (observer: observers) {
      if(willUpdateQuickly(observer))
          observer.update(this);
      else
          new Thread(){
              public static void run() {
                  observer.update(this);
              } 
          }.start();
   }
}

Эй, если это на самом деле "Loops forever", будет ли он потреблять поток для каждого уведомления? Это действительно звучит так, как будто вам, возможно, придется потратить немного больше времени на ваш дизайн.

Answer 7

Все наблюдатели получают уведомление, вот и вся гарантия, которую вы получаете.

Если вы хотите реализовать какой-то необычный порядок, вы можете сделать это:

• Подключите только одного наблюдателя;
• пусть этот основной наблюдатель уведомит своих друзей в порядке, который вы определили в коде или каким-либо другим способом.

Это отвлекает вас от классического паттерна Observer в том, что ваши слушатели зашиты, но если это то, что вам нужно ... сделайте это!