基于OpenHarmony标准系统的C++公共基础类库案例：SafeBlockQueue- 大数跨境

首页

基于OpenHarmony标准系统的C++公共基础类库案例：SafeBlockQueue

凌智电子

2024-08-22

导读：该程序是基于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 <safe_block_queue.h>

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 <safe_block_queue.h>       // SafeBlockQueue的头文件
// 定义常量const int SIZE = 5;// 定义SafeBlockQueue变量OHOS::SafeBlockQueue<int> m_safeBlockQueue(SIZE);

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::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;

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::this_thread::sleep_for(std::chrono::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::this_thread::sleep_for(std::chrono::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::this_thread::sleep_for(std::chrono::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::this_thread::sleep_for(std::chrono::milliseconds(500));    }}

3.5、创建SafeBlockQueueTracking案例源代码

3.5.1、创建SafeBlockQueueTracking变量

#include <safe_block_queue.h>       // SafeBlockQueue的头文件
// 定义常量const int SIZE = 5;// 定义SafeBlockQueue变量OHOS::SafeBlockQueueTracking<int> m_safeBlockQueueTracking(SIZE);

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::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;

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::this_thread::sleep_for(std::chrono::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::this_thread::sleep_for(std::chrono::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::this_thread::sleep_for(std::chrono::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::this_thread::sleep_for(std::chrono::milliseconds(500));    }
}
 
 
 
4、编译步骤
进入OpenHarmony编译环境，运行命令：
  
  
  
   
   
   
    
    
    
   
   
   
   
   
   hb build -f
  
  
  
5、运行结果
  
  
  
   
   
   
    
    
    
   
   
   
   
   
   #

【声明】内容源于网络

凌智电子

凌智电子-Lockzhiner Electronic，专注仪器仪表20年，一定带给您更多的方便与惊喜！

内容 118

粉丝 0

凌智电子凌智电子-Lockzhiner Electronic，专注仪器仪表20年，一定带给您更多的方便与惊喜！

总阅读3

粉丝0

内容118