线程安全的对象生命周期管理 —— Muduo 网络库笔记（1）

声明：本文为个人阅读《Linux 多线程服务端编程：使用 muduo C++ 网络库》的读书笔记，文中部分标题可能有些更改，但主体内容和书中一致。如有侵权，告知即删。

1.0 当析构函数遇到多线程

1. 在即将析构一个对象时，如何得知是否有别的线程正在被析构？
2. 如何保证执行对象成员函数时，保证对象不会被另一个线程析构？
3. 在调用某个对象成员函数之前，如何得知该对象是否存活？

1.1 成员函数的线程安全非常简单

以一个最简单的 Counter 示例，只需要使用同步原语保护类的内部状态，就可以编写线程安全的类了。

class Counter : boost::noncopyable {
 public:
  Counter() : value_(0) {}

  int value() const {
    std::lock_guard<std::mutex> lock(mutex_);
    return value_;
  }

  void increment() {
    std::lock_guard<std::mutex> lock(mutex_);
    value_++;
  }

 private:
  int value_;
  mutable std::mutex mutex_;
};

思考：如果 mutex_ 是 static，是否影响正确性/性能？

1. 正确性影响：若 mutex_ 是 static，则所有 Counter 实例共享同一个互斥锁。当一个实例的 increment() 或 value() 被调用时，即使它们操作的是不同的 value_，其他实例的线程也会被阻塞。
2. 性能影响：所有 Counter 实例的线程操作必须串行执行，无法并行处理不同实例的请求。

1.2 对象的创建很简单（二段式构造即可保证线程安全）

构造函数执行期间对象还没有完成初始化。要做到构造对象的线程安全，唯一的要求就是在构造期间不要泄露 this 指针，哪怕是构造函数的最后一行也不行。所以为了确保构造安全，一般使用二段式构造——即构造函数 + initialize()。

原因：

1. 如果 this 泄露，那么别的线程有可能访问这个半成品对象，出现未定义行为。
2. 最后一行也不能泄露 this，如果构造的是一个基类，虽然基类构造完成，但可能继续执行派生类的构造函数，使用 this 仍然不安全。

1.3 销毁太难（数据成员的 mutex 不能保护析构）

普通成员函数使用 mutex_ 即可保证线程安全。但析构函数违反了这一点，析构函数会把 mutex_ 成员变量销毁！

下面的例子中，如果在 (1) 处进行了资源回收，mutex_ 也被释放了。如果有其他线程正在等待 mutex_，则会发生未定义的事。（阻塞？进入临界区？core dump？都有可能）

Foo::~Foo() {
  std::lock_guard<std::mutex> lock(mutex_);
  // (1)
}

void Foo::update() {
  std::lock_guard<std::mutex> lock(mutex_); // (2)
}

我们可以得出结论：数据成员的 mutex 不能保护析构。当只有一个线程访问对象时，析构才能线程安全。

1.4 对相同类型的对象同时加锁可能也会死锁

书中还说到另一种可能死锁的场景：对相同类型的对象同时加锁。

void swap(Counter& a, Counter &b) {
  std::lock_guard<std::mutex> a_lock(a.mutex_);
  std::lock_guard<std::mutex> b_lock(b.mutex_);
  
  std::swap(a.value_, b.value_);
}

一个函数如果要锁住相同类型的多个对象，为了始终按照相同的顺序加锁，可以比较 mutex_ 地址，始终先加锁地址较小的 mutex_。

1.5 调用方需要检测对象是否被析构

动态创建的对象是否存活，光看指针或引用都看不出来。如果指针指向的内存对象已经被销毁，根本就不能访问。问题就转换成：如何不访问对象又能知道对象状态呢？

书中举了一个 Observer 模式的例子：代码中，一个 Observable 会通知许多 Observer。这些 Observer 通过 register_ 函数将自身注册到了 Observable 中，在析构时调用 unregister_ 解注册。

Observable 调用 notifyObservers 函数逐个去通知 Observer 时，也会发生和 Counter 例子一样的问题：x 指向的 Observer 对象可能正在析构，调用它的任何非静态成员函数都是不安全的。

class Observer : boost::noncopyable {
 public:
  void observe(Observable *s) {
    s->register_(this);
    subject_ = s;
  }

