深入探讨C++多线程性能优化

深入探讨C++多线程性能优化

在现代软件开发中，多线程编程已成为提升应用程序性能和响应速度的关键技术之一。尤其在C++领域，多线程编程不仅能充分利用多核处理器的优势，还能显著提高计算密集型任务的效率。然而，多线程编程也带来了诸多挑战，特别是在性能优化方面。本文将深入探讨影响C++多线程性能的一些关键因素，比较锁机制与原子操作的性能。通过这些内容，希望能为开发者提供有价值的见解和实用的优化策略，助力于更高效的多线程编程实践。

先在开头给一个例子，你认为下面这段benchmark代码结果会是怎样的。这里的逻辑很简单，将0-20000按线程切成n片，每个线程在一个Set里查找这个数字存不存在，存在则计数+1。

#include <benchmark/benchmark.h>
#include <iostream>
#include <memory>
#include <mutex>
#include <unordered_set>

constexpr int kSetSize = 10000;

class MyBenchmark : public benchmark::Fixture {
 public:
  void SetUp(const ::benchmark::State& state) override {
    std::call_once(flag, [this]() {
      s = std::make_shared<std::unordered_set<int>>();
      for (int i = 0; i < kSetSize; i++) {
        s->insert(i);
      }
    });
  }

  std::shared_ptr<std::unordered_set<int>> GetSet() { return s; }

 private:
  std::shared_ptr<std::unordered_set<int>> s;
  std::once_flag flag;
};

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWork)(benchmark::State& state) {
  for (auto _ : state) {
    int size_sum = kSetSize * 2;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      auto inst = GetSet();
      if (inst->count(i) > 0) {
        sum++;
      }
    }
  }
}

// 注册基准测试，并指定线程数
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(8);

BENCHMARK_MAIN();

跑出来的结果如下：
在这里插入图片描述

将任务切分成多个片段并交由多线程执行后，整体性能不仅没有提升，反而下降了，且性能与线程数成反比。那么，问题来了：导致这种结果的原因是什么？如何才能实现合理的并行执行，从而降低CPU的执行时间？在接下来的部分，笔者将为你揭示答案。

影响多线程性能的因素

笔者认为，影响多线程性能的主要因素有以下两个：

 1.Lock Contention。
 2.Cache Coherency。

Lock Contention对应使用锁来处理多线程同步问题的场景，而Cache Coherency则对应使用原子操作来处理多线程同步问题的场景。

Lock Contention

在多线程环境中，多个线程同时尝试获取同一个锁（Lock）时，会发生竞争现象，这就是所谓的锁竞争（Lock Contention）。锁竞争会导致线程或进程被阻塞，等待锁被释放，从而影响系统的性能和响应时间。大多数情况下，开发人员会选择使用锁来解决线程间的同步问题，因此锁竞争问题也变得广为人知且容易理解。由于锁的存在，位于临界区的代码在同一时刻只能由一个线程执行。因此，优化的思路就是尽量避免多个线程同时访问同一资源。常见的优化方向有两种：

 1.减少临界区大小：临界区越小，这段代码的执行时间就越短，从而在整体程序运行时间中所占的比例也越小，冲突也就越少。
 2.对共享资源进行分桶操作：每个线程只会在某个桶上访问资源，理想情况下，每个线程都会访问不同的桶，这样就不会有冲突。

减少临界区大小需要开发者对自己的代码进行仔细思考，将不必要的操作放在临界区外，例如一些初始化和内存分配操作。

对共享资源进行分桶操作在工程实践中也非常常见。例如，LevelDB的LRUCache中，每个Key只会固定在一个桶上。如果hash函数足够优秀且数据分布足够随机，这种方法可以大大提高LRUCache的性能。

Cache Coherency

缓存一致性（Cache Coherency）是指在多处理器系统中，确保各个处理器的缓存中的数据保持一致的机制。由于现代计算机系统通常包含多个处理器，每个处理器都有自己的缓存（如L1、L2、L3缓存），因此在并发访问共享内存时，可能会出现缓存数据不一致的问题。缓存一致性协议旨在解决这些问题，确保所有处理器在访问共享内存时看到的是一致的数据。

当我们对一个共享变量进行写入操作时，实际上需要通过缓存一致性协议将该变量的更新同步到其他线程的缓存中，否则可能会读到不一致的值。实际上，这个同步过程的单位是一个缓存行（Cache Line），而且同步过程相对较慢，因为涉及到跨核通信。

由此引申出两个严重影响性能的现象：

 1.Cache Ping-Pong。
 2.False Sharing。

Cache Ping-Pong

缓存乒乓效应（Cache Ping-Pong）是指在多处理器系统中，多个处理器频繁地对同一个缓存行（Cache Line）进行读写操作，导致该缓存行在不同处理器的缓存之间频繁地来回传递。这种现象会导致系统性能下降，因为缓存行的频繁传递会引起大量的缓存一致性流量和处理器间通信开销。

讲到这里，其实就可以解释为什么开头那段代码会随着线程数量的增加而性能反而下降。代码中的变量 s 是一个共享资源，但它使用了 shared_ptr。在复制 shared_ptr 时，会引起引用计数的增加（计数+1），多个线程频繁对同一个缓存行进行读写操作，从而引发缓存乒乓效应，导致性能下降。最简单的修改方式就是去掉 shared_ptr，代码如下，同时还可以得到我们预期的结果，即CPU时间随着线程数的增加而降低：

#include <benchmark/benchmark.h>
#include <iostream>
#include <memory>
#include <mutex>
#include <unordered_set>

