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

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

Если Вам необходимо готовое решение, обратите внимание на библиотеку Signals2, котрая включена в boost с мая 2009-го года.
  Я не пытаюсь предоставить решение, которое можно было бы использовать в готовом виде. Но тем не менее, ознакомившись с материалом, можно обойтись без использования сторонних библиотек, в тех проектах, в которых они по каким-либо причинам не доступны или нежелательны (драйвера, низкоуровневые приложения и т.п.).

Предметная область

Действующие лица

NotificationSender — объект, рассылающий сообщения.
Как правило это рабочий поток, извещающий об изменении своего состояния, которое необходимо отобразить на пользовательском интерфейсе.
NotificationListener — объект, реализующий обработку уведомлений.
Как правило это объект, который управляет отображением части пользовательского интерфейса связанного с фоновой задачей.
Таких объектов может быть множество, при этом они могут подключаться/отключаться динамически (к примеру открытие далогового окна, где показываются детали выполнения задачи)
NotificationDispatcher — объект, управляющий подписчиками и рассылкой сообщений.

Взаимодействие между объектами

Рассылка сообщений всем подписчикам.
Процесс подписки/прекшащения подписки.
Время жизни объектов.
  В данной статье описан метод синхронной рассылки сообщений. Это означает, что вызов функции SendMessage будет синхронным, и поток, вызывающий этот метод будет ожидать завершения обработки сообщений всеми подписчиками. В ряде случаев такой подход удобней ассинхронной рассылки, но при этом в нем есть трудности с прекращением подписки.

Простейшая реализация для однопоточной среды

 typedef unsigned __int64 SubscriberId;
class CSubscriber
{
public:
 virtual ~CSubscriber(){}
 virtual void MessageHandler(void* pContext) = 0;
 SubscriberId GetSubscriberId() {return (SubscriberId)this;}
};

class CDispatcher
{
private:
 typedef std::vector<CSubscriber*> CSubscriberList;
public:
 SubscriberId Subscribe(CSubscriber* pNewSubscriber)
 {
 for(size_t i = 0; i < m_SubscriberList.size(); ++i)
 {
 if(m_SubscriberList[i]->GetSubscriberId() == pNewSubscriber->GetSubscriberId())
 {
 return 0;
 }
 }
 m_SubscriberList.push_back(pNewSubscriber);
 return pNewSubscriber->GetSubscriberId();
 }
 bool Unsubscribe(SubscriberId id)
 {
 for(size_t i = 0; i < m_SubscriberList.size(); ++i)
 {
 if(m_SubscriberList[i]->GetSubscriberId() == id)
 {
 m_SubscriberList.erase(m_SubscriberList.begin() + i);
 return true;
 }
 }
 return false;
 }
 void SendMessage(void* pContext)
 {
 for(size_t i = 0; i < m_SubscriberList.size(); ++i)
 {
 m_SubscriberList[i]->MessageHandler(pContext);
 }
 }
private:
 CSubscriberList m_SubscriberList;
};

Здесь уникальный идентификатор подписчика — адресс объекта подписчика, функция GetSubscriberId возвращает всегда одинаковое значение для одного объекта подписчика в не зависимости от преобразования типов.

Пример использования

 class CListener:
 public CSubscriber
{
 virtual void MessageHandler(void* pContext)
 {
 wprintf(L"%d\n", *((int*)pContext));
 }
};
int _tmain(int argc, _TCHAR* argv[])
{
 CDispatcher Dispatcher;
 CListener Listener1;
 CListener Listener2;
 Dispatcher.Subscribe(&Listener1);
 Dispatcher.Subscribe(&Listener2);
 for(int i = 0; i < 5; ++i)
 {
 Dispatcher.SendMessage(&i);
 }
 Dispatcher.Unsubscribe(Listener2.GetSubscriberId());
 Dispatcher.Unsubscribe(Listener1.GetSubscriberId());
 return 0;
}

Отключение подписчика внутри обработчика сообщений

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

Переходим к многопоточной среде

С одним потоком такой код будет работать довольно стабильно.
Давайте посмотрим что будет при работе нескольких потоков.

 CDispatcher g_Dispatcher;