  ~Observer() {
    subject_->unregister_(this);
  }
  
  virtual void update() = 0;

  Observable *subject_;
};

class Observable {
 public:
  void register_(Observer *x);
  void unregister_(Observer *x);

  void notifyObservers() {
    for (Observer *x : observers_) {
      x->update();
    }
  }

 private:
  std::vector<Observer *> observers_;
};

上述的 race condition 可以通过外部加锁来解决。但在哪加锁？谁维护锁变量？都需要考虑。我们实际上期望有什么存活的对象，通过提供一个 isAlice() 之类的函数，告诉我们该对象是否活着就好了。

1.6 智能指针闪亮登场！

空悬指针

图 1.1 中两个指针 p1 和 p2 都指向堆上同一个对象 Object。线程 A 通过 p1 指针将对象销毁了，此时 p2 就成为了空悬指针。

安全地销毁

引入一层代理对象和引用计数，让这个 Proxy 对象持有一个指向 Object 的指针，同时记录有多少个引用该 Proxy 对象。当 sp1 和 sp2 析构时，Proxy 对象的引用计数会原子减 1。当引用计数为 0 时，就可以安全地销毁 Proxy 对象和 Object 对象了。

其实这就是引用计数型智能指针！

• shared_ptr：控制对象的生命期，最后一个 shared_ptr 析构时，对象即刻销毁。
• weak_ptr：不控制对象的生命期，但能知道对象是否存活，若活着可以提升为有效的 shared_ptr。

1.7 避免各种指针错误

1. 缓冲区溢出：用 std::vector<char> 或 std::string 再或是自己编写的类来管理缓冲区，自动记住缓冲区长度。通过成员函数来读写缓冲区，而不是通过裸指针。
2. 空悬指针/野指针：用 shared_ptr 或 weak_ptr。
3. 重复释放：智能指针。
4. 内存泄露：智能指针。
5. 不配对的 new[] 和 delete：智能指针，创建时可传入自定的 deleter。

1.8 应用智能指针到 Observer 上

现在我们有 weak_ptr，可以通过它来窥探对象的生死，更改后的代码如下。

class Observable {
 public:
  void register_(std::weak_ptr<Observer> x);
  void unregister_(std::weak_ptr<Observer> x);

  void notifyObservers() {
    std::lock_guard<std::mutex> lock(mutex_);
    for (auto it = observers_.begin(); it != observers_.end();) {
      if (auto p = it->lock()) {  // 尝试提升，这一步线程安全
        // 提升成功，引用计数 + 1
        p->update();
        ++it;
      } else {
        // 如果 Observer 已经被销毁，则从列表中移除
        it = observers_.erase(it);
      }
    }
  }

 private:
  mutable std::mutex mutex_;
  std::vector<std::weak_ptr<Observer>> observers_;
};

思考：如果把 std::vector<std::weak_ptr<Observer>> observers_ 改为 shared_ptr ，会有什么后果？

1. 自动清理机制失效：改为 shared_ptr 后，需手动调用 unregister_，否则列表会持续累积无效观察者（即使对象已无外部引用，但因 shared_ptr 存在，无法自动感知失效）。
2. 循环引用风险：若 Observer 持有 Observable 的 shared_ptr，而 Observable 又持有 Observer 的 shared_ptr，会形成循环引用，两者都无法释放。

虽然 weak_ptr 解决了部分 Observer 模式的线程安全，但还存在以下问题：

1. 侵入性：强制要求 Observer 必须以 shared_ptr 来管理。
2. 不完全线程安全：还要求 Observable 本身是用 shared_ptr 管理的，subject_ 大概率是一个 weak_ptr。
3. 锁争用：Observable 三个成员函数都依赖互斥锁 mutex_ 来同步，notifyObservers() 可能会执行很久，导致 register_() 和 unregister() 会阻塞很久。
4. 死锁：若 update() 内部调用了 unregister() 会发生死锁。

1.9 shared_ptr 读写不是线程安全的

shared_ptr 可以保证线程安全的对象释放，但它自己本身读写不是线程安全的，并发读写仍需要加锁。注意！这不是它管理的对象线程安全级别，而是 shared_ptr 对象本身的线程安全级别。