constexpr int kSetSize = 10000;

class MyBenchmark : public benchmark::Fixture {
 public:
  void SetUp(const ::benchmark::State& state) override {
    std::call_once(flag, [this]() {
      for (int i = 0; i < kSetSize; i++) {
        s.insert(i);
      }
    });
  }

  const std::unordered_set<int>& GetSet() { return s; }

 private:
  std::unordered_set<int> s;
  std::once_flag flag;
};

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWork)(benchmark::State& state) {
  for (auto _ : state) {
    int size_sum = kSetSize * 2;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        benchmark::DoNotOptimize(sum++);
      }
    }
  }
}

// 注册基准测试，并指定线程数
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(8);

BENCHMARK_MAIN();

在这里插入图片描述

False Sharing

伪共享（False Sharing）实际上是一种特殊的缓存乒乓效应（Cache Ping-Pong）。它指的是在多处理器系统中，多个处理器访问不同的数据，但这些数据恰好位于同一个缓存行中，导致该缓存行在不同处理器的缓存之间频繁传递。尽管处理器访问的是不同的数据，但由于它们共享同一个缓存行，仍然会引发缓存一致性流量，导致性能下降。

为了更好地理解这一现象，我们可以对上面的代码进行一些修改。假设我们使用一个 vector<atomic> 来记录不同线程的 sum 值，这样虽然不同线程修改的是不同的sum，但是还是在一个缓存行上。使用 atomic 是为了强制触发缓存一致性协议，否则操作系统可能会进行优化，不会立即将修改反映到主存。

#include <benchmark/benchmark.h>
#include <atomic>
#include <iostream>
#include <memory>
#include <mutex>
#include <unordered_set>

constexpr int kSetSize = 10000;

class MyBenchmark : public benchmark::Fixture {
 public:
  void SetUp(const ::benchmark::State& state) override {
    std::call_once(flag, [this, &state]() {
      for (int i = 0; i < kSetSize; i++) {
        s.insert(i);
      }
      sums = std::vector<std::atomic<int>>(state.threads());
    });
  }

  const std::unordered_set<int>& GetSet() { return s; }
  std::vector<std::atomic<int>>& GetSums() { return sums; }

 private:
  std::unordered_set<int> s;
  std::once_flag flag;
  std::vector<std::atomic<int>> sums;
};

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWork)(benchmark::State& state) {
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        benchmark::DoNotOptimize(GetSums()[state.thread_index()]++);
      }
    }
  }
}

// 注册基准测试，并指定线程数
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(8);

BENCHMARK_MAIN();

在这里插入图片描述

可以看到，尽管不同线程没有使用同一个变量，但由于 sums 里面的元素共享同一个缓存行（Cache Line），同样会导致性能急剧下降。

针对这种情况，只要我们将 sums 中的元素隔离，使它们不在同一个缓存行上，就不会引发这个问题。一般来说，缓存行的大小为64字节，我们可以使用一个类填充到64个字节来实现隔离。优化后的代码如下：

#include <benchmark/benchmark.h>
#include <atomic>
#include <iostream>
#include <memory>
#include <mutex>
#include <unordered_set>

constexpr int kSetSize = 10000;

struct alignas(64) PaddedCounter {
  std::atomic<int> value{0};
  char padding[64 - sizeof(std::atomic<int>)];  // 填充到缓存行大小
};

class MyBenchmark : public benchmark::Fixture {
 public:
  void SetUp(const ::benchmark::State& state) override {
    std::call_once(flag, [this, &state]() {
      for (int i = 0; i < kSetSize; i++) {
        s.insert(i);
      }
      sums = std::vector<PaddedCounter>(state.threads());
    });
  }

  const std::unordered_set<int>& GetSet() { return s; }
  std::vector<PaddedCounter>& GetSums() { return sums; }

 private:
  std::unordered_set<int> s;
  std::once_flag flag;
  std::vector<PaddedCounter> sums;
};

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWork)(benchmark::State& state) {
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        benchmark::DoNotOptimize(GetSums()[state.thread_index()].value++);
      }
    }
  }
}

// 注册基准测试，并指定线程数
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWork)->Threads(8);

BENCHMARK_MAIN();

在这里插入图片描述

Lock VS Atomic

Lock Atomic Benchmark

很多人都认为锁（lock）比原子操作（atomic）要更慢，那么实际上真的是这样吗？下面我们通过两个测试来进行对比。

公平起见，我们将使用一个基于 atomic 变量实现的自旋锁（SpinLock）与 std::mutex 进行性能对比。自旋锁的实现摘自 Folly 库。其原理是使用一个 atomic 变量来标记是否被占用，并使用 acquire-release 内存序来保证临界区的正确性。在冲突过大时，自旋锁会使用 sleep 让出 CPU。代码如下：

#pragma once

#include <atomic>
#include <cstdint>

class Sleeper {
  static const uint32_t kMaxActiveSpin = 4000;
  uint32_t spin_count_;

 public:
  constexpr Sleeper() noexcept : spin_count_(0) {}

  inline __attribute__((always_inline)) static void sleep() noexcept {
    struct timespec ts = {0, 500000};
    nanosleep(&ts, nullptr);
  }

  inline __attribute__((always_inline)) void wait() noexcept {
    if (spin_count_ < kMaxActiveSpin) {
      ++spin_count_;
#ifdef __x86_64__
      asm volatile("pause" ::: "memory");
#elif defined(__aarch64__)
      asm volatile("yield" ::: "memory");
#else
      // Fallback for other architectures
#endif

    } else {
      sleep();
    }
  }
};