DWORD WINAPI WorkingThread(PVOID pParam)
{
 for(int i = 0;;++i)
 {
 g_Dispatcher.SendMessage(&i);
 }
};
int _tmain(int argc, _TCHAR* argv[])
{
 ::CreateThread(NULL, 0, WorkingThread, NULL, 0, NULL);
 CListener Listener1;
 CListener Listener2;
 for(;;)
 {
 g_Dispatcher.Subscribe(&Listener1);
 g_Dispatcher.Subscribe(&Listener2);

 g_Dispatcher.Unsubscribe(Listener1.GetSubscriberId());
 g_Dispatcher.Unsubscribe(Listener2.GetSubscriberId());
 }
 return 0;
}

Рано или позно произойдет креш.
Проблема заключается в добавлении/удалении подписчиков и одновременной рассылке уведомлений (многопоточный доступ к CDispatcher::m_SubscriberList в нашем примере).
  Здесь необходима синхронизация доступа к списку подписчиков.

Синхронизация доступа к списку подписчиков

 class CDispatcher
{
private:
 typedef std::vector<CSubscriber*> CSubscriberList;
public:
 SubscriberId Subscribe(CSubscriber* pNewSubscriber)
 {
 CScopeLocker ScopeLocker(m_Lock);
 for(size_t i = 0; i < m_SubscriberList.size(); ++i)
 {
 if(m_SubscriberList[i]->GetSubscriberId() == pNewSubscriber->GetSubscriberId())
 {
 return 0;
 }
 }
 m_SubscriberList.push_back(pNewSubscriber);
 return pNewSubscriber->GetSubscriberId();
 }
 bool Unsubscribe(SubscriberId id)
 {
 CScopeLocker ScopeLocker(m_Lock);
 for(size_t i = 0; i < m_SubscriberList.size(); ++i)
 {
 if(m_SubscriberList[i]->GetSubscriberId() == id)
 {
 m_SubscriberList.erase(m_SubscriberList.begin() + i);
 return true;
 }
 }
 return false;
 }
 void SendMessage(void* pContext)
 {
 CScopeLocker ScopeLocker(m_Lock);
 for(size_t i = 0; i < m_SubscriberList.size(); ++i)
 {
 m_SubscriberList[i]->MessageHandler(pContext);
 }
 }
private:
 CSubscriberList m_SubscriberList;
 CLock m_Lock;
};

Синхронизация доступа была реализована при помощи объектов синхронизации (Critical section или Mutex).
Для большей переносимости и для того, чтобы не отвлекаться от сути происходящего, абстрагируемся от прямых вызовов платформенно-зависимых функций, типа EnterCriticalSection. Для этого служит класс CLock.
Для устойчивости к с++ исключениям удобно использовать технологию RAII, а именно класс CScopeLocker, который в конструкторе захватывает объект синхронизации, а в деструкторе освобождает его.
  При такой реализации программа не будет падать, но нас поджидает еще одна неприятная ситуация.

Борьба с взаимной блокировкой потоков (deadlock)

Допустим у нас есть некий поток, выполняющий какую-то фоновую задачу и есть окно, где отображается ход выполнения этой задачи.
Как правило, поток посылает уведомление классу окна, который в свою очередь вызывает системную функцию SendMessage, которая инициирует какие-то действия в контексте оконной процедуры.
Системная функция SendMessage является блокирующей, она отсылает сообщение потоку окна и ждет пока тот его обработает.
Если подключение/отключение объекта-слушателя будет происходить также в контексте оконной процедуры (в потоке окна) возможна взаимная блокировка потоков, так называемый deadlock.
Такой deadlock может воспроизоводится крайне редко (в момент вызова Subscribe/Unsubscribe и одновременном вызове MessageHandler в отдельном потоке)
  Следующий код эмулирует ситуацию с блокирующим вызовом системной ф-ции SendMessage.

 CDispatcher g_Dispatcher;
CLock g_Lock;
class CListener:
 public CSubscriber
{
 virtual void MessageHandler(void* pContext)
 {
 //Эмулируем блокирующий вызов SendMessage
 g_Lock.Lock();
 wprintf(L"%d\n", *((int*)pContext));
 g_Lock.Unlock();
 }
};
DWORD WINAPI WorkingThread(PVOID pParam)
{
 for(int i = 0;;++i)
 {
 g_Dispatcher.SendMessage(&i);
 }
};
int _tmain(int argc, _TCHAR* argv[])
{
 ::CreateThread(NULL, 0, WorkingThread, NULL, 0, NULL);
 CListener Listener1;
 CListener Listener2;
 for(;;)
 {
 //Эмулируем контекст оконной процедуры (обработчик оконного сообщения)
 g_Lock.Lock();
 g_Dispatcher.Subscribe(&Listener1);
 g_Dispatcher.Subscribe(&Listener2);
 g_Lock.Unlock();
 Sleep(0);
 g_Lock.Lock();
 g_Dispatcher.Unsubscribe(Listener1.GetSubscriberId());
 g_Dispatcher.Unsubscribe(Listener2.GetSubscriberId());
 g_Lock.Unlock();
 }
 return 0;
}

