Шаблон наблюдателя с использованием weak_ptr

Я пытаюсь написать безопасный класс Subject из шаблон наблюдателя. Я хочу знать, является ли использование weak_ptr лучшим способом хранения экземпляров IObserver таким образом, чтобы:

  • Невозможно использовать экземпляр IObserver после того, как он был свободен.
  • Класс Subject не поддерживает IObserver ссылки, которые должны быть бесплатными (проблема с истекшим слушателем).
  • Класс Subject должен быть потокобезопасным.

К сожалению, наши стандарты кодирования говорят, что нам не разрешено использовать boost. Наверное, я был плохим человеком в предыдущей жизни. К счастью, мне разрешено использовать С++ 11 (что поставляется с Visual Studio 2012).

Вот пример класса Observer.

// Observer interface that supports notify() method
class IObserver
{
public:
    virtual void notify() const = 0;
    virtual ~IObserver() {}
};

// Concrete observer implementation that prints a message
class Observer : public IObserver
{
public:
    Observer( const std::string& message) : m_message( message ){}

    void notify() const {
        printf( "%s\r\n", m_message.c_str() );
    }

private:
    std::string m_message;
};

И вот класс Subject.

// Subject which registers observers and notifies them as needed.
class Subject
{
public:
    // Use shared_ptr to guarantee the observer is valid right now
    void registerObserver( const std::shared_ptr<IObserver>& o )
    {
        std::lock_guard<std::mutex> guard( m_observersMutex );
        m_observers.push_back( o );
    }

    void unregisterObserver( const std::shared_ptr<IObserver>& o )
    {
        std::lock_guard<std::mutex> guard( m_observersMutex );
        // Code to remove the observer from m_observersMutex
    }

    // This is a method that is run in its own thread that notifies observers of some event
    void doNotify()
    {
        std::lock_guard<std::mutex> guard( m_observersMutex );
        // Notify any valid observers of events.
        std::for_each( m_observers.cbegin(), m_observers.cend(), 
            []( const std::weak_ptr<IObserver>& o )
        {
            auto observer = o.lock();
            if ( observer ) {
                observer->notify();
            } 
        } );

        // Remove any dead observers.  These are ones which have expired().
        m_observers.erase( std::remove_if( m_observers.begin(), m_observers.end(), 
            []( const std::weak_ptr<IObserver>& o )
        {
            return o.expired();
        } ), m_observers.end() );

    }


private:
    std::vector<std::weak_ptr<IObserver>> m_observers;
    std::mutex m_observersMutex;
};

Вот код, который выполняет Subject:

int main(int argc, wchar_t* argv[])
{

    Subject subject;
    auto observerHello = std::make_shared<Observer>( "Hello world" );
    subject.registerObserver( observerHello );
    {
        // Create a scope to show unregistration.
        auto observerBye = std::make_shared<Observer>( "Good bye" );
        subject.registerObserver( observerBye );

        subject.doNotify();
    }
    printf( "%s\r\n", "Observer good bye is now be destructed" );
    subject.doNotify();
    return 0;
}

Я использую поток weak_ptr в потоковом режиме? Отсюда qaru.site/info/406688/... Я думаю, что это так.

Является ли это законным способом решения проблемы с задержкой слушателя?

Ответы

Ответ 1

Я бы немного остерегался вашего doNotify - предположил, что что-то в наблюдателе, которого вы стреляете, добавляет или удаляет наблюдателей? - происходят плохие вещи (включая аварии). Или блокирование действия другого потока, который блокирует попытку добавления наблюдателя? - происходят плохие вещи (тупики!)

Это сложно решить. В принципе, это проблема с повторной установкой.

Никогда не оставляйте контроль над своим кодом, когда вы держите замок. Удержание блокировки при вызове обратного вызова - нет-нет.

Итак, как минимум:

Заблокировать, затем скопировать список, затем разблокировать. При выполнении этой копии вы также можете удалить истекшие наблюдатели (как из оригинала, так и из копии, списка).

Затем запустите наблюдателей из скопированного списка.

Это оставляет некоторые проблемы нерешенными. Например, тот факт, что удаление наблюдателя не гарантирует, что он не будет вызван в будущем! Это просто означает, что в конечном итоге он не будет называться.

Насколько важно это зависит от того, как вы используете прослушивание.

Один из подходов, который может работать, - это очередь задач, которая включает события add/remove/notify/killthread (что делает killthread задачей в очереди делает закрытие гораздо менее раздражающим). Теперь вся синхронизация находится в очереди. Если вы не готовы к написанию незаблокированной незаблокированной очереди, код уведомления может просто заблокировать, std::move очередь, разблокировать, а затем продолжить ее выполнение. Или вы могли бы написать такую ​​очередь, чтобы pop блокировалось, пока что-то не прочитано, а push не блокируется.

Быстрая и грязная "копия и трансляция" может выглядеть так:

std::vector<std::shared_ptr<IObserver>> targets;
{
  std::lock_guard<std::mutex> guard( m_observersMutex );
  m_observers.erase( std::remove_if( m_observers.begin(), m_observers.end(), 
        [&targets]( const std::weak_ptr<IObserver>& o )
    {
      std::shared_ptr<IObserver> ptr = o.lock();
      if (ptr) {
        targets.push_back(ptr);
        return false;
      } else {
        return true;
      }
    } ), m_observers.end() );
}

for( auto& target:targets ) {
  target->notify();
}