class SpinLock {
  enum { FREE = 0, LOCKED = 1 };

 public:
  constexpr SpinLock() : lock_(FREE) {}

  inline __attribute__((always_inline)) bool try_lock() noexcept { return cas(FREE, LOCKED); }

  inline __attribute__((always_inline)) void lock() noexcept {
    Sleeper sleeper;
    while (!try_lock()) {
      do {
        sleeper.wait();
      } while (AtomicCast(&lock_)->load(std::memory_order_relaxed) == LOCKED);
    }
  }

  inline __attribute__((always_inline)) void unlock() noexcept {
    AtomicCast(&lock_)->store(FREE, std::memory_order_release);
  }

 private:
  inline __attribute__((always_inline)) bool cas(uint8_t compare, uint8_t new_val) noexcept {
    return AtomicCast(&lock_)->compare_exchange_strong(compare, new_val, std::memory_order_acquire,
                                                       std::memory_order_relaxed);
  }

  inline __attribute__((always_inline)) static std::atomic<uint8_t>* AtomicCast(uint8_t* value) {
    return reinterpret_cast<std::atomic<uint8_t>*>(value);
  }

 private:
  uint8_t lock_;
};

在第一个benchmark中，我们测试了无竞争情况下的性能。也就是说，原子变量的CAS操作只会执行一次，不会进入 sleep 状态。在这种情况下，自旋锁（SpinLock）等价于一次原子 set 操作。
代码如下：

#include <benchmark/benchmark.h>
#include <atomic>
#include <iostream>
#include <memory>
#include <mutex>
#include <unordered_set>
#include "spin_lock.h"

constexpr int kSetSize = 10000;

// struct alignas(64) PaddedCounter {
//   std::atomic<int> value{0};
//   char padding[64 - sizeof(std::atomic<int>)];  // 填充到缓存行大小
// };

struct alignas(64) PaddedCounterLock {
  int value{0};
  char padding[64 - sizeof(std::atomic<int>)];  // 填充到缓存行大小
};

class MyBenchmark : public benchmark::Fixture {
 public:
  void SetUp(const ::benchmark::State& state) override {
    std::call_once(flag, [this, &state]() {
      for (int i = 0; i < kSetSize; i++) {
        s.insert(i);
      }
      sums_atomic = std::vector<PaddedCounterLock>(state.threads());
      sum_lock = std::vector<PaddedCounterLock>(state.threads());
    });
  }

  const std::unordered_set<int>& GetSet() { return s; }
  std::vector<PaddedCounterLock>& GetSumsAtomic() { return sums_atomic; }
  std::vector<PaddedCounterLock>& GetSumLock() { return sum_lock; }

 private:
  std::unordered_set<int> s;
  std::once_flag flag;
  std::vector<PaddedCounterLock> sums_atomic;
  std::vector<PaddedCounterLock> sum_lock;
};

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkAtomic)(benchmark::State& state) {
  SpinLock m;
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        std::lock_guard<SpinLock> lg(m);
        benchmark::DoNotOptimize(GetSumsAtomic()[state.thread_index()].value++);
      }
    }
  }
}

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkLock)(benchmark::State& state) {
  std::mutex m;
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        std::lock_guard<std::mutex> lg(m);
        benchmark::DoNotOptimize(GetSumLock()[state.thread_index()].value++);
      }
    }
  }
}

// 注册基准测试，并指定线程数
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(8);

BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(8);

BENCHMARK_MAIN();

benchmark结果：
在这里插入图片描述

第二个benchmark是对比竞争激烈时的性能，代码如下：

#include <benchmark/benchmark.h>
#include <atomic>
#include <iostream>
#include <memory>
#include <mutex>
#include <unordered_set>
#include "spin_lock.h"

constexpr int kSetSize = 10000;

class MyBenchmark : public benchmark::Fixture {
 public:
  void SetUp(const ::benchmark::State& state) override {
    std::call_once(flag, [this, &state]() {
      for (int i = 0; i < kSetSize; i++) {
        s.insert(i);
      }
      count_ = 0;
    });
  }

  const std::unordered_set<int>& GetSet() { return s; }
  void SpinLockAndAdd() {
    std::lock_guard<SpinLock> lg(m1_);
    count_++;
  }

  void MutexLockAndAdd() {
    std::lock_guard<std::mutex> lg(m2_);
    count_++;
  }

 private:
  std::unordered_set<int> s;
  std::once_flag flag;
  uint32_t count_;
  SpinLock m1_;
  std::mutex m2_;
};

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkAtomic)(benchmark::State& state) {
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        SpinLockAndAdd();
      }
    }
  }
}

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkLock)(benchmark::State& state) {
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        MutexLockAndAdd();
      }
    }
  }
}

// 注册基准测试，并指定线程数
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(8);

BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(8);

BENCHMARK_MAIN();

benchmark结果：
在这里插入图片描述

可以看到，无论是哪一种情况，std::mutex 的性能都更优。当然，这个测试结果可能会因不同的操作系统而有所不同，但至少可以得出一个结论：这两者的性能是一个量级的，并不存在 atomic 一定比 std::mutex 更快的说法。这其实是因为现代 C++ 中的 std::mutex 实现已经高度优化，其实现与上面的自旋锁（SpinLock）非常相似，在低竞争的情况下并不会陷入内核态。

那么，按上面的说法，是不是我们根本不需要 atomic 变量呢？先来分析一下 atomic 的优点。

