基于OpenHarmony标准系统的C++公共基础类库案例：SafeBlockQueue
1、程序简介
该程序是基于OpenHarmony的C++公共基础类库的读写锁：SafeBlockQueue。
线程安全阻塞队列SafeBlockQueue类，提供阻塞和非阻塞版的入队入队和出队接口，并提供可最追踪任务完成状态的的SafeBlockQueueTracking类。
本案例主要完成如下工作：
（1）使用SafeBlockQueue接口的案例
判断命令行是否使用阻塞，还是非阻塞；
创建子线程生产者，使用阻塞/非阻塞方式，入队操作；
创建子线程消费者，使用阻塞/非阻塞方式，出队操作；
主线程等待所有子线程结束
（2）使用SafeBlockQueueTracking接口的案例
判断命令行是否使用阻塞，还是非阻塞；
创建子线程生产者，使用阻塞/非阻塞方式，入队操作；
创建子线程消费者，使用阻塞/非阻塞方式，出队操作；
主线程等待所有子线程结束
该案例已在凌蒙派-RK3568开发板验证过，如需要完整源代码，请参考：
https://gitee.com/Lockzhiner-Electronics/lockzhiner-rk3568-openharmony/tree/master/samples/a29_utils_safeblockqueue
2、基础知识
C++公共基础类库为标准系统提供了一些常用的C++开发工具类，包括：
文件、路径、字符串相关操作的能力增强接口
安全数据容器、数据序列化等接口
各子系统的错误码相关定义
2.1、添加C++公共基础类库依赖
修改需调用模块的BUILD.gn，在external_deps或deps中添加如下：
ohos_shared_library("xxxxx") { ... external_deps = [ ... # 动态库依赖(可选) "c_utils:utils", # 静态库依赖(可选) "c_utils:utilsbase", # Rust动态库依赖(可选) "c_utils:utils_rust", ] ...}
一般而言，我们只需要填写"c_utils:utils"即可。
2.2、SafeBlockQueue头文件
C++公共基础类库的SafeBlockQueue头文件在：//commonlibrary/c_utils/base/include/safe_block_queue.h
可在源代码中添加如下：
#include
2.3、OHOS::SafeBlockQueue接口说明
2.3.1、SafeBlockQueue
构造函数。
SafeBlockQueue(int capacity)
参数说明：
| 参数名称 | 类型 | 参数说明 |
|---|---|---|
| capacity | int | SafeBlockQueue的容量，即能存储多少个单元 |
2.3.2、~SafeBlockQueue
析构函数。
~SafeBlockQueue();
2.3.3、Push
入队操作(阻塞版)。
void virtual Push(T const& elem);
2.3.4、PushNoWait
入队操作(非阻塞版)。
bool virtual PushNoWait(T const& elem);
返回值说明：
| 类型 | 返回值说明 |
|---|---|
| bool | true表示成功，false表示失败 |
2.3.5、Pop
出队操作(阻塞版)。
T Pop();
返回值说明：
| 类型 | 返回值说明 |
|---|---|
| T | 出队的单元 |
2.3.6、PopNotWait
出队操作(非阻塞版)。
bool PopNotWait(T& outtask);
参数说明：
| 参数名称 | 类型 | 参数说明 |
|---|---|---|
| outtask | T | 出队的单元 |
返回值说明：
| 类型 | 返回值说明 |
|---|---|
| bool | true表示成功，false表示失败 |
2.3.7、Size
获取队列容量。
unsigned int Size();
返回值说明：
| 类型 | 返回值说明 |
|---|---|
| unsigned int | 返回队列的容量 |
2.3.8、IsEmpty
队列判空。
bool IsEmpty;
返回值说明：
| 类型 | 返回值说明 |
|---|---|
| bool | true表示成功，false表示失败 |
2.3.9、IsFull
判断map是否为满。
bool IsFull();
返回值说明：
| 类型 | 返回值说明 |
|---|---|
| bool | true表示空，false表示非空 |
2.4、OHOS::SafeBlockQueueTracking接口说明
2.4.1、SafeBlockQueue
构造函数。
SafeBlockQueue(int capacity)
参数说明：
| 参数名称 | 类型 | 参数说明 |
|---|---|---|
| capacity | int | SafeBlockQueue的容量，即能存储多少个单元 |
2.4.2、~SafeBlockQueue
析构函数。
~SafeBlockQueue();
2.4.3、Push
入队操作(阻塞版)。
void virtual Push(T const& elem);
2.4.4、PushNoWait
入队操作(非阻塞版)。
bool virtual PushNoWait(T const& elem);
返回值说明：
| 类型 | 返回值说明 |
|---|---|
| bool | true表示成功，false表示失败 |
3、程序解析
3.1、创建编译引导
在上一级目录BUILD.gn文件添加一行编译引导语句。
import("//build/ohos.gni")
group("samples") { deps = [ "a29_utils_safeblockqueue:utils_safeblockqueue", # 添加该行 ]}
"a29_utils_safeblockqueue:utils_safeblockqueue",该行语句表示引入 参与编译。
3.2、创建编译项目
创建a29_utils_safeblockqueue目录，并添加如下文件：
a29_utils_safeblockqueue├── utils_safeblockqueue_sample.cpp # .cpp源代码├── utils_safeblockqueuetracking_sample.cpp # .cpp源代码├── BUILD.gn # GN文件
3.3、创建BUILD.gn
编辑BUILD.gn文件。
import("//build/ohos.gni")
ohos_executable("utils_safeblockqueue_test") { sources = [ "utils_safeblockqueue_sample.cpp" ] include_dirs = [ "//commonlibrary/c_utils/base/include", "//commonlibrary/c_utils/base:utils", "//third_party/googletest:gtest_main", "//third_party/googletest/googletest/include" ] external_deps = [ "c_utils:utils" ] part_name = "product_rk3568" install_enable = true}
ohos_executable("utils_safeblockqueue_tracking") { sources = [ "utils_safeblockqueuetracking_sample.cpp" ] include_dirs = [ "//commonlibrary/c_utils/base/include", "//commonlibrary/c_utils/base:utils", "//third_party/googletest:gtest_main", "//third_party/googletest/googletest/include" ] external_deps = [ "c_utils:utils" ] part_name = "product_rk3568" install_enable = true}
group("utils_safeblockqueue") { deps = [ ":utils_safeblockqueue_test", # 构建SafeBlockQueue案例 ":utils_safeblockqueue_tracking", # 构建SafeBlockQueueTracking案例 ]}
注意：
（1）BUILD.gn中所有的TAB键必须转化为空格，否则会报错。如果自己不知道如何规范化，可以：
# 安装gn工具sudo apt-get install ninja-buildsudo apt install generate-ninja# 规范化BUILD.gngn format BUILD.gn
3.4、创建SafeBlockQueue案例源代码
3.4.1、创建SafeBlockQueue变量
#include // SafeBlockQueue的头文件
// 定义常量const int SIZE = 5;// 定义SafeBlockQueue变量OHOS::SafeBlockQueue
3.4.2、获知运行阻塞/非阻塞方式
命令有1个参数，分别是：
wait：使用阻塞方式的SafeBlockQueue；
nowait：使用非阻塞方式的SafeBlockQueue；
int main(int argc, char **argv){ bool enable_wait = true; ...... // 获取命令行参数 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; }}
3.4.3、创建线程池并设置子线程数量
int main(int argc, char **argv){ OHOS::ThreadPool threads("threads"); string str_name; ...... threads.SetMaxTaskNum(4); threads.Start(2); ......}
3.4.4、启动2个子线程，并等待结束
调用AddTask()添加子线程，并调用Stop()等待所有子进程结束。
// 创建生产者线程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::sleep_for(std::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;
3.4.5、子线程生产者，使用阻塞方式
static void product_wait(const string &name){ for (int i = 0; i < (2 * SIZE); i++) { cout << get_curtime() << ", " << __func__ << ": Push Start, i = " << i << endl; // 使用阻塞方式的SafeBlockQueue m_safeBlockQueue.Push(i); cout << get_curtime() << ", " << __func__ << ": Push Success, i = " << i << endl; // 等待1秒 cout << get_curtime() << ", " << __func__ << ": Sleep 1 sec " << endl; std::sleep_for(std::milliseconds(1000)); }}
3.4.6、子线程消费者，使用阻塞方式
static void consume_wait(const string &name){ for (int i = 0; i < (2 * SIZE); i++) { cout << get_curtime() << ", " << __func__ << ": Pop Start, i = " << i << endl; // 使用阻塞方式的SafeBlockQueue int 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::sleep_for(std::milliseconds(500)); }}
3.4.7、子线程生产者，使用非阻塞方式
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; // 使用非阻塞方式的SafeBlockQueue ret = m_safeBlockQueue.PushNoWait(i); cout << get_curtime() << ", " << __func__ << ": Push ret = " << ret << ", i = " << i << endl; // 等待1秒 cout << get_curtime() << ", " << __func__ << ": Sleep 1 sec " << endl; std::sleep_for(std::milliseconds(1000)); }}
3.4.8、子线程消费者，使用非阻塞方式
static void consume_nowait(const string &name){ for (int i = 0; i < (SIZE * 2); i++) { // 等待有新数据 int value = 0; // 使用非阻塞方式的SafeBlockQueue bool ret = m_safeBlockQueue.PopNotWait(value); cout << get_curtime() << ", " << __func__ << ": PopNotWait ret = " << ret << ", value = " << value << endl; // 等待500毫秒 std::sleep_for(std::milliseconds(500)); }}
3.5、创建SafeBlockQueueTracking案例源代码
3.5.1、创建SafeBlockQueueTracking变量
#include // SafeBlockQueue的头文件
// 定义常量const int SIZE = 5;// 定义SafeBlockQueue变量OHOS::SafeBlockQueueTracking
3.5.2、获知运行阻塞/非阻塞方式
命令有1个参数，分别是：
wait：使用阻塞方式的SafeBlockQueue；
nowait：使用非阻塞方式的SafeBlockQueue；
int main(int argc, char **argv){ bool enable_wait = true; ...... // 获取命令行参数 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; }}
3.5.3、创建线程池并设置子线程数量
int main(int argc, char **argv){ OHOS::ThreadPool threads("threads"); string str_name; ...... threads.SetMaxTaskNum(4); threads.Start(2); ......}
3.5.4、启动2个子线程，并等待结束
调用AddTask()添加子线程，并调用Stop()等待所有子进程结束。
// 创建生产者线程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::sleep_for(std::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;
3.5.5、子线程生产者，使用阻塞方式
static void product_wait(const string &name){ for (int i = 0; i < (2 * SIZE); i++) { cout << get_curtime() << ", " << __func__ << ": Push Start, i = " << i << endl; // 使用阻塞方式的SafeBlockQueueTracking m_safeBlockQueueTracking.Push(i); cout << get_curtime() << ", " << __func__ << ": Push Success, i = " << i << endl; // 等待1秒 cout << get_curtime() << ", " << __func__ << ": Sleep 1 sec " << endl; std::sleep_for(std::milliseconds(1000)); }}
3.5.6、子线程消费者，使用阻塞方式
static void consume_wait(const string &name){ for (int i = 0; i < (2 * SIZE); i++) { cout << get_curtime() << ", " << __func__ << ": Pop Start, i = " << i << endl; // 使用阻塞方式的SafeBlockQueueTracking int 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::sleep_for(std::milliseconds(500)); }
}
3.5.7、子线程生产者，使用非阻塞方式
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; // 使用非阻塞方式的SafeBlockQueueTracking ret = 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::sleep_for(std::milliseconds(1000)); }}
3.5.8、子线程消费者，使用非阻塞方式
static void consume_nowait(const string &name){ for (int i = 0; i < (SIZE * 2); i++) { // 等待有新数据 int value = 0; // 使用非阻塞方式的SafeBlockQueueTracking bool ret = m_safeBlockQueueTracking.PopNotWait(value); cout << get_curtime() << ", " << __func__ << ": PopNotWait ret = " << ret << ", value = " << value << endl; // 等待500毫秒 std::sleep_for(std::milliseconds(500)); }
}
4、编译步骤
进入OpenHarmony编译环境，运行命令：
hb build -f
5、运行结果
#
-
Queue
+关注
关注
0文章
16浏览量
7261 -
系统
+关注
关注
1文章
1014浏览量
21332 -
OpenHarmony
+关注
关注
25文章
3713浏览量
16251
发布评论请先 登录
相关推荐
基于OpenHarmony标准系统的C++公共基础类库案例：ThreadPoll
基于OpenHarmony标准系统的C++公共基础类库案例：Semaphore
基于OpenHarmony标准系统的C++公共基础类库案例：rwlock
基于OpenHarmony标准系统的C++公共基础类库案例：SafeMap
基于OpenHarmony标准系统的C++公共基础类库案例：SafeQueue
基于OpenHarmony标准系统的C++公共基础类库案例：SafeStack
基于OpenHarmony标准系统的C++公共基础类库案例：SafeBlockQueue
OpenHarmony C++公共基础类库应用案例：Thread
OpenHarmony C++公共基础类库应用案例：Thread
OpenHarmony C++公共基础类库应用案例：HelloWorld
OpenHarmony标准系统C++公共基础类库案例：HelloWorld
基于OpenHarmony标准系统的C++公共基础类库案例：SafeStack
基于OpenHarmony标准系统的C++公共基础类库案例：SafeQueue
基于OpenHarmony标准系统的C++公共基础类库案例：SafeMap
基于OpenHarmony标准系统的C++公共基础类库案例：rwlock
工商网监
评论