觀察者模式的應用，主要的行為就是註冊和移除觀察者(observer),以及通知所有已註冊的Observers。這裡介紹的是Chromium專案中實現的執行緒安全的觀察者管理及通知的基礎類ObserverListThreadSafe, 它的能力包括:

• 觀察者可以在任意執行緒中註冊，訊息回撥會發生在註冊時所在的執行緒。
• 任意執行緒可以Notify()觸發通知訊息。
• 觀察者可以在回撥時從列表中移除自己。
• 如果一個執行緒正在通知觀察者, 此時一個觀察者正在從列表中移除自己, 通知會被丟棄。

執行緒安全的基礎

實現這個基礎是記錄執行緒與觀察者列表的對應關係，即某個執行緒上存在的觀察者的列表。定義如下:

typedef std::map<base::PlatformThreadId, ObserverListContext*>
      ObserversListMap;

其中PlatformThreadID為執行緒的ID, 而ObserverListContext是一個數組，其定義如下:

  struct ObserverListContext {
    scoped_refptr<base::MessageLoopProxy> loop;
    ObserverList<ObserverType> list;
  };

其中loop為Chromium執行緒機制中的MessageLoopProxy, 也可以理解為執行緒的訊息佇列代理，使用它就可以完成將某個操作拋到指定執行緒上執行。list就很好理解了，記錄的是該執行緒中的觀察者列表。

只要全域性的持有這個表(ObserversListMap)，就可以將觀察者與執行緒關聯起來，進而保證通知一定可以執行到它註冊時所在的執行緒。

設計總覽及使用方式

其中ObserverListThreadSafe就是我們的主角。它本身是為一個模板類:

template <class ObserverType>
class ObserverListThreadSafe
    : public base::RefCountedThreadSafe<
        ObserverListThreadSafe<ObserverType>,
        ObserverListThreadSafeTraits
 
<ObserverType> > {

下面是一個使用的示例:

  // The interface to receive mouse movement events.
  class MEDIA_EXPORT MouseEventListener {
   public:
    // |position| is the new mouse position.
    virtual void OnMouseMoved(const SkIPoint& position) = 0;

   protected:
    virtual ~MouseEventListener() {}
  };
  typedef ObserverListThreadSafe<UserInputMonitor::MouseEventListener>
      MouseListenerList;
scoped_refptr<MouseListenerList> mouse_listeners_;

新增/刪除Observer方法很簡單:

void UserInputMonitor::AddMouseListener(MouseEventListener* listener) {
  mouse_listeners_->AddObserver(listener);
  ......
}

void UserInputMonitor::RemoveMouseListener(MouseEventListener* listener) {
  mouse_listeners_->RemoveObserver(listener);
  ......
}

當需要通知各個觀察者時，程式碼如下:

SkIPoint position(SkIPoint::Make(event->u.keyButtonPointer.rootX,
                                     event->u.keyButtonPointer.rootY));
    mouse_listeners_->Notify(
        &UserInputMonitor::MouseEventListener::OnMouseMoved, position);

ObserverListThreadSafe實現要點

操作執行緒-觀察者列表時加鎖

即內部成員變數list_lock_，在操作observer_list_要使用如下方法加鎖:

base::AutoLock lock(list_lock_);

新增及刪除觀察者

基本思路就是:
* 以當前執行緒ID，找到ObserverListContext。如果新增但又沒有，則新建。
* 操作找到的ObserverListContext進行新增或刪除操作。
下面是AddObserver()的實現:

  // Add an observer to the list.  An observer should not be added to
  // the same list more than once.
  void AddObserver(ObserverType* obs) {
    // If there is not a current MessageLoop, it is impossible to notify on it,
    // so do not add the observer.
    if (!base::MessageLoop::current())
      return;

    ObserverList<ObserverType>* list = NULL;
    base::PlatformThreadId thread_id = base::PlatformThread::CurrentId();
    {
      base::AutoLock lock(list_lock_);
      if (observer_lists_.find(thread_id) == observer_lists_.end())
        observer_lists_[thread_id] = new ObserverListContext(type_);
      list = &(observer_lists_[thread_id]->list);
    }
    list->AddObserver(obs);
  }

事件通知

這個過程實現上，因為需要相容不同數理的引數，所以定義了一組模板方法。先說明一下基本思路:
* 封裝要呼叫的通知方法為Callback形式:函式，及使用tuple(不是C++11的元組，而是base::tuple)封裝起來的引數。
* 遍歷observer_lists_，找到每個執行緒對應的ObserverListContext。
* 使用ObserverListContext中記錄的MessageProxyLoop，執行NotifyWrapper，並傳入Callback作為引數。
以上這個過程就是在通知的執行緒的完成的, 具體的程式碼如下：

  template <class Method, class Params>
  void Notify(const UnboundMethod<ObserverType, Method, Params>& method) {
    base::AutoLock lock(list_lock_);
    typename ObserversListMap::iterator it;
    for (it = observer_lists_.begin(); it != observer_lists_.end(); ++it) {
      ObserverListContext* context = (*it).second;
      context->loop->PostTask(
          FROM_HERE,
          base::Bind(&ObserverListThreadSafe<ObserverType>::
              template NotifyWrapper<Method, Params>, this, context, method));
    }
  }

下一步就是在各個observer所在的執行緒上觸發NotifyWrapper了。它主要做兩件事:
* 為每一個observer執行通知的函式:

    {
      typename ObserverList<ObserverType>::Iterator it(context->list);
      ObserverType* obs;
      while ((obs = it.GetNext()) != NULL)
        method.Run(obs);
    }

• 如果發現這個執行緒上已經沒有可用的觀察者，則將它從observer_list_中移除。

封裝的通知方法

再回頭看一下對回撥方法封裝，上面提到了引數是使用base::tuple封裝的，它同時也提供了DispatchToMethod方法，把引數解開，再呼叫方法，詳見base::tuple中的說明。
下面是UnboundMethod的定義:

// An UnboundMethod is a wrapper for a method where the actual object is
// provided at Run dispatch time.
template <class T, class Method, class Params>
class UnboundMethod {
 public:
  UnboundMethod(Method m, const Params& p) : m_(m), p_(p) {
    COMPILE_ASSERT(
        (base::internal::ParamsUseScopedRefptrCorrectly<Params>::value),
        badunboundmethodparams);
  }
  void Run(T* obj) const {
    DispatchToMethod(obj, m_, p_);
  }
 private:
  Method m_;
  Params p_;
};

如果沒有Chromium的執行緒機制，也是可以實現的，核心是執行緒的拋轉。

小結

這種方法適用於多執行緒下以函式+引數通知的方式。引數直接拋轉到指定的執行緒不用特別擔心執行緒安全問題。對於獲得通知後仍然要跨執行緒訪問資料的情況，則可以考慮: 1.以類似的方式，將資料通過函式引數傳遞(目前最多為6個)。 2. 如果資料量大，則可以考慮使用Multiversion Concurrency Control的演算法，儘量避免加鎖的開銷。