atomic 的优点有：

 1.可以实现内存占用极小的锁。
 2.当临界区操作可以等价于一个原子操作时，性能会更高。

对于第二个结论，我们可以做个测试。同样，拿前面的例子稍作修改。

case 1如下:

#include <benchmark/benchmark.h>
#include <atomic>
#include <iostream>
#include <memory>
#include <mutex>
#include <unordered_set>
#include "spin_lock.h"

constexpr int kSetSize = 10000;

class MyBenchmark : public benchmark::Fixture {
 public:
  void SetUp(const ::benchmark::State& state) override {
    std::call_once(flag, [this, &state]() {
      for (int i = 0; i < kSetSize; i++) {
        s.insert(i);
      }
    });
  }

  const std::unordered_set<int>& GetSet() { return s; }

 private:
  std::unordered_set<int> s;
  std::once_flag flag;
};

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkAtomic)(benchmark::State& state) {
  std::atomic<uint32_t> sum = 0;
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        benchmark::DoNotOptimize(sum++);
      }
    }
  }
}

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkLock)(benchmark::State& state) {
  std::mutex m;
  uint32_t sum = 0;
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        std::lock_guard<std::mutex> lg(m);
        sum++;
      }
    }
  }
}

// 注册基准测试，并指定线程数
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(8);

BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(8);

BENCHMARK_MAIN();

benchmark结果：
在这里插入图片描述

case 2如下:

#include <benchmark/benchmark.h>
#include <atomic>
#include <iostream>
#include <memory>
#include <mutex>
#include <unordered_set>
#include "spin_lock.h"

constexpr int kSetSize = 10000;

class MyBenchmark : public benchmark::Fixture {
 public:
  void SetUp(const ::benchmark::State& state) override {
    std::call_once(flag, [this, &state]() {
      for (int i = 0; i < kSetSize; i++) {
        s.insert(i);
      }
      count_ = 0;
      atomic_count_ = 0;
    });
  }

  const std::unordered_set<int>& GetSet() { return s; }
  void AtomicAdd() { atomic_count_++; }
  void MutexLockAndAdd() {
    std::lock_guard<std::mutex> lg(m);
    count_++;
  }

 private:
  std::unordered_set<int> s;
  std::once_flag flag;
  uint32_t count_;
  std::atomic<uint32_t> atomic_count_;
  std::mutex m;
};

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkAtomic)(benchmark::State& state) {
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        AtomicAdd();
      }
    }
  }
}

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkLock)(benchmark::State& state) {
  for (auto _ : state) {
    int size_sum = kSetSize;
    int size_per_thread = (size_sum + state.threads() - 1) / state.threads();
    int sum = 0;
    int start = state.thread_index() * size_per_thread;
    int end = std::min((state.thread_index() + 1) * size_per_thread, size_sum);
    for (int i = start; i < end; i++) {
      const auto& inst = GetSet();
      if (inst.count(i) > 0) {
        MutexLockAndAdd();
      }
    }
  }
}

// 注册基准测试，并指定线程数
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkAtomic)->Threads(8);

BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(1);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(2);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkLock)->Threads(8);

BENCHMARK_MAIN();

benchmark结果：
在这里插入图片描述

接下来结合这两个优点来看，链式数据结构的场景非常适合使用 atomic 变量。

 1.内存占用少：即使每个节点都实现一个自旋锁（SpinLock），也不会浪费太多内存。

 2.链式数据结构的临界区通常可以优化成一个指针的 CAS 操作。

Epoch Based Reclamation

虽然如此，但要写一个高性能的并发安全的链式数据结构是非常困难的，这主要是因为写操作包含了删除操作。举个最简单的例子：

假设有一个链表 A->B->C，一个线程正在读B节点，另一个线程正在删除B节点，如何保证读线程在读B节点期间不会被另一个线程给删掉？

再举个更复杂的例子：

假设有一个链表 A->B->C。一个线程正在读取 B 节点，另一个线程正在修改 B 节点。显然，最简单的实现是锁住 B，同时只允许一个操作，但显然这样从各方面来看性能都不是最佳的，这是第一个方法。

第二个方法是类似于 Copy On Write（COW）。写操作时先重新构造一个节点 B1，再修改对应的数据，最后通过 CAS 操作修改指针连接 A->B1。

我们来分析一下为什么第二个方法远比第一个方法要好。

首先，上锁会触发原子写，意味着即便是你只是为了读数据，也会触发一次 Cache Line 一致性同步的问题。而且在找到 B 节点之前的每一个节点都要依次上锁来保证读取的正确性，这意味着极大概率会发生 Cache Ping-Pong 问题。

再来看写操作，写操作除了上锁以外还需要修改节点的数据。第二个方法需要先构造一个新的节点再修改，意味着这个节点在插入链表之前一定不在其他线程的 Cache 里（排除刚好有某个变量和这个新节点的内存在同一个 Cache Line 的情况）。而第一个方法修改的节点已经在链表里，这表示在之前一定有线程已经访问过这个节点，那么它很可能在 Cache 里面，从而触发一次 Cache Line 一致性同步的问题。

然而事情没有这么简单。试想一下，在修改完指针 A->B1 后，B 节点需要被丢弃释放，这时候其他线程有可能正在访问 B 节点而导致崩溃。

可以看出这些问题都是因为删除操作引起的，这个问题有几个著名的解决方案，比如 Epoch Based Reclamation 和 Hazard Pointer 等。这里只介绍其中的 Epoch Based Reclamation，感兴趣的话请自行搜索了解其他实现方式。