下面举个线程安全读写 shared_ptr 的例子：我们的目的是将 global_ptr 安全地传递给 doit()。

std::mutex mutex;
std::shared_ptr<Foo> global_ptr;

void doit(const std::shared_ptr<Foo> &p);

为了拷贝 global_ptr，我们在读写的时候都需要对其加锁，将其拷贝为一个 local copy，这样临界区之外就不会访问 global_ptr 了。

void read() {
  std::shared_ptr<Fop> local_ptr;
  {
    std::lock_guard<std::mutex> lock(mutex);
    local_ptr = global_ptr;
  }
  doit(local_ptr);
}

void write() {
  std::shared_ptr<Foo> new_ptr = std::make_shared<Foo>();
  {
    std::lock_guard<std::mutex> lock(mutex);
    global_ptr = new_ptr;
  }
  doit(new_ptr);
}

只有这样的 local copy，shared_ptr 作为函数参数传递时不必值传递，只需要 reference to const 传递即可线程安全。这种写法比在临界区内写 global_ptr.reset(new Foo) 更好，可以缩短临界区长度。

思考：在 write() 函数中，global_ptr = new_ptr 这一句有可能会在临界区内销毁原来的 global_ptr 指向的 Foo 对象，我们应该设法将销毁行为移出临界区。

答案：使用临时 shared_ptr 变量，让旧的 global_ptr 生命期延长至 write() 函数中，随 write() 结束而析构。

void write() {
  std::shared_ptr<Foo> new_ptr = std::make_shared<Foo>();
  std::shared_ptr<Foo> old_ptr;  // 用于临时保存旧指针
  
  {
    std::lock_guard<std::mutex> lock(mutex);
    old_ptr = global_ptr;  // 复制旧指针到临时变量
    global_ptr = new_ptr;  // 更新全局指针
  }  // 临界区结束，old_ptr 仍存在引用，未析构
  
  // 旧对象的析构发生在临界区外
  doit(new_ptr);
  // 函数结束后，old_ptr 发生析构
}

1.10 shared_ptr 好与坏

意外延长对象的生命周期：如 1.8 中的思考，shared_ptr 可能会出现循环引用，导致无法自动析构。另一个可能是 std::bind，它总是把实参拷贝一份，若传递的参数是 shared_ptr，那么它的生命期就不会短于 std::funcion 对象。

class Foo {
 public:
  void doit();
};

std::shared_ptr<Foo> p_foo(new Foo);

// p_foo 的生命期不会短于 func 的生命期
std::function<void()> func = std::bind(&Foo::doit, p_foo);

函数参数值拷贝开销大：拷贝时要加锁，且需要修改引用计数，shared_ptr 的拷贝开销比原始指针高。但需要拷贝的时候不多，多数情况下它可以以 const reference 方式传递。

析构动作在创建时被捕获：虚析构函数可能就不是必须的了，而且虚构动作也可以自定义。这不仅可以让其持有任何对象，还可以安全地跨越模块边界，且拥有二进制兼容性。（用简单的话说，析构一定会以期望的形式发生，且能正确析构）

析构所在的线程速度可能会被拖慢：最后一个指向 x 的 shared_ptr 离开其作用域时，x 会在同一个线程析构。如果对象析构比较耗时，最后一个 shared_ptr 引发的析构发生在关键线程，可能会拖慢关键线程的速度。一般做法是将对象的析构转移到一个单独专门做析构的线程。

天然的 RAII：new 和 delete 通过 shared_ptr 纯天然配对。避免循环引用的通常做法是，owner 持有指向 child 的 shared_ptr，child 持有 指向 owner 的 weak_ptr。

1.11 对象池

下面是一个简单的股票对象池：

1. 为了节省系统资源，每一只出现的股票只有一个 Stock 对象；
2. 如果多处用到同一只股票，那么 Stock 对象应该被共享；
3. 如果某一只股票没有在任何地方用到，其对应的 Stock 对象应该被析构，释放资源。