Проблема заключается в том, что главный поток захватывает глобальный объект синхронизации g_Lock (при аналогии с оконной процедурой — выполняется в контексте оконного потока), и затем вызывает метод Subscribe/Unsubscribe, который внутри пытается захватить второй объект синхронизации CDispatcher::m_Lock.
  В этот момент рабочий поток посылает уведомление, захватив при этом CDispatcher::m_Lock в функции CDispatcher::SendMessage, и затем пытается захватить глобальный объект синхронизации g_Lock (при аналогии с оконом — вызывает системную функцию SendMessage).

Поток окна A -> B
Рабочий поток B -> A

Это можно назвать класическим deadlock-ом.
Проблема скрывается в функции CDispatcher::SendMessage().
Здесь должно соблюдаться правило — нельзя вызывать callback-функцию захватив при этом какой-либо объект синхронизации.
  Итак, убираем блокировку при рассылке уведомлений.

 void SendMessage(void* pContext)
{
 CSubscriberList SubscriberList;
 {
 CScopeLocker ScopeLocker(m_Lock);
 SubscriberList = m_SubscriberList;
 }
 for(size_t i = 0; i < SubscriberList.size(); ++i)
 {
 SubscriberList[i]->MessageHandler(pContext);
 }
}

Контроль времени жизни подписчиков

После того, как мы убрали deadlock, появилась другая проблема — время жизни объектов-подписчиков.
У нас больше нет гарантии, что метод MessageHandler не будет вызван после вызова Unsubscribe, и по этому мы не можем удалять объект-подписчик непосредственно после вызова Unsubscribe.
В данной ситуации проще всего контролировать время жизни объектов-подписчиков с использованием счетчика ссылок.
Для этого можно исползовать технологию COM — унаследовать интерфейс CSubscriber от IUnknown и использовать ATL CComPtr для списка объектов-подписчиков, тоесть заменить std::vector<CSubscriber*> на std::vector<CComPtr>.
Но такая реализация чревата дополнительными расходами на реализацию классов-подписчиков, так как в каждом из них должны быть реализованы методы AddRef/Release и ненужный QueryInterface, хотя если в проекте активно используется COM, то такой подход может иметь приемущество.
  Для контроля времени жизни объектов-подписчиков с исползованием счетчика ссылок хорошо подойдут умные указатели.

Простая реализация для многопоточной среды

 typedef unsigned __int64 SubscriberId;
class CSubscriber
{
public:
 virtual ~CSubscriber(){}
 virtual void MessageHandler(void* pContext) = 0;
 SubscriberId GetSubscriberId() {return (SubscriberId)this;}
};
typedef boost::shared_ptr<CSubscriber> CSubscriberPtr;

class CDispatcher
{
private:
 typedef std::vector<CSubscriberPtr> CSubscriberList;
public:
 SubscriberId Subscribe(CSubscriberPtr pNewSubscriber)
 {
 CScopeLocker ScopeLocker(m_Lock);
 for(size_t i = 0; i < m_SubscriberList.size(); ++i)
 {
 if(m_SubscriberList[i]->GetSubscriberId() == pNewSubscriber->GetSubscriberId())
 {
 return 0;
 }
 }
 m_SubscriberList.push_back(pNewSubscriber);
 return pNewSubscriber->GetSubscriberId();
 }
 bool Unsubscribe(SubscriberId id)
 {
 CSubscriberPtr toRelease;
 CScopeLocker ScopeLocker(m_Lock);
 for(size_t i = 0; i < m_SubscriberList.size(); ++i)
 {
 if(m_SubscriberList[i]->GetSubscriberId() == id)
 {
 toRelease = m_SubscriberList[i];
 m_SubscriberList.erase(m_SubscriberList.begin() + i);
 return true;
 }
 }
 return false;
 }
 void SendMessage(void* pContext)
 {
 CSubscriberList SubscriberList;
 {
 CScopeLocker ScopeLocker(m_Lock);
 SubscriberList = m_SubscriberList;
 }
 for(size_t i = 0; i < SubscriberList.size(); ++i)
 {
 SubscriberList[i]->MessageHandler(pContext);
 }
 }
private:
 CSubscriberList m_SubscriberList;
 CLock m_Lock;
};