该算法的思路是删除操作会尝试触发版本 +1，但只有当所有线程都是最新版本 e 时才能成功，成功后会回收 e-1 版本的内存。因此，最多会累积 3 个版本未释放节点的内存。是个以空间换时间，轻读重写的方案。

首先，每个线程维护自己的线程变量：

 1.active：标记该线程是否正在读数据。

 2.epoch：标记该线程当前的版本。

全局维护变量：

 1.global_epoch：全局最新的版本。

 2.retire_list：等待释放的节点。

读操作：

 1.首先把线程 active 标记为 true，表示正在读数据。

 2.然后把 global_epoch 赋值给 epoch，记录当前正在读的版本。

 3.如果线程需要删除节点，则把节点放到全局的 retire_list 末尾。

 4.结束读后，将 active 标记为 false。

写操作：

 1.如果要删除节点，则把节点放到全局的 retire_list 末尾，并且尝试增加版本。

 2.增加版本时检查所有线程的状态，当所有线程满足 epoch 等于当前版本 e 或者 active 为 false 时，进行版本 e = e + 1 操作。

 3.清空 e-2 版本的 retire_list。

这里给出一个简单的实现，代码如下：

#pragma once

#include <array>
#include <atomic>
#include <mutex>
#include <numeric>
#include <vector>

constexpr uint8_t kEpochSize = 3;
constexpr uint8_t kCacheLineSize = 64;

template <uint32_t kReadThreadNum>
class ThreadIDManager;

template <uint32_t kReadThreadNum>
struct ThreadID {
  ThreadID() { tid = ThreadIDManager<kReadThreadNum>::GetInstance().AcquireThreadID(); }
  ~ThreadID() { ThreadIDManager<kReadThreadNum>::GetInstance().ReleaseThreadID(tid); }
  uint32_t tid;
};

template <uint32_t kReadThreadNum>
class ThreadIDManager {
 public:
  ThreadIDManager() : tid_list_(kReadThreadNum) { std::iota(tid_list_.begin(), tid_list_.end(), 1); }
  ThreadIDManager(const ThreadIDManager &) = delete;
  ThreadIDManager(ThreadIDManager &&) = delete;
  ThreadIDManager &operator=(const ThreadIDManager &) = delete;
  ~ThreadIDManager() = default;

  static ThreadIDManager &GetInstance() {
    static ThreadIDManager inst;
    return inst;
  }

  uint32_t AcquireThreadID() {
    std::lock_guard<std::mutex> lock(tid_list_mutex_);
    auto tid = tid_list_.back();
    tid_list_.pop_back();
    return tid;
  }

  void ReleaseThreadID(const uint32_t tid) {
    std::lock_guard<std::mutex> lock(tid_list_mutex_);
    tid_list_.emplace_back(tid);
  }

 private:
  std::vector<uint32_t> tid_list_;
  std::mutex tid_list_mutex_;
};

struct TLS {
  TLS() : active(false), epoch(0) {}
  TLS(TLS &) = delete;
  TLS(TLS &&) = delete;
  void operator=(const TLS &) = delete;
  ~TLS() = default;
  std::atomic_flag active;
  std::atomic<uint8_t> epoch;
} __attribute__((aligned(kCacheLineSize)));

template <class RCObject, class DestroyClass, uint32_t kReadThreadNum>
class EbrManager {
 public:
  EbrManager() : tls_list_(), global_epoch_(0), update_(false), write_cnt_(0) {
    for (int i = 0; i < kEpochSize; i++) {
      retire_list_[i].store(nullptr, std::memory_order_release);
    }
  }
  EbrManager(const EbrManager<RCObject, DestroyClass, kReadThreadNum> &) = delete;
  EbrManager(EbrManager<RCObject, DestroyClass, kReadThreadNum> &&) = delete;
  EbrManager &operator=(const EbrManager<RCObject, DestroyClass, kReadThreadNum> &) = delete;
  ~EbrManager() { ClearAllRetireList(); }

  void ClearAllRetireList() {
    for (int i = 0; i < kEpochSize; i++) {
      ClearRetireList(i);
    }
  }

  inline void StartRead() {
    auto &tls = GetTLS();
    tls.active.test_and_set(std::memory_order_release);
    tls.epoch.store(global_epoch_.load(std::memory_order_acquire), std::memory_order_release);
  }

  inline void EndRead() { GetTLS().active.clear(std::memory_order_release); }

  inline void FreeObject(RCObject *object) {
    auto epoch = global_epoch_.load(std::memory_order_acquire);
    auto *node = new RetireNode;
    node->obj = object;
    do {
      node->next = retire_list_[epoch].load(std::memory_order_acquire);
    } while (!retire_list_[epoch].compare_exchange_weak(node->next, node, std::memory_order_acq_rel));

    auto write_cnt = write_cnt_.fetch_add(1, std::memory_order_relaxed);
    if (write_cnt > kReadThreadNum) {
      if (!update_.test_and_set(std::memory_order_acq_rel)) {
        TryGC();
        update_.clear(std::memory_order_release);
      }
    }
  }

 private:
  inline TLS &GetTLS() {
    thread_local ThreadID<kReadThreadNum> thread_id;
    return tls_list_[thread_id.tid];
  }

