openharmony5.0.0中C++公共基础类测试-线程相关(一)
C++公共基础类测试及源码剖析
延续传统，show me the code，除了给出应用示例还重点分析了下openharmony中的实现。
简介
openharmony中提供了C++公共基础类库，为标准系统提供了一些常用的C++开发工具类，本文分析其实现，并给出使用示例,库主要涉及内容如下：
- 文件、路径、字符串相关操作的能力增强接口
- 读写锁、信号量、定时器、线程增强及线程池等接口
- 安全数据容器、数据序列化等接口
- 各子系统的错误码相关定义
环境
系统:openharmony5.0.0
部署:参照源码中的说明文档(quickstart-pkg-3568-helloworld.md)
本节重点说明关于线程相关的内容。
源码目录
commonlibrary/c_utils
├─ base
│ ├── include # 对各子系统开放的接口头文件
│ ├── src # 源文件
│ └── test # 测试代码
├─ Docs├── en # 英文文档└── zh-cn # 中文文档
线程相关
概述
包含强化线程能力、线程池、线程安全Map、线程安全栈与队列、线程安全阻塞队列
- 强化线程能力:提供例如启动线程、同步通知、异步通知等功能的接口
- 线程池:提供线程安全的线程池功能。线程安全是对于线程池本身而非池内线程而言的。 维护一个任务队列，一个线程组。使用者向任务队列中注册需要进行的任务，线程组执行任务队列中的任务。
- 线程安全Map：提供了一个线程安全的map实现。SafeMap在STL map基础上封装互斥锁，以确保对map的操作安全。
- 线程安全栈与队列:线程安全队列，是在dequeue的基础上封装std::lock_guard，以此实现线程的相关操作。根据继承SafeQueueInner抽象类，并对dequeue的pop方法的重写，可以实现SafeStack和SafeQueue的相关方法。
- 线程安全阻塞队列：线程安全阻塞队列SafeBlockQueue类，提供阻塞和非阻塞版的入队入队和出队接口，并提供可最追踪任务完成状态的的SafeBlockQueueTracking类。
强化线程能力
接口说明
OHOS::Thread
| 返回值类型 | 名称 |
|---|---|
| Thread() 构造函数, 构造一个Thread对象，但并不会启动线程。 | |
| virtual ~Thread() 析构函数 | |
| ThreadStatus | Start(const std::string& name, int32_t priority = THREAD_PROI_NORMAL, size_t stack = 0); 创建并启动一个子线程，循环执行Run()，当Run()返回false或通知退出时停止。 |
| ThreadStatus | NotifyExitSync() 同步通知线程退出,即阻塞式停止子线程。 当前线程被阻塞，等待子线程结束。 |
| void | virtual NotifyExitAsync() 异步通知线程退出，即子线程退出与否不阻塞当前线程。 通知子线程停止，当前线程继续运行。 |
| bool | virtual ReadyToWork() 判断线程是否已经准备就绪，始终返回true。 |
| bool | IsExitPending() const 获取线程退出待定标志位。 |
| bool | IsRunning() const 判断线程是否在运行 |
| pthread_t | GetThread() const 获取线程ID |
类图关系
源码剖析
从接口和类图中我们可以看到强化线程能力这一节，主要是提供例如启动线程、同步通知、异步通知等功能的接口，此库主要是通过thread类进行接口的封装，客户需要实现一个run的接口函数，此函数由库内部调用。异步通知是使用c++的一个线程同步的工具类
condition_variable来实现的，其他的接口由对应的标志位来实现的(不再详细说)。
对condition_variable不熟悉的码友可以看下这篇介绍
启动函数调用流程如下:
创建并启动子线程的函数(start)，主要通过linux系统函数pthread_create创建线程，将入口函数设置为ThreadParam类中的代理函数(Proxy)，在Proxy函数中设置线程名、优先级等，并通过循环函数(ThreadStart)循环调用用户端的run函数，并实时检测更新线程状态。具体流程如下：
ThreadStatus Thread::Start(const std::string& name, int32_t priority, size_t stack)|-->status_ = ThreadStatus::OK //一些变量的初始化|-->ThreadParam para; //定义线程参数|-->para.startRoutine = ThreadStart;//初始化参数的值，由线程启动时调用|-->para.args = this;//将本线程(thread)传到参数中，作为ThreadStart的传参|-->para.name = name;//线程名称|-->para.priority = priority//线程优先级|-->bool res = CreatePThread(para, stack, &thread_)//创建线程|-->auto t = new ThreadParam;//新创建线程参数对象|-->t->startRoutine = para.startRoutine//将相关参数赋值到t对象中|-->para.args = t//将参数对象t作为ThreadParam::Proxy传参|-->para.startRoutine = reinterpret_cast<ThreadFunc>(&ThreadParam::Proxy)//强制进行类型转换|-->int result = pthread_create(&thread, &attr, reinterpret_cast<PThreadRoutine>(para.startRoutine), para.args)//核心函数将ThreadParam::Proxy作为线程的入口函数
线程创建后会触发代理函数(Proxy)如下:
static int Proxy(const ThreadParam* t)|-->(void)setpriority(PRIO_PROCESS, 0, prio);//系统函数，设置当前用户的优先级|-->prctl(PR_SET_NAME, threadName.substr(0, MAX_THREAD_NAME_LEN).c_str(), 0, 0, 0)//系统函数，设置进程名|-->t->startRoutine(t->args)//回调ThreadStart(由start函数设置)
t->startRoutine为start函数中设置的循环函数(ThreadStart),执行过程如下:
int Thread::ThreadStart(void* args)|-->循环执行以下操作|-->result = self->Run()//回调客户端重写的run函数|-->std::unique_lock<std::mutex> lk(self->lock_);|-->if ((!result) || self->exitPending_)//当退出标志为true时|-->self->cvThreadExited_.notify_all();//唤醒所有等待的线程|-->break;//线程退出循环
在同步通知线程退出时，采用了线程间同步的一种高级工具(std::condition_variable)，允许线程在某些条件不满足时挂起，直到其他线程通知它们条件已经满足。
ThreadStatus Thread::NotifyExitSync()|-->std::unique_lock<std::mutex> lk(lock_);//C++标准库中的一种 RAll风格的互斥锁管理器，，它可以方便地管理互斥锁的锁定和解锁操作。在进入代码块时自动锁定 self->lock_，在离开代码块时(无论是通寸正常执行结束还是通过异常退出)，自动解锁 self->lock从而确保互斥锁的正确使用和避免死锁情况的发生|-->exitPending_ = true;//退出标志|-->while (running_) {cvThreadExited_.wait(lk);}//会释放互斥锁，然后在等待期间重新获取它,|-->exitPending_ = false;//退出标志|-->return status_;
由ThreadStart函数中已确认线程退出标志(exitPending_)后通过cvThreadExited_.notify_all()唤醒，此函数再进行返回。
应用示例
本案例完成如下工作：
- 主线程每1秒打印子进程的相关信息。主线程在第5秒时，关闭子线程运行。
- 创建1个子线程，每隔1秒打印当前运行次数。
// 时间标记量
static const int FORMAX = 5;// 自定义类，继承OHOS::Thread，Run()重新写
class ThreadSample : public OHOS::Thread {
public:ThreadSample() : OHOS::Thread::Thread(){}~ThreadSample(){}
protected:bool Run() override;
};// 程序运行代码，每隔1秒打印相关信息
bool ThreadSample::Run(){static int current = 0;current++;cout << "Run(): current = " << current << endl;sleep(1);return true;
}int main(int argc, char **argv){ThreadSample thread;// 启动线程thread.Start("thread sample", OHOS::THREAD_PROI_NORMAL, 0);// 打印线程相关信息for (int i = 0; i < (2 * FORMAX); i++) {cout << "main: i = " << i << endl;cout << " ThreadId = " << thread.GetThread() << endl;cout << " ReadyToWork = " << thread.ReadyToWork() << endl;cout << " IsExitPending = " << thread.IsExitPending() << endl;cout << " IsRunning = " << thread.IsRunning() << endl;if (i == (1 * FORMAX)) {// 异步停止线程，不用等待，直接返回cout << "main: NotifyExitAsync" << endl;thread.NotifyExitAsync();}sleep(1);}thread.NotifyExitSync();// 等待退出return 0;
}
- 运行结果
# ./utils_thread
main: i = 0ThreadId = 4152769824ReadyToWork = 1IsExitPending = 0
Run(): current = 1 IsRunning = 1Run(): current = main: i = 21ThreadId = 4152769824ReadyToWork = 1IsExitPending = 0IsRunning = 1
Run(): current = 3
main: i = 2ThreadId = 4152769824ReadyToWork = 1IsExitPending = 0IsRunning = 1
Run(): current = 4
main: i = 3ThreadId = 4152769824ReadyToWork = 1IsExitPending = 0IsRunning = 1
Run(): current = 5
main: i = 4ThreadId = 4152769824ReadyToWork = 1IsExitPending = 0IsRunning = 1
Run(): current = 6
main: i = 5ThreadId = 4152769824ReadyToWork = 1IsExitPending = 0IsRunning = 1
main: NotifyExitAsync
main: i = 6ThreadId = 4294967295ReadyToWork = 1IsExitPending = 1IsRunning = 0
main: i = 7ThreadId = 4294967295ReadyToWork = 1IsExitPending = 1IsRunning = 0
main: i = 8ThreadId = 4294967295ReadyToWork = 1IsExitPending = 1IsRunning = 0
main: i = 9ThreadId = 4294967295ReadyToWork = 1IsExitPending = 1IsRunning = 0
注意在第32行Run(): current = 6之后,便不再打印current，说明此线程已经退出，所以此后打印的信息即为线程退出后的状态。通过多次启动此进程(utils_thread)，可见ThreadId在IsRunning为true的状态时呈现的是动态的，在为false时是固定的，也进一步说明threadid退出后固定为INVALID_PTHREAD_T。
线程池
接口说明
- OHOS::ThreadPool
| Name | |
|---|---|
| ThreadPool(const std::string& name = std::string()) 构造ThreadPool。为线程池内线程命名。 | |
| ~ThreadPool() override | |
| void | AddTask(const Task& f) 向任务队列中添加一个Task。若未调用Start()则直接执行Task且不会向任务队列添加该Task. |
| size_t | GetCurTaskNum() 获取当前任务数。 |
| size_t | GetMaxTaskNum() const 获取最大任务数。 |
| std::string | GetName() const 获取线程池命名。 |
| size_t | GetThreadsNum() const 获取线程池内线程数。 |
| void | SetMaxTaskNum(size_t maxSize) 设置任务队列中最大任务数。 |
| uint32_t | Start(int threadsNum) 启动给定数量threadsNum的线程，执行任务队列中的任务。 |
| void | Stop() 停止线程池，等待所有线程结束。 |
类关系图
ThreadPool
源码剖析
从以上类图可知线程池类(ThreadPool)继承了NoCopyable类，禁止拷贝操作，通过成员变量tasks维护一个任务队列，通过成员变量(threads)维护一个线程组。使用者向任务队列中注册需要进行的任务，线程组执行任务队列中的任务实现对任务的管理。以上主要操作涉及接口主要为Start(线程组执行任务队列中的任务)和AddTask(向任务队列中注册需要进行的任务)。
禁止拷贝操作的实现:
讨论禁止拷贝操作的实现首先要了解c++11引入的一种语法，用于显式地禁用某个函数，包括构造函数、析构函数、拷贝构造函数、赋值运算符等。通过将函数声明为 =delete，可以显式地阻止该函数被调用，即使在类内部或者友元函数中也无法使用。
class NoCopyable {
protected: //允许子类继承，但防止直接实例化!NoCopyable() {}//构造函数virtual ~NoCopyable() {}//析构函数
private://设置为private可以确保NoCopyable类的复制构造函数和移动赋值操作符完全不可访问，包括在其子类中。这样可以更好地隐藏类的实现细节，防止意外的复制和移动操作。如果将DISALLOW_COPY_AND_MOVE设置为protected，子类可能会尝试调用这些被删除的函数，导致编译错误。将它们设置为private可以确保编译器阻止任何对这些函数的调用，包括在子类中。DISALLOW_COPY_AND_MOVE(NoCopyable);//DISALLOW_COPY_AND_MOVE为定义的宏，可展开为如下:
DISALLOW_COPY_AND_MOVE(NoCopyable)展开后如下:
NoCopyable(const NoCopyable&) = delete;NoCopyable& operator= (const NoCopyable&) = delete//=delete;用于禁用复制构造函数，这意味着任何尝试复制该对象的代码都会导致编译错误
NoCopyable(NoCopyable&&) = delete;NoCopyable& operator= (NoCopyable&&) = delete //&& 表示右值引用，意味着该构造函数可以接受一个临时对象(即即将被销毁的对象)作为参数//=delete;用于禁用移动构造函数。任何尝试通过移动构造函数来复制该对象的代码都会导致编译错误
线程组执行任务队列中的任务
线程组的执行主要通过start函数执行，传入的参数为启动的线程数量，主要执行过程如下：
uint32_t ThreadPool::Start(int numThreads)
|-->threads_.reserve(numThreads)//预留对应启动线程的空间
|-->for (int i = 0; i < numThreads; ++i) {
|-->std::thread t([this] { this->WorkInThread(); });//这个函数会一直循环，直到线程池停止|-->while (running_) {//只要是启动状态便一直循环|-->Task task = ScheduleTask();//任务调度，会将最新的任务取出来|--> while (tasks_.empty() && running_) {hasTaskToDo_.wait(lock);}//如果任务队列为空，且在运行状态时，便开启等待(hasTaskToDo_:线程间同步的一种高级工具(std::condition_variable))|-->task = tasks_.front();//取出最新的任务|-->if (maxTaskNum_ > 0)acceptNewTask_.notify_one();//通知给新任务(添加新任务时当超过最大任务数时便会wait)|-->task();//执行取出来的任务
|-->int err = pthread_setname_np(t.native_handle(), (myName_ + std::to_string(i)).c_str());//设置线程名称
|-->threads_.push_back(std::move(t));//将新创建的线程对象t移动到threads容器中。使用std::move可以避免不必要的复制操作，提高性能。
向任务队列中注册需要进行的任务
通过AddTask接口向任务队列(tasks_)中注册任务,主要执行过程如下：
void ThreadPool::AddTask(const Task &f)
|-->if (threads_.empty())f();//如果线程组为空则直接执行任务函数
|--> while (Overloaded()) acceptNewTask_.wait(lock)//如果过载时则一直等待|-->return (maxTaskNum_ > 0) && (tasks_.size() >= maxTaskNum_)//判读是否过载
|-->tasks_.push_back(f)//将任务函数放入任务队列中
|-->hasTaskToDo_.notify_one()//通知线程组有新的任务来了
其中hasTaskToDo_和acceptNewTask_皆为std::condition_variable类型，此类型的详细说明可参考这篇文章
应用示例
本案例完成如下工作：
- 创建1个线程池，设置该线程池内部有1024个线程空间。
- 启动5个线程。每个线程每秒打印1段字符串，10秒后停止。
// 线程执行函数
void func(const std::string &name)
{for (int i = 0; i < 10; i++) {cout << "func: " << name << " and i = " << i << endl;sleep(1);}
}int main(int argc, char **argv)
{OHOS::ThreadPool thread_poll("thread_poll_name");int max_task_num = 1024;int start_task_num = 5;string str_name;cout << "get max task num(default): " << thread_poll.GetMaxTaskNum() << endl;// 查看默认的线程池最大任务数cout << "set max task num: " << max_task_num << endl;thread_poll.SetMaxTaskNum(max_task_num);// 设置线程池的最大任务数cout << "get max task num(set): " << thread_poll.GetMaxTaskNum() << endl;// 再查看线程池最大任务数cout << "start thread: " << start_task_num << endl; // 开启启动线程thread_poll.Start(start_task_num);for (int i = 0; i < start_task_num; i++) {cout << "add task: i = " << i << endl;str_name = "thread_pool_" + to_string(i);auto task = std::bind(func, str_name);用于绑定函数和参数，不用立即执行// 添加任务到线程池中，并启动运行thread_poll.AddTask(task);sleep(1);}cout << "stop thread: start" << endl; // 等待关闭所有的线程，会等待线程池程序全部结束才返回thread_poll.Stop();cout << "stop thread: end" << endl;return 0;
}
- 执行过程
由于是多个线程同时执行所以会导致串口打印的时候反应不过来，有一些字段会稍微乱一些。
# ./utils_thread_poll
get max task num(default): 0
set max task num: 1024
get max task num(set): 1024
start thread: 5
add task: i = 0
func: thread_pool_0 and i = 0
add task: i = 1
func: thread_pool_0 and i = 1
func: thread_pool_1 and i = 0
func: thread_pool_0 and i = add task: i = 22func: thread_pool_2 and i = 0
func: thread_pool_1 and i = 1
add task: i = func: thread_pool_0 and i = 33func: thread_pool_1 and i = 2
func: thread_pool_2 and i = 1
func: thread_pool_3 and i = 0
func: thread_pool_0 and i = 4
add task: i = 4
func: thread_pool_2 and i = 2
func: thread_pool_3 and i = 1
func: thread_pool_4 and i = func: 0
thread_pool_1 and i = 3
func: thread_pool_0 and i = func: thread_pool_2 and i = 5
3
func: thread_pool_1 and i = 4
func: thread_pool_3 and i = 2
func: thread_pool_4 and i = 1
stop thread: start
func: thread_pool_0 and i = 6
func: thread_pool_1 and i = 5
func: thread_pool_3 and i = 3
func: thread_pool_4 and i = 2
func: thread_pool_2 and i = 4
func: thread_pool_0 and i = 7
func: thread_pool_4 and i = 3
func: thread_pool_3 and i = func: thread_pool_2func: thread_pool_1 and i = 6and i = 5
4
func: thread_pool_0 and i = 8
func: thread_pool_2 and i = func: 6thread_pool_1 and i =
func: thread_pool_3 and i = 5
7
func: thread_pool_4 and i = 4
func: thread_pool_0 and i = 9
func: thread_pool_3 and i = func: 6
func: thread_pool_4 and i = func: thread_pool_2 and i = 57thread_pool_1 and i = 8
func: thread_pool_3 and i = 7
func: thread_pool_4 and i = 6func:
thread_pool_2 and i = 8
func: thread_pool_1 and i = 9
func: thread_pool_3 and i = 8
func: thread_pool_4 and i = 7
func: thread_pool_2 and i = 9
func: thread_pool_3 and i = 9
func: thread_pool_4 and i = 8
func: thread_pool_4 and i = 9
stop thread: end
- 疑问答疑
问题一:WorkInThread函数中使用while(running_)死循环，发现只有当调用stop时，才会将running_置为false，当不调用stop时便会一直死循环吧？这样明显不合理吧？
答疑:这个循环不会导致死循环，因为它依赖于 ScheduleTask() 函数来获取任务，并且只有在有任务的情况下才会执行 task()。如果 ScheduleTask() 返回一个空的任务（即 task 为 false），线程会在下一次循环中再次尝试获取任务，而不会无限执行同一个任务。在某些情况下确实可能导致线程一直循环而不做任何工作，根据代码片段，ScheduleTask() 函数内部调用了 acceptNewTask_.notify_one();，这表明线程池使用了条件变量来管理任务的调度，从而避免了不必要的CPU消耗。
线程安全Map
接口说明
| 返回类型 | 名称 |
|---|---|
| SafeMap() | |
| SafeMap(const SafeMap& rhs) | |
| ~SafeMap() | |
| void | Clear() 删除map中存储的所有键值对。 |
| void | EnsureInsert(const K& key, const V& value) 在map中插入元素。 |
| void | Erase(const K& key) 删除map中键为key的键值对。 |
| bool | Find(const K& key, V& value) 在map中查找元素。 |
| bool | FindOldAndSetNew(const K& key, V& oldValue, const V& newValue) 在map中查找元素并将key对应的oldValue替换为newValue。 |
| bool | Insert(const K& key, const V& value) 在map中插入新元素。 |
| bool | IsEmpty() 判断map是否为空。 |
| void | Iterate(const SafeMapCallBack& callback) 遍历map中的元素。 |
| SafeMap& | operator=(const SafeMap& rhs) |
| V | ReadVal(const K& key) 线程安全地读map内元素 |
| void | ChangeValueByLambda(const K& key, LambdaCallback callback) 线程安全地操作safemap内元素，操作行为需要自定义 |
| int | Size() 获取map的size大小。 |
类关系图
源码剖析
源码是在safe_map.h文件中直接实现的。
主要是对std::map<K, V> map_的二次封装，和c++标准map的接口类似，就不再详细说明了。只看个例子
bool FindOldAndSetNew(const K& key, V& oldValue, const V& newValue)//查找并替换键值对
|-->auto iter = map_.find(key);//查找键为key的键值对
|-->if (iter != map_.end()) {//判断是否找到了键值对
|-->oldValue = iter->second;//将旧值赋给oldValue
|-->map_.erase(iter);//删除原有键值对
|-->map_.insert(std::pair<K, V>(key, newValue));//插入新的键值对
应用示例
本案例主要完成如下工作：
- 创建1个子线程，负责每秒调用EnsureInsert()插入元素；
- 创建1个子线程，负责每秒调用Insert()插入元素；
- 创建1个子线程，负责每秒调用Erase()删除元素；
- 创建1个子线程，负责每秒调用FindOldAndSetNew()替换元素的值；
- 主线程等待上述线程结束，Iterate()和Find()查看所有元素；
- 主线程等待上述线程结束，清空SafeMap，并调用IsEmpty()查看是否确实是空。
通过创建线程池，设置最大任务数，并启动线程,通过AddTask的方式添加不同的任务，例如EnsureInsert、Insert等。
OHOS::ThreadPool threads("name_rwlock_threads")
threads.SetMaxTaskNum(128);
threads.Start(4);
str_name = "Thread_xxx";//Thread_EnsureInsert
auto task_ensure_insert = std::bind(<任务函数名称>, str_name);
threads.AddTask(task_ensure_insert);//添加任务
测试在map中插入元素(EnsureInsert)
static struct MapInfo m_map1_insert[] = {{ 1, "aaa" },{ 2, "bbb" },{ 3, "ccc" },{ 4, "ddd" },{ 5, "eee" },{ 6, "fff" },{ 7, "ggg" },{ 8, "hhh" },{ 9, "iii" },{ 10, "jjj" }
};
// 使用EnsureInsert()插入元素
void map_ensure_insert(const string& name)
{int key = 0;string value = "";for (int i = 0; i < (sizeof(m_map1_insert) / sizeof(struct MapInfo)); i++) {key = m_map1_insert[i].key;value = m_map1_insert[i].str;m_safemap.EnsureInsert(key, value);cout << name << ": insert successful and key = " << key << " and value = " << value << endl;sleep(1);}
}
测试在map中插入元素(Insert)
static struct MapInfo m_map2_insert[] = {{ 101, "111" },{ 102, "222" },{ 103, "333" },{ 104, "444" },{ 105, "555" },{ 106, "666" },{ 107, "777" },{ 108, "888" },{ 109, "999" },{ 110, "000" }
};
void map_insert(const string& name)
{int key = 0;string value = "";for (int i = 0; i < (sizeof(m_map2_insert) / sizeof(struct MapInfo)); i++) {key = m_map2_insert[i].key;value = m_map2_insert[i].str;if (m_safemap.Insert(key, value) == false) {cout << name << ": insert failed and key = " << to_string(key) << " and value = " << value << endl;} else {cout << name << ": insert successful and key = " << to_string(key) << " and value = " << value << endl;}sleep(1);}
}
测试在map中使用erase删除元素
void map_erase(const string& name)
{int key = 0;string value = "";for (int i = 0; i < (sizeof(m_map2_insert) / sizeof(struct MapInfo)); i++) {key = m_map2_insert[i].key;m_safemap.Erase(key);cout << name << ": Erase successful and key = " << to_string(key) << endl;sleep(1);}
}
测试使用FindOldAndSetNew替换元素的值
void map_findold_and_setnew(const string& name)
{int key = 0;string old_value = "";string new_value = "";for (int i = 0; i < (sizeof(m_map1_insert) / sizeof(struct MapInfo)); i++) {key = m_map1_reset[i].key;old_value = "";new_value = m_map1_reset[i].str;if (m_safemap.FindOldAndSetNew(key, old_value, new_value) == false) {cout << name << ": FindOldAndSetNew failed and key = " << to_string(key) << " and old_value = " << old_value << endl;} else {cout << name << ": FindOldAndSetNew successful and key = " << to_string(key)<< " and old_value = " << old_value << " and new_value = " << new_value << endl;}sleep(1);}
}
最后设置结束，并等待结束，然后打印SafeMap所有元素，最后清空并确认是否清空完成。
// 设置结束，并等待结束threads.Stop();cout << "Threads Stop" << endl;// 打印SafeMap所有元素cout << "SafeMap Iterate: " << endl;m_safemap.Iterate(map_iterate_print);cout << "SafeMap Find: " << endl;map_find_print();// 清空SafeMapcout << "SafeMap Clear" << endl;m_safemap.Clear();// 查看是否是空的cout << "SafeMap IsEmpty: " << m_safemap.IsEmpty() << endl;
- 执行结果
# ./utils_thread_map
Thread_EnsureInsert: insert successful and key = 1Thread_Insert and value = aaa: insert successful and key =
101 and value = 111
Thread_Erase: Erase successful and key = 101
Thread_FindOldAndSetNew: FindOldAndSetNew successful and key = 1 and old_value = aaa and new_value = abc
Thread_EnsureInsert: insert successful and key = 2 and value = bbb
Thread_Erase: Erase successful and key = 102
Thread_FindOldAndSetNew: FindOldAndSetNew successful and key = 2 and old_value = bbb and new_value = bcd
Thread_Insert: insert successful and key = 102 and value = 222
Thread_Erase: Erase successful and key = 103
Thread_EnsureInsert: insert successful and key = Thread_InsertThread_FindOldAndSetNew: FindOldAndSetNew failed and key = 3 and old_value =
3 and value = ccc
: insert successful and key = 103 and value = 333
Thread_Erase: Erase successful and key = 104
Thread_FindOldAndSetNew: FindOldAndSetNew successful and key = 4 and old_value = ddd and new_value = def
Thread_EnsureInsert: insert successful and key = Thread_Insert: insert successful and key = 104 and value = 444
4 and value = ddd
Thread_Erase: Erase successful and key = 105
Thread_FindOldAndSetNew: FindOldAndSetNew failed and key = 5 and old_value =
Thread_Insert: insert successful and key = 105 and value = 555
Thread_EnsureInsert: insert successful and key = 5 and value = eee
Thread_Erase: Erase successful and key = 106
Thread_Insert: insert successful and key = Thread_FindOldAndSetNew: FindOldAndSetNew failed and key = 1066 and value = 666 and old_value =
Thread_EnsureInsert: insert successful and key = 6 and value = fffThread_Erase: Erase successful and key = 107
Thread_Insert: insert successful and key = 107 and value = 777
Thread_EnsureInsert: insert successful and key = Thread_FindOldAndSetNew: FindOldAndSetNew successful and key = 7 and old_value = ggg and new_value = ghi
7 and value = ggg
Thread_Erase: Erase successful and key = 108
Thread_EnsureInsert: insert successful and key = 8 and value = hhh
Thread_Insert: insert successful and key = 108Thread_FindOldAndSetNew and value = : FindOldAndSetNew failed and key = 8 and old_value =
888
Thread_Erase: Erase successful and key = Thread_FindOldAndSetNew109: FindOldAndSetNew failed and key = 9 and old_value = Thread_Insert: insert successful and key = 109 and value = 999
Thread_EnsureInsert: insert successful and key = 9 and value = iii
Thread_Erase: Erase successful and key = 110
Thread_Insert: insert successful and key = 110 and value = 000
Thread_FindOldAndSetNew: FindOldAndSetNew failed and key = 10 and old_value =
Thread_EnsureInsert: insert successful and key = 10 and value = jjj
Threads Stop
SafeMap Iterate:
key = 1, value = abc
key = 2, value = bcd
key = 3, value = ccc
key = 4, value = def
key = 5, value = eee
key = 6, value = fff
key = 7, value = ghi
key = 8, value = hhh
key = 9, value = iii
key = 10, value = jjj
key = 103, value = 333
key = 104, value = 444
key = 105, value = 555
key = 106, value = 666
key = 107, value = 777
key = 108, value = 888
key = 109, value = 999
key = 110, value = 000
SafeMap Find:
key = 1, value = abc
key = 2, value = bcd
key = 3, value = ccc
key = 4, value = def
key = 5, value = eee
key = 6, value = fff
key = 7, value = ghi
key = 8, value = hhh
key = 9, value = iii
key = 10, value = jjj
key = 103, value = 333
key = 104, value = 444
key = 105, value = 555
key = 106, value = 666
key = 107, value = 777
key = 108, value = 888
key = 109, value = 999
key = 110, value = 000
SafeMap Clear
SafeMap IsEmpty: 1
线程安全阻塞队列
这些类模板使用 C++ 标准库中的互斥锁（std::mutex）、条件变量（std::condition_variable）和队列（std::queue）来实现线程安全的操作。
接口说明
#include <safe_block_queue.h>
OHOS::SafeBlockQueue
| 返回值 | 名称 |
|---|---|
SafeBlockQueue(int capacity) 构造函数，整数参数 capacity，用于设置队列的最大容量 | |
| virtual ~SafeBlockQueue() 析构函数 | |
| void | virtual Push(T const& elem) 入队操作(阻塞版)，将一个元素插入队列的末尾，如果队列已满，则使用条件变量阻塞当前线程，直到队列中有空位。插入完成后，通知等待队列不为空的线程。 |
| bool | virtual PushNoWait(T const& elem) 入队操作(非阻塞版)，如果队列已满，直接返回 false；否则插入元素并返回 true。 |
| T | Pop() 出队操作(阻塞版)，从队列的开头移除一个元素，如果队列为空，则使用条件变量阻塞当前线程，直到队列中有元素。移除完成后，通知等待队列不满的线程。 |
| bool | PopNotWait(T& outtask) 出队操作(非阻塞版)，如果队列为空，直接返回 false；否则移除元素并返回 true |
| unsigned int | Size() 获取队列容量 |
| bool | IsEmpty() 队列判空 |
| bool | IsFull() 队列判满 |
OHOS::SafeBlockQueueTracking，该类继承自 SafeBlockQueue。SafeBlockQueueTracking 旨在提供线程安全的阻塞队列功能，并在此基础上增加对任务完成情况的跟踪能力。
class SafeBlockQueueTracking : public SafeBlockQueue
| 返回值 | 名称 |
|---|---|
| explicit | SafeBlockQueueTracking(int capacity) 构造函数，接受一个 int 类型的参数 capacity，用于设置队列的最大容量，并初始化 |
| virtual ~SafeBlockQueueTracking() 析构函数 | |
| void | virtual Push(T const& elem) 入队操作(阻塞版)，首先增加 unfinishedTaskCount_ 计数。如果队列已满，则使用条件变量 cvNotFull_ 阻塞当前线程，直到队列有空位。插入完成后，通知等待队列不为空的线程。 |
| bool | virtual PushNoWait(T const& elem) 入队操作(非阻塞版)先检查队列是否已满。如果已满，则直接返回 false；否则插入元素，并增加 unfinishedTaskCount_ 计数，通知等待队列不为空的线程，返回 true |
| bool | OneTaskDone() 一个任务完成时的响应函数，它会减少 unfinishedTaskCount_ 计数。如果计数减少到 0，则通知所有等待任务完成的线程 |
| void | Join() 等待未完成队列，会阻塞当前线程，直到队列中的所有任务都已完成（即 unfinishedTaskCount_ 计数变为 0） |
| int | GetUnfinishTaskNum() 获取未完成任务数 |
类关系图
源码剖析
源码是在safe_block_queue.h文件中直接实现的。
主要是对std::queue queueT_的二次封装，和c++标准map的接口类似，就不再详细说明了，直接看应用示例吧。
应用示例
（1）使用SafeBlockQueue接口的案例
- 判断命令行是否使用阻塞，还是非阻塞；
- 创建子线程生产者，使用阻塞/非阻塞方式，入队操作；
- 创建子线程消费者，使用阻塞/非阻塞方式，出队操作；
- 主线程等待所有子线程结束
// 定义常量
const char STRING_WAIT[] = "wait";
const char STRING_NOWAIT[] = "nowait";// 定义常量
const int SIZE = 5;
// 定义SafeBlockQueue变量
OHOS::SafeBlockQueue<int> m_safeBlockQueue(SIZE);// 返回时间字符串
static string get_curtime()
{string str = "";time_t time_now = time(nullptr);struct tm tm_now;localtime_r(&time_now, &tm_now);str += to_string(tm_now.tm_year + 1900) + "-" + to_string(tm_now.tm_mon + 1) + "-" + to_string(tm_now.tm_mday);str += " ";str += to_string(tm_now.tm_hour) + ":" + to_string(tm_now.tm_min) + ":" + to_string(tm_now.tm_sec);return str;
}static void product_wait(const string &name)
{for (int i = 0; i < (2 * SIZE); i++) {cout << get_curtime() << ", " << __func__ << ": Push Start, i = " << i << endl;// 使用阻塞方式的SafeBlockQueuem_safeBlockQueue.Push(i);cout << get_curtime() << ", " << __func__ << ": Push Success, i = " << i << endl;// 等待1秒cout << get_curtime() << ", " << __func__ << ": Sleep 1 sec " << endl;std::this_thread::sleep_for(std::chrono::milliseconds(1000));}
}static void consume_wait(const string &name)
{for (int i = 0; i < (2 * SIZE); i++) {cout << get_curtime() << ", " << __func__ << ": Pop Start, i = " << i << endl;// 使用阻塞方式的SafeBlockQueueint value = m_safeBlockQueue.Pop();cout << get_curtime() << ", " << __func__ << ": Pop Success, i = " << i << ", value = " << value << endl;// 等待0.5秒cout << get_curtime() << ", " << __func__ << ": Sleep 0.5 sec " << endl;std::this_thread::sleep_for(std::chrono::milliseconds(500));}
}static void product_nowait(const string &name)
{bool ret;for (int i = 0; i < (2 * SIZE); i++) {cout << get_curtime() << ", " << __func__ << ": Push Start, i = " << i << endl;// 使用非阻塞方式的SafeBlockQueueret = m_safeBlockQueue.PushNoWait(i);cout << get_curtime() << ", " << __func__ << ": Push ret = " << ret << ", i = " << i << endl;// 等待1秒cout << get_curtime() << ", " << __func__ << ": Sleep 1 sec " << endl;std::this_thread::sleep_for(std::chrono::milliseconds(1000));}
}static void consume_nowait(const string &name)
{for (int i = 0; i < (SIZE * 2); i++) {// 等待有新数据int value = 0;// 使用非阻塞方式的SafeBlockQueuebool ret = m_safeBlockQueue.PopNotWait(value);cout << get_curtime() << ", " << __func__ << ": PopNotWait ret = " << ret << ", value = " << value << endl;// 等待500毫秒std::this_thread::sleep_for(std::chrono::milliseconds(500));}
}int main(int argc, char **argv)
{bool enable_wait = true;OHOS::ThreadPool threads("threads");string str_name = "";// 获取命令行参数if (argc != 2) {cout << "Usage: " << argv[0] << " " << STRING_WAIT << "/" << STRING_NOWAIT << endl;return -1;}if (strncmp(argv[1], STRING_WAIT, sizeof(STRING_WAIT)) == 0) {enable_wait = true;} else if (strncmp(argv[1], STRING_NOWAIT, sizeof(STRING_NOWAIT)) == 0) {enable_wait = false;} else {cout << "Usage: " << argv[0] << " " << STRING_WAIT << "/" << STRING_NOWAIT << endl;return -1;}threads.SetMaxTaskNum(4);threads.Start(2);// 创建生产者线程cout << get_curtime() << ", " << __func__ << ": task_product start" << endl;auto task_product = (enable_wait) ? (std::bind(product_wait, str_name)) : (std::bind(product_nowait, str_name));threads.AddTask(task_product);// 等待SIZE秒，将SafeBlockQueue容器填满cout << get_curtime() << ", " << __func__ << ": sleep " << SIZE << " sec" << endl;std::this_thread::sleep_for(std::chrono::milliseconds(1000 * SIZE));// 创建消费者线程cout << get_curtime() << ", " << __func__ << ": consume start" << endl; auto task_consumer = (enable_wait) ? (std::bind(consume_wait, str_name)) : (std::bind(consume_nowait, str_name));threads.AddTask(task_consumer);threads.Stop();cout << get_curtime() << ", " << __func__ << ": Queue Wait End" << endl; return 0;
}
- 执行结果
# ./utils_thread_queue wait
2017-8-5 17:2:16, main: task_product start
2017-8-5 17:2:16, main: sleep 5 sec
2017-8-5 17:2:16, product_wait: Push Start, i = 0
2017-8-5 17:2:16, product_wait: Push Success, i = 0
2017-8-5 17:2:16, product_wait: Sleep 1 sec
2017-8-5 17:2:17, product_wait: Push Start, i = 1
2017-8-5 17:2:17, product_wait: Push Success, i = 1
2017-8-5 17:2:17, product_wait: Sleep 1 sec
2017-8-5 17:2:18, product_wait: Push Start, i = 2
2017-8-5 17:2:18, product_wait: Push Success, i = 2
2017-8-5 17:2:18, product_wait: Sleep 1 sec
2017-8-5 17:2:19, product_wait: Push Start, i = 3
2017-8-5 17:2:19, product_wait: Push Success, i = 3
2017-8-5 17:2:19, product_wait: Sleep 1 sec
2017-8-5 17:2:20, product_wait: Push Start, i = 4
2017-8-5 17:2:20, product_wait: Push Success, i = 4
2017-8-5 17:2:20, product_wait: Sleep 1 sec
2017-8-5 17:2:21, main: consume start
2017-8-5 17:2:21, consume_wait: Pop Start, i = 0
2017-8-5 17:2:21, consume_wait: Pop Success, i = 0, value = 0
2017-8-5 17:2:21, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:21, product_wait: Push Start, i = 5
2017-8-5 17:2:21, product_wait: Push Success, i = 5
2017-8-5 17:2:21, product_wait: Sleep 1 sec
2017-8-5 17:2:21, consume_wait: Pop Start, i = 1
2017-8-5 17:2:21, consume_wait: Pop Success, i = 1, value = 1
2017-8-5 17:2:21, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:22, consume_wait: Pop Start, i = 2
2017-8-5 17:2:22, consume_wait: Pop Success, i = 2, value = 2
2017-8-5 17:2:22, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:22, product_wait: Push Start, i = 6
2017-8-5 17:2:22, product_wait: Push Success, i = 6
2017-8-5 17:2:22, product_wait: Sleep 1 sec
2017-8-5 17:2:22, consume_wait: Pop Start, i = 3
2017-8-5 17:2:22, consume_wait: Pop Success, i = 3, value = 3
2017-8-5 17:2:22, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:23, consume_wait: Pop Start, i = 4
2017-8-5 17:2:23, consume_wait: Pop Success, i = 4, value = 4
2017-8-5 17:2:23, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:23, product_wait: Push Start, i = 7
2017-8-5 17:2:23, product_wait: Push Success, i = 7
2017-8-5 17:2:23, product_wait: Sleep 1 sec
2017-8-5 17:2:23, consume_wait: Pop Start, i = 5
2017-8-5 17:2:23, consume_wait: Pop Success, i = 5, value = 5
2017-8-5 17:2:23, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:24, product_wait: Push Start, i = 8
2017-8-5 17:2:24, product_wait: Push Success, i = 8
2017-8-5 17:2:24, product_wait: Sleep 1 sec
2017-8-5 17:2:24, consume_wait: Pop Start, i = 6
2017-8-5 17:2:24, consume_wait: Pop Success, i = 6, value = 6
2017-8-5 17:2:24, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:24, consume_wait: Pop Start, i = 7
2017-8-5 17:2:24, consume_wait: Pop Success, i = 7, value = 7
2017-8-5 17:2:24, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:25, product_wait: Push Start, i = 9
2017-8-5 17:2:25, product_wait: Push Success, i = 9
2017-8-5 17:2:25, product_wait: Sleep 1 sec
2017-8-5 17:2:25, consume_wait: Pop Start, i = 8
2017-8-5 17:2:25, consume_wait: Pop Success, i = 8, value = 8
2017-8-5 17:2:25, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:25, consume_wait: Pop Start, i = 9
2017-8-5 17:2:25, consume_wait: Pop Success, i = 9, value = 9
2017-8-5 17:2:25, consume_wait: Sleep 0.5 sec
2017-8-5 17:2:26, main: Queue Wait End
（2）使用SafeBlockQueueTracking接口的案例
- 判断命令行是否使用阻塞，还是非阻塞；
- 创建子线程生产者，使用阻塞/非阻塞方式，入队操作；
- 创建子线程消费者，使用阻塞/非阻塞方式，出队操作；
- 主线程等待所有子线程结束
// 定义常量
const char STRING_WAIT[] = "wait";
const char STRING_NOWAIT[] = "nowait";// 定义常量
const int SIZE = 5;
// 定义SafeBlockQueue变量
OHOS::SafeBlockQueueTracking<int> m_safeBlockQueueTracking(SIZE);// 返回时间字符串
static string get_curtime()
{string str = "";time_t time_now = time(nullptr);struct tm tm_now;localtime_r(&time_now, &tm_now);str += to_string(tm_now.tm_year + 1900) + "-" + to_string(tm_now.tm_mon + 1) + "-" + to_string(tm_now.tm_mday);str += " ";str += to_string(tm_now.tm_hour) + ":" + to_string(tm_now.tm_min) + ":" + to_string(tm_now.tm_sec);return str;
}static void product_wait(const string &name)
{for (int i = 0; i < (2 * SIZE); i++) {cout << get_curtime() << ", " << __func__ << ": Push Start, i = " << i << endl;// 使用阻塞方式的SafeBlockQueueTrackingm_safeBlockQueueTracking.Push(i);cout << get_curtime() << ", " << __func__ << ": Push Success, i = " << i << endl;// 等待1秒cout << get_curtime() << ", " << __func__ << ": Sleep 1 sec " << endl;std::this_thread::sleep_for(std::chrono::milliseconds(1000));}
}static void consume_wait(const string &name)
{for (int i = 0; i < (2 * SIZE); i++) {cout << get_curtime() << ", " << __func__ << ": Pop Start, i = " << i << endl;// 使用阻塞方式的SafeBlockQueueTrackingint value = m_safeBlockQueueTracking.Pop();cout << get_curtime() << ", " << __func__ << ": Pop Success, i = " << i << ", value = " << value << endl;m_safeBlockQueueTracking.OneTaskDone();cout << get_curtime() << ", " << __func__ << ": Push OneTaskDone successful" << endl;// 等待0.5秒cout << get_curtime() << ", " << __func__ << ": Sleep 0.5 sec " << endl;std::this_thread::sleep_for(std::chrono::milliseconds(500));}
}static void product_nowait(const string &name)
{bool ret;for (int i = 0; i < (2 * SIZE); i++) {cout << get_curtime() << ", " << __func__ << ": Push Start, i = " << i << endl;// 使用非阻塞方式的SafeBlockQueueTrackingret = m_safeBlockQueueTracking.PushNoWait(i);cout << get_curtime() << ", " << __func__ << ": Push ret = " << ret << ", i = " << i << endl;if (ret == true) {m_safeBlockQueueTracking.OneTaskDone();cout << get_curtime() << ", " << __func__ << ": Push OneTaskDone successful" << endl;}// 等待1秒cout << get_curtime() << ", " << __func__ << ": Sleep 1 sec " << endl;std::this_thread::sleep_for(std::chrono::milliseconds(1000));}
}static void consume_nowait(const string &name)
{for (int i = 0; i < (SIZE * 2); i++) {// 等待有新数据int value = 0;// 使用非阻塞方式的SafeBlockQueueTrackingbool ret = m_safeBlockQueueTracking.PopNotWait(value);cout << get_curtime() << ", " << __func__ << ": PopNotWait ret = " << ret << ", value = " << value << endl;// 等待500毫秒std::this_thread::sleep_for(std::chrono::milliseconds(500));}
}int main(int argc, char **argv)
{bool enable_wait = true;OHOS::ThreadPool threads("threads");string str_name = "";// 获取命令行参数if (argc != 2) {cout << "Usage: " << argv[0] << " " << STRING_WAIT << "/" << STRING_NOWAIT << endl;return -1;}if (strncmp(argv[1], STRING_WAIT, sizeof(STRING_WAIT)) == 0) {enable_wait = true;} else if (strncmp(argv[1], STRING_NOWAIT, sizeof(STRING_NOWAIT)) == 0) {enable_wait = false;} else {cout << "Usage: " << argv[0] << " " << STRING_WAIT << "/" << STRING_NOWAIT << endl;return -1;}threads.SetMaxTaskNum(4);threads.Start(2);// 创建生产者线程cout << get_curtime() << ", " << __func__ << ": task_product start" << endl;auto task_product = (enable_wait) ? (std::bind(product_wait, str_name)) : (std::bind(product_nowait, str_name));threads.AddTask(task_product);// 等待SIZE秒，将SafeBlockQueue容器填满cout << get_curtime() << ", " << __func__ << ": sleep " << SIZE << " sec" << endl;std::this_thread::sleep_for(std::chrono::milliseconds(1000 * SIZE));// 创建消费者线程cout << get_curtime() << ", " << __func__ << ": consume start" << endl; auto task_consumer = (enable_wait) ? (std::bind(consume_wait, str_name)) : (std::bind(consume_nowait, str_name));threads.AddTask(task_consumer);threads.Stop();cout << get_curtime() << ", " << __func__ << ": Queue Wait End" << endl; return 0;
}
- 执行结果
# ./utils_thread_queue_track nowait
2017-8-5 17:37:0, main: task_product start
2017-8-5 17:37:0, main: sleep 2017-8-5 17:37:0, product_nowait: Push Start, i = 5 sec
0
2017-8-5 17:37:0, product_nowait: Push ret = 1, i = 0
2017-8-5 17:37:0, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:0, product_nowait: Sleep 1 sec
2017-8-5 17:37:1, product_nowait: Push Start, i = 1
2017-8-5 17:37:1, product_nowait: Push ret = 1, i = 1
2017-8-5 17:37:1, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:1, product_nowait: Sleep 1 sec
2017-8-5 17:37:2, product_nowait: Push Start, i = 2
2017-8-5 17:37:2, product_nowait: Push ret = 1, i = 2
2017-8-5 17:37:2, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:2, product_nowait: Sleep 1 sec
2017-8-5 17:37:3, product_nowait: Push Start, i = 3
2017-8-5 17:37:3, product_nowait: Push ret = 1, i = 3
2017-8-5 17:37:3, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:3, product_nowait: Sleep 1 sec
2017-8-5 17:37:4, product_nowait: Push Start, i = 4
2017-8-5 17:37:4, product_nowait: Push ret = 1, i = 4
2017-8-5 17:37:4, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:4, product_nowait: Sleep 1 sec
2017-8-5 17:37:5, main: consume start
2017-8-5 17:37:5, consume_nowait: PopNotWait ret = 1, value = 0
2017-8-5 17:37:5, product_nowait: Push Start, i = 5
2017-8-5 17:37:5, product_nowait: Push ret = 1, i = 5
2017-8-5 17:37:5, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:5, product_nowait: Sleep 1 sec
2017-8-5 17:37:5, consume_nowait: PopNotWait ret = 1, value = 1
2017-8-5 17:37:6, consume_nowait: PopNotWait ret = 1, value = 2
2017-8-5 17:37:6, product_nowait: Push Start, i = 6
2017-8-5 17:37:6, product_nowait: Push ret = 1, i = 6
2017-8-5 17:37:6, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:6, product_nowait: Sleep 1 sec
2017-8-5 17:37:6, consume_nowait: PopNotWait ret = 1, value = 3
2017-8-5 17:37:7, consume_nowait: PopNotWait ret = 1, value = 4
2017-8-5 17:37:7, product_nowait: Push Start, i = 7
2017-8-5 17:37:7, product_nowait: Push ret = 1, i = 7
2017-8-5 17:37:7, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:7, product_nowait: Sleep 1 sec
2017-8-5 17:37:7, consume_nowait: PopNotWait ret = 1, value = 5
2017-8-5 17:37:8, consume_nowait: PopNotWait ret = 1, value = 6
2017-8-5 17:37:8, product_nowait: Push Start, i = 8
2017-8-5 17:37:8, product_nowait: Push ret = 1, i = 8
2017-8-5 17:37:8, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:8, product_nowait: Sleep 1 sec
2017-8-5 17:37:8, consume_nowait: PopNotWait ret = 1, value = 7
2017-8-5 17:37:9, product_nowait: Push Start, i = 2017-8-5 17:37:9, consume_nowait: PopNotWait ret = 19, value = 82017-8-5 17:37:9, product_nowait: Push ret = 1, i = 9
2017-8-5 17:37:9, product_nowait: Push OneTaskDone successful
2017-8-5 17:37:9, product_nowait: Sleep 1 sec
2017-8-5 17:37:9, consume_nowait: PopNotWait ret = 1, value = 9
2017-8-5 17:37:10, main: Queue Wait End
线程安全栈与队列
线程安全队列，是在dequeue的基础上封装std::lock_guard，以此实现线程的相关操作。根据继承SafeQueueInner抽象类，并对dequeue的pop方法的重写，可以实现SafeStack和SafeQueue的相关方法。
对std::lock_guard不熟悉的码友可以看这篇文章
接口说明
- OHOS::SafeQueueInner
| 返回值 | 名称 |
|---|---|
| SafeQueueInner() 构造函数 | |
| virtual ~SafeQueueInner() 析构函数 | |
| void | Erase(T& object) 移除某个元素 |
| bool | Empty() 队列判空 |
| void | Clear() 清空队列元素 |
| int | Size() 获取队列的容量 |
| void | Push(const T& pt) 入队操作 |
| void | virtual void DoPush(const T& pt) = 0 Push底层调用DoPush，需要重写 |
| bool | Pop(T& pt) 出队操作 |
| bool | virtual DoPop(T& pt) = 0 Push底层调用DoPop，需要重写 |
- OHOS::SafeQueue
class SafeQueue : public SafeQueueInner
| 返回值 | 名称 |
|---|---|
| void | DoPush(const T& pt) 入队操作 |
| bool | DoPop(T& pt) 出队操作 |
- OHOS::SafeStack
class SafeStack : public SafeQueueInner
| 返回值 | 名称 |
|---|---|
| void | DoPush(const T& pt) 入栈操作 |
| bool | DoPop(T& pt) 出栈操作 |
类关系图
源码剖析
源码是在safe_queue.h文件中直接实现的，主要包含两个类SafeStack、SafeQueue。
SafeStack模板类，可以存储任何类型的对象。它继承自 SafeQueueInner<T>，这意味着 SafeStack将继承 SafeQueueInner 的所有成员变量和成员函数，除了那些被重写的虚函数。SafeQueue模板类，可以存储任何类型的对象。它继承自 SafeQueueInner<T>，这意味着 SafeQueue 将继承 SafeQueueInner 的所有成员变量和成员函数，除了那些被重写的虚函数。两个类只有重写的push和pop函数有区别，主要在于实现队列为先进先出，栈为后进先出的特点。
应用示例
本案例主要完成如下工作：
- 创建2个子线程，1个线程负责入队操作，1个线程负责出队操作
- 子线程入队操作，每1秒做1次入队操作，循环5次
- 子线程入队操作，每2秒做1次出队操作，循环5次
// 定义栈变量
static OHOS::SafeStack<int> m_safeStack;static void funcSafeStackPush(const string &name)
{for (int i = 0; i < 5; i++) {// 入队操作cout << name << ", Push Start and i = " << i << endl;m_safeStack.Push(i);cout << name << ", Push Successful and i = " << i << " and value = " << i << endl;// 睡眠1秒cout << name << ", Sleep 1 sec" << endl;std::this_thread::sleep_for(std::chrono::milliseconds(1000));}
}static void funcSafeStackPop(const string &name)
{bool ret;int value;for (int i = 0; i < 5; i++) {// 出队操作cout << name << ", Pop Start and i = " << i << endl;ret = m_safeStack.Pop(value);if (ret) {cout << name << ", Pop Successful and i = " << i << " and ret = " << ret << " and value = " << value << endl;} else {cout << name << ", Pop Failed and i = " << i << endl;}// 睡眠2秒cout << name << ", Sleep 2 sec" << endl;std::this_thread::sleep_for(std::chrono::milliseconds(1000 * 2));}
}int main(int argc, char **argv)
{OHOS::ThreadPool threads("threads");string str_name;// 清空SafeStackm_safeStack.Clear();threads.SetMaxTaskNum(128);threads.Start(2);// 开启子线程，使用Pushstr_name = "Thread_SafeStack_Push";auto task_push = std::bind(funcSafeStackPush, str_name);threads.AddTask(task_push);// 开启子线程，使用Popstr_name = "Thread_SafeStack_Pop";auto task_pop = std::bind(funcSafeStackPop, str_name);threads.AddTask(task_pop);// 设置结束，并等待结束threads.Stop();cout << "Threads Stop" << endl;return 0;
}
- 执行结果
# ./utils_thread_safestack
Thread_SafeStack_Push, Push Start and i = 0
Thread_SafeStack_Push, Push Successful and i = 0 and value = 0
Thread_SafeStack_Push, Sleep 1 sec
Thread_SafeStack_Pop, Pop Start and i = 0
Thread_SafeStack_Pop, Pop Successful and i = 0 and ret = 1 and value = 0
Thread_SafeStack_Pop, Sleep 2 sec
Thread_SafeStack_Push, Push Start and i = 1
Thread_SafeStack_Push, Push Successful and i = 1 and value = 1
Thread_SafeStack_Push, Sleep 1 sec
Thread_SafeStack_Pop, Pop Start and i = 1
Thread_SafeStack_Pop, Pop Successful and i = 1 and ret = 1 and value = 1
Thread_SafeStack_Pop, Sleep 2 sec
Thread_SafeStack_Push, Push Start and i = 2
Thread_SafeStack_Push, Push Successful and i = 2 and value = 2
Thread_SafeStack_Push, Sleep 1 sec
Thread_SafeStack_Push, Push Start and i = 3
Thread_SafeStack_Push, Push Successful and i = 3 and value = 3
Thread_SafeStack_Push, Sleep 1 sec
Thread_SafeStack_Pop, Pop Start and i = 2
Thread_SafeStack_Pop, Pop Successful and i = 2 and ret = 1 and value = 3
Thread_SafeStack_Pop, Sleep 2 sec
Thread_SafeStack_Push, Push Start and i = 4
Thread_SafeStack_Push, Push Successful and i = 4 and value = 4
Thread_SafeStack_Push, Sleep 1 sec
Thread_SafeStack_Pop, Pop Start and i = 3
Thread_SafeStack_Pop, Pop Successful and i = 3 and ret = 1 and value = 4
Thread_SafeStack_Pop, Sleep 2 sec
Thread_SafeStack_Pop, Pop Start and i = 4
Thread_SafeStack_Pop, Pop Successful and i = 4 and ret = 1 and value = 2
Thread_SafeStack_Pop, Sleep 2 sec
Threads Stop
参考文档
C++公共基础库:包含简要说明、编译构建及各个功能节点的使用说明
凌智电子开发板