В данной реализации я заменил «голый» указатель CSubscriber* на «умный» указатель со счетчиком ссылок, такой оказался в библиотеке boost.
Также в функцию Unsubscribe я добавил переменную toRelease для того, чтобы вызвать деструктор объекта-подписчика уже после вызова Unlock (нельзя вызывать callback-функцию, включая деструктор объекта подписчика, захватив при этом какой-либо объект синхронизации).
  Cтоит обратить внимание на то, что в функции SendMessage происходит копирование списка умных указателей (после копирования все указатели увеличивают свои счетчики ссылок, а при выходе из функции уменьшают, что и контролирует время жизни объектов-подписчиков)

Тестируем

 CDispatcher g_Dispatcher;
CLock g_Lock;
class CListener:
 public CSubscriber
{
 virtual void MessageHandler(void* pContext)
 {
 //Эмулируем блокирующий вызов SendMessage
 g_Lock.Lock();
 wprintf(L"%d\n", *((int*)pContext));
 g_Lock.Unlock();
 }
};
DWORD WINAPI WorkingThread(PVOID pParam)
{
 for(int i = 0;;++i)
 {
 g_Dispatcher.SendMessage(&i);
 }
};
int _tmain(int argc, _TCHAR* argv[])
{
 ::CreateThread(NULL, 0, WorkingThread, NULL, 0, NULL);
 for(;;)
 {
 boost::shared_ptr<CListener> pListener1(new CListener);
 boost::shared_ptr<CListener> pListener2(new CListener);
 //Эмулируем контекст оконной процедуры (обработчик оконного сообщения)
 g_Lock.Lock();
 g_Dispatcher.Subscribe(pListener1);
 g_Dispatcher.Subscribe(pListener2);
 g_Lock.Unlock();
 Sleep(0);
 g_Lock.Lock();
 g_Dispatcher.Unsubscribe(pListener1->GetSubscriberId());
 g_Dispatcher.Unsubscribe(pListener2->GetSubscriberId());
 g_Lock.Unlock();
 }
 return 0;
}

Соптимизированная реализация для многопоточной среды

Как правило вызов функции SendMessage будет происходить намного чаще чем Subscribe/Unsubscribe. При большом количестве подписчиков узким местом может стать копирование списка подписчиков внутри SendMessage.
Копирование списка подписчиков можно перенести в функции Subscribe/Unsubscribe. Это будет похоже на методику из lock-free алгоритмов.
Объект CDispatcher будет хранить список подписчиков не на прямую, а при помощи умного указателя. Внутри функции SendMessage мы будем получать указатель на текущий список подписчиков и работать с ним. В функциях Subscribe/Unsubscribe мы будем каждый раз создавать новый список подписчиков и перенаправлять указатель внутри объекта CDispatcher на новый список подписчиков. Таким образом в то время, когда указатель на список подписчиков в объекте CDispatcher будет указывать уже на новый список подписчиков, ф-ция SendMessage по прежнему будет работать со старым списком. Так как старый список подписчиков никто не изменяет, то все будет работать стабильно в многопоточной среде.
В принципе, можно несколько модифицировать функции Subscribe/Unsubscribe и реализовать полностью lock-free алгоритм, но это уже другая тема.
Медот Unsubscribe является асинхронным и не гарантирует после своего завершения полное прекращение рассылки, половинное решение — подписчик получает уведомление о прекращении подписки при помощи ф-ции UnsubscribeHandler. Для реализации этого поведения добавлен промежуточный класс CSubscriberItem, который в своем деструкоторе вызывает ф-цию UnsubscribeHandler.

 namespace Observer
{
 //////////////////////////
 // Subscriber
 //////////////////////////
 typedef unsigned __int64 SubscriberId;
 class CSubscriber
 {
 public:
 virtual ~CSubscriber(){}
 virtual void MessageHandler(void* pContext) = 0;
 virtual void UnsubscribeHandler() = 0;
 SubscriberId GetSubscriberId() {return (SubscriberId)this;}
 };
 typedef boost::shared_ptr<CSubscriber> CSubscriberPtr;