  inline void TryGC() {
    auto epoch = global_epoch_.load(std::memory_order_acquire);
    // TODO 优化记录上一次搜索到的位置
    for (int i = 0; i < tls_list_.size(); i++) {
      if (tls_list_[i].active.test(std::memory_order::memory_order_acquire) &&
          tls_list_[i].epoch.load(std::memory_order::memory_order_acquire) != epoch) {
        return;
      }
    }
    global_epoch_.store((epoch + 1) % kEpochSize, std::memory_order_release);
    ClearRetireList((epoch + 2) % kEpochSize);
    write_cnt_.store(0, std::memory_order_relaxed);
  }

  inline void ClearRetireList(int index) {
    auto *retire_node = retire_list_[index].load(std::memory_order_acquire);
    while (retire_node != nullptr) {
      DestroyClass destroy(retire_node->obj);
      auto *old_node = retire_node;
      retire_node = retire_node->next;
      delete old_node;
    }
    retire_list_[index].store(nullptr, std::memory_order_release);
  }

  struct RetireNode {
    RCObject *obj;
    RetireNode *next;
  };
  std::array<char, kCacheLineSize> start_padding_;
  std::array<TLS, kReadThreadNum> tls_list_;
  std::atomic<uint8_t> global_epoch_;
  std::array<char, kCacheLineSize> mid_padding_;
  std::atomic_flag update_;
  std::atomic<uint32_t> write_cnt_;
  std::atomic<RetireNode *> retire_list_[kEpochSize];
  std::array<char, kCacheLineSize> end_padding_;
};

这里再给出一个benchmark，对比一下使用 Epoch Based Reclamation（EBR）和不使用 EBR 的区别。由于笔者时间有限，只能写一个非常简单的版本，仅供参考。

#include <benchmark/benchmark.h>
#include <mutex>
#include "ebr.h"
#include "spin_lock.h"

struct Node {
  Node() : lock(), next(nullptr) {}
  int key;
  int value;
  Node *next;
  SpinLock lock;
};

class NodeFree {
 public:
  NodeFree(Node *node) { delete node; }
};

/*
 * 快速测试起见，简单写了个list版本的kv结构，里面只会有3个元素，然后只支持Get和Modify，Modify也必定会命中key。
 * 不是直接把key，value，next设置成atomic变量而是使用SpinLock的原因是模拟复杂情况，真实情况下会存在Add和Remove操作，实现没有如此简单。
 */
class MyList {
 public:
  MyList() {
    Node *pre_node = nullptr;
    auto *&cur_node = root_;
    // 这里虽然插入了10个元素，但后面的实现会假设第一个key 9作为header是绝对不会被修改或者读到的。
    for (int i = 0; i < 10; i++) {
      cur_node = new Node;
      cur_node->key = i;
      cur_node->value = i;
      cur_node->next = pre_node;
      pre_node = cur_node;
    }
  }

  int Get(int key, int *value) {
    root_->lock.lock();
    auto *cur_node = root_->next;
    auto *pre_node = root_;
    while (cur_node != nullptr) {
      cur_node->lock.lock();
      pre_node->lock.unlock();
      if (key == cur_node->key) {
        *value = cur_node->value;
        cur_node->lock.unlock();
        return 0;
      }
      pre_node = cur_node;
      cur_node = cur_node->next;
    }
    pre_node->lock.unlock();
    return 1;
  }
  int Modify(int key, int value) {
    root_->lock.lock();
    auto *cur_node = root_->next;
    auto *pre_node = root_;
    while (cur_node != nullptr) {
      cur_node->lock.lock();
      pre_node->lock.unlock();
      if (key == cur_node->key) {
        cur_node->value = value;
        cur_node->lock.unlock();
        return 0;
      }
      pre_node = cur_node;
      cur_node = cur_node->next;
    }
    pre_node->lock.unlock();
    return 1;
  }

  int GetUseEbr(int key, int *value) {
    ebr_mgr_.StartRead();
    auto *cur_node = root_->next;
    while (cur_node != nullptr) {
      if (key == cur_node->key) {
        *value = cur_node->value;
        ebr_mgr_.EndRead();
        return 0;
      }
      cur_node = cur_node->next;
    }
    ebr_mgr_.EndRead();
    return 1;
  }

  int ModifyUseEbr(int key, int value) {
    root_->lock.lock();
    auto *cur_node = root_->next;
    auto *pre_node = root_;
    while (cur_node != nullptr) {
      cur_node->lock.lock();
      if (key == cur_node->key) {
        auto *new_node = new Node;
        new_node->key = cur_node->key;
        new_node->value = value;
        new_node->next = cur_node->next;
        pre_node->next = new_node;
        cur_node->lock.unlock();
        pre_node->lock.unlock();
        ebr_mgr_.FreeObject(cur_node);
        return 0;
      }
      auto *next_node = cur_node->next;
      pre_node->lock.unlock();
      pre_node = cur_node;
      cur_node = next_node;
    }
    pre_node->lock.unlock();
    return 1;
  }

 private:
  Node *root_;
  EbrManager<Node, NodeFree, 15> ebr_mgr_;
};

class MyBenchmark : public benchmark::Fixture {
 public:
  void SetUp(const ::benchmark::State &state) override {}

  MyList &GetMyList() { return l; }

 private:
  MyList l;
  std::once_flag flag;
};

constexpr int kKeySize = 10000;

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkNoUseEbr)(benchmark::State &state) {
  for (auto _ : state) {
    auto &mylist = GetMyList();
    if (0 == state.thread_index()) {
      // modify
      for (int i = 0; i < kKeySize; i++) {
        mylist.Modify(i % 9, i);
      }
    } else {
      // get
      for (int i = 0; i < kKeySize; i++) {
        int value;
        mylist.Get(i % 9, &value);
      }
    }
  }
}