class StockFactory : boost::noncopyable {
 public:
  std::shared_ptr<Stock> get(const std::string &key) {
    std::shared_ptr<Stock> p_stock;
    std::lock_guard<std::mutex> lock(mutex_);
    std::weak_ptr<Stock> &weak_ptr = stocks_[key];
    p_stock = weak_ptr.lock();
    if (!p_stock) {
      p_stock = std::shared_ptr<Stock>(new Stock(key));
      weak_ptr = p_stock;
    }
    return p_stock;
  }

 private:
  mutable std::mutex mutex_;
  std::map<std::string, std::weak_ptr<Stock>> stocks_;
};

现在 stocks_ 的大小只增不减，如果想在 Stock 对象删除时也一同删除 stocks_ 中的 weak_ptr，就需要利用 shared_ptr 的定制析构功能。不过得警惕下面的代码，虽然能正确析构和删除 stocks_ 里的对象，但是必须确保 StockFactory 的生命期大于 Stock 的生命期，否则 Stock 的析构里会调用一个已经析构 StockFactory 的成员函数。

class StockFactory : boost::noncopyable {
 public:
  std::shared_ptr<Stock> get(const std::string &key) {
    std::shared_ptr<Stock> p_stock;
    std::lock_guard<std::mutex> lock(mutex_);
    std::weak_ptr<Stock> &weak_ptr = stocks_[key];
    p_stock = weak_ptr.lock();
    if (!p_stock) {
      // 注意这里！
      p_stock = std::shared_ptr<Stock>(
          new Stock(key),
          std::bind(
            &StockFactory::deleteStock, this, std::placeholders::_1));
      // std::placeholders::_1 表示在调用 std::bind 生成的可调用对象时，
      // 第一个参数会被替换到此处的位置
      weak_ptr = p_stock;
    }
    return p_stock;
  }

  void deleteStock(Stock *stock) {
    if (stock) {
      std::lock_guard<std::mutex> lock(mutex_);
      stocks_.erase(stock->key());
    }
    delete stock;
  }

 private:
  mutable std::mutex mutex_;
  std::map<std::string, std::weak_ptr<Stock>> stocks_;
};

1.12 enable_shared_from_this 获取 this 的智能指针

1.11 中 std::bind 把原始指针 this 保存到了 boost::function 中。如果 StockFactory 的生命期比 Stock 短，那么 Stock 析构时去回调 deleteStock 就会 core dump。如何在 StockFactory 类的成员函数中获得一个指向当前对象的 shared_ptr<StockFactory> 对象呢？

答案是！继承 std::enable_shared_from_this<StockFactory>。

class StockFactory : std::enable_shared_from_this<StockFactory>,
                     boost::noncopyable { /* ... */}

接下来只需要将 std::bind 中的 this 指针改成 shared_from_this()，传入的指针就是智能指针了。

p_stock = std::shared_ptr<Stock>(
          new Stock(key),
          std::bind(
            &StockFactory::deleteStock, 
            shared_from_this(), // shared_ptr
            std::placeholders::_1)
          );

切记！shared_from_this() 也不能在构造函数里调用！对象未构造完全，使用 this 可能会出现未定义行为。

1.13 弱回调防止生命期意外延长

把 shared_ptr 绑到 std::fucntion 里，回调的时候安全了，但也延长了 StockFactory 的生命期。实际上可以将指针从 shared_ptr 转换为 weak_ptr，析构调用时再尝试提升为 shared_ptr。这样可以保证安全调用地对象，也能防止生命期被意外延长。

p_stock = std::shared_ptr<Stock>(
          new Stock(key),
          std::bind(
            &StockFactory::deleteStock, 
            std::weak_ptr<StockFactory>(shared_from_this()), // 看这里！
            std::placeholders::_1)
          );

这下彻底完美了，无论 Stock 和 StockFactory 谁先析构都不会发生 core dump。我们成功地借助了 shared_ptr 和 weak_ptr 完美地解决了两个对象相互引用的问题。

小结

• 不要在构造函数中暴露 this 指针。
• 数据成员的 mutex 不能保护析构。
• 对相同类型的对象加锁，需要根据 mutex 地址顺序加锁，防止死锁。
• 原始指针暴露给多个线程造成 race condition 的概率极大。
• 尽量统一用智能指针管理对象生命期。
• shared_ptr 是值语意，当心意外延长对象的生命期。