 //////////////////////////////////////////////////////////////////////
 // Dispatcher
 ///////////////////////////////////
 class CDispatcher
 {
 private:
 class CSubscriberItem
 {
 public:
 CSubscriberItem(CSubscriberPtr pSubscriber)
 :m_pSubscriber(pSubscriber)
 {
 }
 ~CSubscriberItem()
 {
 m_pSubscriber->UnsubscribeHandler();
 };
 CSubscriberPtr Subscriber()const {return m_pSubscriber;}
 private:
 CSubscriberPtr m_pSubscriber;
 };
 typedef boost::shared_ptr<CSubscriberItem> CSubscriberItemPtr;
 typedef std::vector<CSubscriberItemPtr> CSubscriberList;
 typedef boost::shared_ptr<CSubscriberList> CSubscriberListPtr;
 public:
 CDispatcher()
 {
 }
 private:
 CDispatcher(const CDispatcher&){}
 CDispatcher& operator=(const CDispatcher&){return *this;}
 public:
 SubscriberId Subscribe(CSubscriberPtr pNewSubscriber)
 {
 //Declaration of the next shared pointer before ScopeLocker
 //prevents release of subscribers from under lock
 CSubscriberListPtr pNewSubscriberList(new CSubscriberList());
 //Enter to locked section
 CScopeLocker ScopeLocker(m_Lock);
 if(m_pSubscriberList)
 {
 //Copy existing subscribers
 pNewSubscriberList->assign(m_pSubscriberList->begin(), m_pSubscriberList->end());
 }
 for(size_t i = 0; i < pNewSubscriberList->size(); ++i)
 {
 CSubscriberItemPtr pSubscriberItem = (*pNewSubscriberList)[i];
 if(pSubscriberItem->Subscriber()->GetSubscriberId() == pNewSubscriber->GetSubscriberId())
 {
 return 0;
 }
 }
 //Add new subscriber to new subscriber list
 pNewSubscriberList->push_back(CSubscriberItemPtr(new CSubscriberItem(pNewSubscriber)));
 //Exchange subscriber lists
 m_pSubscriberList = pNewSubscriberList;
 return pNewSubscriber->GetSubscriberId();
 }
 bool Unsubscribe(SubscriberId id)
 {
 //Declaration of the next shared pointers before ScopeLocker
 //prevents release of subscribers from under lock
 CSubscriberItemPtr pSubscriberItemToRelease;
 CSubscriberListPtr pNewSubscriberList;
 //Enter to locked section
 CScopeLocker ScopeLocker(m_Lock);
 if(!m_pSubscriberList)
 {
 //No subscribers
 return false;
 }
 pNewSubscriberList = CSubscriberListPtr(new CSubscriberList());
 for(size_t i = 0; i < m_pSubscriberList->size(); ++i)
 {
 CSubscriberItemPtr pSubscriberItem = (*m_pSubscriberList)[i];
 if(pSubscriberItem->Subscriber()->GetSubscriberId() == id)
 {
 pSubscriberItemToRelease = pSubscriberItem;
 }
 else
 {
 pNewSubscriberList->push_back(pSubscriberItem);
 }
 }
 //Exchange subscriber lists
 m_pSubscriberList = pNewSubscriberList;
 if(!pSubscriberItemToRelease.get())
 {
 return false;
 }
 return true;
 }
 void SendMessage(void* pContext)
 {
 CSubscriberListPtr pSubscriberList;
 {
 CScopeLocker ScopeLocker(m_Lock);
 if(!m_pSubscriberList)
 {
 //No subscribers
 return;
 }
 //Get shared pointer to an existing list of subscribers
 pSubscriberList = m_pSubscriberList;
 }
 //pSubscriberList pointer to copy of subscribers' list
 for(size_t i = 0; i < pSubscriberList->size(); ++i)
 {
 (*pSubscriberList)[i]->Subscriber()->MessageHandler(pContext);
 }
 }
 private:
 CSubscriberListPtr m_pSubscriberList;
 CLock m_Lock;
 };

}; //namespace Observer

Ссылки

Библиотека boost::signals2 статья
Умные указатели Джефф Элджер
Resource Acquisition Is Initialization (RAII) википедия
Комментарии к первой версии этой статьи можно найти здесь