BENCHMARK_DEFINE_F(MyBenchmark, MultiThreadedWorkUseEbr)(benchmark::State &state) {
  for (auto _ : state) {
    auto &mylist = GetMyList();
    if (0 == state.thread_index()) {
      // modify
      for (int i = 0; i < kKeySize; i++) {
        mylist.ModifyUseEbr(i % 9, i);
      }
    } else {
      // get
      for (int i = 0; i < kKeySize; i++) {
        int value;
        mylist.GetUseEbr(i % 9, &value);
      }
    }
  }
}

// 注册基准测试，并指定线程数
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkNoUseEbr)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkNoUseEbr)->Threads(8);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkNoUseEbr)->Threads(12);

BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkUseEbr)->Threads(4);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkUseEbr)->Threads(8);
BENCHMARK_REGISTER_F(MyBenchmark, MultiThreadedWorkUseEbr)->Threads(12);

BENCHMARK_MAIN();

benchmark结果：
在这里插入图片描述

结语

以上是笔者的一些个人见解。由于水平和时间有限，测试用例尽可能地简化，可能不够全面。如果内容中有任何错误，欢迎指出。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/895694.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

Redis应用高频面试题

Redis应用高频面试题

Redis 作为一个高性能的分布式缓存系统，广泛应用于后端开发中，因此在后端研发面试中，关于 Redis 的问题十分常见。 本文整理了30个常见的 Redis 面试题目，涵盖了 Redis 的源码、数据结构、原理、集群模式等方面的知识，并附上简要的回答，帮助大家更好地准备相关的面试。 …
阅读更多...
【Windows】【DevOps】Windows Server 2022 采用WinSW将一个控制台应用程序作为服务启动（方便）

【Windows】【DevOps】Windows Server 2022 采用WinSW将一个控制台应用程序作为服务启动（方便）

下载WinSW 项目地址&#xff1a; GitHub - winsw/winsw: A wrapper executable that can run any executable as a Windows service, in a permissive license. 下载地址&#xff1a; https://github.com/winsw/winsw/releases/download/v2.12.0/WinSW-x64.exe 参考配置模…
阅读更多...
深度学习 之 模型部署 使用Flask和PyTorch构建图像分类Web服务

深度学习 之 模型部署 使用Flask和PyTorch构建图像分类Web服务

引言 随着深度学习的发展&#xff0c;图像分类已成为一项基础的技术&#xff0c;被广泛应用于各种场景之中。本文将介绍如何使用Flask框架和PyTorch库来构建一个简单的图像分类Web服务。通过这个服务&#xff0c;用户可以通过HTTP POST请求上传花朵图片&#xff0c;然后由后端…
阅读更多...
Nginx(Linux)：服务器版本升级和新增模块

Nginx(Linux)：服务器版本升级和新增模块

目录 1、概述2、使用Nginx服务信号完成Nginx升级2.1 备份当前版本的Nginx2.2 向服务器导入新的Nginx2.3 向服务器导入新的Nginx2.4 停止老版本Nginx 3、使用Nginx安装目录的make命令完成升级3.1 备份当前版本的Nginx3.2 向服务器导入新的Nginx3.3 执行更新命令 1、概述 如果想…
阅读更多...
E41.【C语言】练习:斐波那契函数的空间复杂度的计算及函数调用分析

E41.【C语言】练习:斐波那契函数的空间复杂度的计算及函数调用分析

目录 1.题目 2.解 Fib嵌套函数调用细则的分析 调用堆栈分析 之后的具体内容见视频 附:一张核心图 附:一张堆栈图 注意 1.题目 求下列代码的时间复杂度 long long f(size_t n) {if(n < 3)return 1;return f(n-1) f(n-2); } 2.解 显然是递归算法(递归讲解见35.【…
阅读更多...
推荐一款多显示器管理工具：DisplayMagician

推荐一款多显示器管理工具：DisplayMagician

DisplayMagician是一款开源工具&#xff0c;专为Windows用户设计&#xff0c;能够通过一个快捷方式轻松自动配置屏幕和声音。它特别适合游戏玩家和应用程序用户&#xff0c;可以实现屏幕配置、声音设备切换以及启动额外程序等功能&#xff0c;最后在游戏或应用程序关闭时&#…
阅读更多...
实现vlan间的通信

实现vlan间的通信

方法一&#xff1a;单臂路由 概述 单臂路由是一种网络配置&#xff0c;它允许在路由器的一个物理接口上通过配置多个子接口来处理不同VLAN的流量&#xff0c;从而实现VLAN间的通信。 原理 路由器重新封装MAC地址&#xff0c;转换Vlan标签 基础模型 1、配置交换机的链…
阅读更多...
Vxe UI vue vxe-table grid 如何滚动、定位到指定行或列

Vxe UI vue vxe-table grid 如何滚动、定位到指定行或列

Vxe UI vue vxe-table vxe-grid 在表格中有时候需要对数据会列进行操作。可以会定位到某一行或某一列&#xff0c;vxe-table 中提供了丰富的函数式 API&#xff0c;可以轻松对行与列进行各种的灵活的操作。 定位到指定行与列 通过调用 scrollColumn(columnOrField) 方法&…
阅读更多...
阿里云云盘在卸载时关联到PHP进程，如何在不影响PHP进程情况下卸载磁盘

阿里云云盘在卸载时关联到PHP进程，如何在不影响PHP进程情况下卸载磁盘

1.问题&#xff1a; 在使用umount /dev/vdc1 卸载磁盘时&#xff0c;提示如下&#xff0c;导致无法在Linux系统下卸载磁盘 umount /dev/vdc1 umount: /var/www/html/*/eshop/IFile3: target is busy.(In some cases useful info about processes that usethe device is found…
阅读更多...
WPF -- LiveCharts的使用和源码

WPF -- LiveCharts的使用和源码

LiveCharts 是一个开源的 .NET 图表库&#xff0c;特别适用于 WPF、WinForms 和其他 .NET 平台。它提供了丰富的图表类型和功能&#xff0c;使开发者能够轻松地在应用程序中创建动态和交互式图表。下面我将使用WPF平台创建一个测试实例。 一、LiveCharts的安装和使用 1.安装N…
阅读更多...
网盘直链下载神器NDM

网盘直链下载神器NDM

工具介绍 ​Neat Download Manager分享一款网盘不限速神器,安装步骤稍微有一点繁琐,但实际体验下载速度飞快,个人实际体验还是非常不错的 NDM是一款免费且强大的下载工具。可以帮助你下载各种文件&#xff0c;还能够在多任务下载中保持出色的速度及其稳定性 通过网盘分享的文…
阅读更多...
五年三次冲刺IPO失败，企业业绩成长性恐不足，三年分红约1.5亿元

五年三次冲刺IPO失败，企业业绩成长性恐不足，三年分红约1.5亿元

中超股份终止原因如下&#xff1a;首先&#xff0c;报告期&#xff0c;中超股份营收和净利润增幅出现下降趋势&#xff0c;公司业绩规模成长性恐不足。其次&#xff0c;公司货币资金较为紧张情况下&#xff0c;仍在报告期内连续三年分红&#xff0c;累计1.46亿元&#xff0c;募…
阅读更多...
Java爬虫：获取直播带货数据的实战指南

Java爬虫：获取直播带货数据的实战指南

在当今数字化时代&#xff0c;直播带货已成为电商领域的新热点&#xff0c;通过直播平台展示商品并进行销售&#xff0c;有效促进了产品的曝光和销售量的提升。然而&#xff0c;如何在直播带货过程中进行数据分析和评估效果&#xff0c;成为了摆在商家面前的一个重要问题。本文…
阅读更多...
边缘计算技术的优势与挑战

边缘计算技术的优势与挑战

如今&#xff0c;随着5G快速无线网络的到来&#xff0c;将计算存储和物联网&#xff08;IoT&#xff09;分析的部署放在靠近数据产生的地方&#xff0c;使得边缘计算成为可能。 物联网设备和新应用的扩展需要实时计算能力。5G无线正在考虑边缘系统&#xff0c;以快速跟踪支持实…
阅读更多...
016集——c# 实现CAD类库 与窗体的交互（CAD—C#二次开发入门）

016集——c# 实现CAD类库 与窗体的交互（CAD—C#二次开发入门）

第一步&#xff1a;搭建CAD类库dll开发环境。 第二步&#xff1a;添加窗体 第三步&#xff1a;添加控件 第四步&#xff1a;双击控件&#xff0c;在控件点击方法内输入代码 第五步&#xff1a;在主程序内实例化新建的form类&#xff0c;并弹窗form窗体 第六步&#xff1a;CAD命…
阅读更多...
1.2.3 TCP IP模型

1.2.3 TCP IP模型

TCP/IP模型&#xff08;接网叔用&#xff09; 网络接口层 网络层 传输层 应用层 理念&#xff1a;如果某些应用需要“数据格式转换”“会话管理功能”&#xff0c;就交给应用层的特定协议去实现 tip&#xff1a;数据 局部正确不等于全局正确 但是&#xff0c;数据的 全局正…
阅读更多...
Codeforces Round 770 (Div. 2)

Codeforces Round 770 (Div. 2)

比赛链接&#xff1a;Dashboard - Codeforces Round 770 (Div. 2) - Codeforces A. Reverse and Concatenate 题意&#xff1a; 思路&#xff1a; 假设 s "abba" 经过1次操作后 -> "abbaabba" s "abcd" 经过一次操作后 -> "abcd…
阅读更多...
EditPlus的安装软件包

EditPlus的安装软件包

解压并粘贴到C:\Program Files (x86)中 点击激活密匙,并一直同意 确认并选择默认的位置: 关闭并重新激活密匙 就好了 无需添加快捷方式: 只需要选择任意文件 并选择该应用打开一次即可 通过百度网盘分享的文件&#xff1a;EditPlus_5.0.611.zip 链接&#xff1a;https://pa…
阅读更多...
Sentinel 快速入门

Sentinel 快速入门

前置推荐阅读:Sentinel 介绍-CSDN博客 前置推荐阅读&#xff1a;Nacos快速入门-CSDN博客 快速开始 欢迎来到 Sentinel 的世界&#xff01;这篇新手指南将指引您快速入门 Sentinel。 Sentinel 的使用可以分为两个部分: 核心库&#xff08;Java 客户端&#xff09;&#xff1a…
阅读更多...
C语言之练习题

C语言之练习题

欢迎来到我的&#xff1a;世界 希望作者的文章对你有所帮助&#xff0c;有不足的地方还请指正&#xff0c;大家一起学习交流 !&#x1f60a; 目录 内容第一题&#xff1a;加一第二题&#xff1a;移动零第三题 &#xff1a;分发饼干第四题&#xff1a;买股票的最佳时机第五题&a…
阅读更多...
最新文章

Copyright @ 2022~2023