目录

Function Flow Runtime并发队列(C)

概述

FFRT并发队列提供了设置任务优先级（Priority）和队列并发度的能力，使得队列中的任务能同时在多个线程上执行，获得更高的并行效果。

示例：银行服务系统

举例实现一个银行服务系统，每个客户向系统提交一个服务请求，可以区分普通用户和VIP用户，VIP用户的服务请求可以优先得到执行。银行系统中有2个窗口，可以并行取出用户提交的服务请求办理。

可以利用FFRT的并行队列范式做如下建模：

实现代码如下所示：

#include <cstdio>
#include <unistd.h>
#include "hilog/log.h"
#include "ffrt/ffrt.h" // 来自 OpenHarmony 第三方库 "@ppd/ffrt"

#undef LOG_TAG
#define LOG_TAG "ConcurrentTag"
const int SLEEP_TIME = 100 * 1000; // 100ms
const int BANK_CONCURRENCY = 2;

ffrt_queue_t CreateBankSystem(const char *name, int concurrency)
{
    ffrt_queue_attr_t queue_attr;
    (void)ffrt_queue_attr_init(&queue_attr);
    ffrt_queue_attr_set_max_concurrency(&queue_attr, concurrency);

    // 创建一个并发队列
    ffrt_queue_t queue = ffrt_queue_create(ffrt_queue_concurrent, name, &queue_attr);

    // 队列创建完后需要销毁队列属性
    ffrt_queue_attr_destroy(&queue_attr);
    if (!queue) {
        OH_LOG_ERROR(LOG_APP, "create queue failed");
        return NULL;
    }

    OH_LOG_INFO(LOG_APP, "create bank system successfully");
    return queue;
}

void DestroyBankSystem(ffrt_queue_t queue_handle)
{
    ffrt_queue_destroy(queue_handle);
    OH_LOG_INFO(LOG_APP, "destroy bank system successfully");
}

void BankBusiness(void *arg)
{
    usleep(SLEEP_TIME);
    const char *data = (const char *)arg;
    OH_LOG_INFO(LOG_APP, "saving or withdraw for %{public}s", data);
}

// 封装提交队列任务函数
ffrt_task_handle_t CommitRequest(ffrt_queue_t bank, void (*func)(void *), const char *name,
    ffrt_queue_priority_t level, int delay)
{
    ffrt_task_attr_t task_attr;
    (void)ffrt_task_attr_init(&task_attr);
    ffrt_task_attr_set_name(&task_attr, name);
    ffrt_task_attr_set_queue_priority(&task_attr, level);
    ffrt_task_attr_set_delay(&task_attr, delay);

    ffrt_task_handle_t handle = ffrt_queue_submit_h_f(bank, func, (void*)name, &task_attr);
    ffrt_task_attr_destroy(&task_attr);
    return handle;
}

// 封装取消队列任务函数
int CancelRequest(ffrt_task_handle_t request)
{
    return ffrt_queue_cancel(request);
}

// 封装等待队列任务函数
void WaitForRequest(ffrt_task_handle_t task)
{
    ffrt_queue_wait(task);
}

int ConcurrentQueueCExec()
{
    ffrt_queue_t bank = CreateBankSystem("Bank", BANK_CONCURRENCY);
    if (!bank) {
        printf("create bank system failed\n");
        OH_LOG_ERROR(LOG_APP, "create bank system failed");
        return -1;
    }

    ffrt_task_handle_t task1 = CommitRequest(bank, BankBusiness, "customer1", ffrt_queue_priority_low, 0);
    ffrt_task_handle_t task2 = CommitRequest(bank, BankBusiness, "customer2", ffrt_queue_priority_low, 0);
    // VIP享受更优先的服务
    ffrt_task_handle_t task3 = CommitRequest(bank, BankBusiness, "customer3 VIP", ffrt_queue_priority_high, 0);
    ffrt_task_handle_t task4 = CommitRequest(bank, BankBusiness, "customer4", ffrt_queue_priority_low, 0);
    ffrt_task_handle_t task5 = CommitRequest(bank, BankBusiness, "customer5", ffrt_queue_priority_low, 0);

    // 取消客户4的服务
    CancelRequest(task4);

    // 等待所有的客户服务完成
    WaitForRequest(task5);
    DestroyBankSystem(bank);

    ffrt_task_handle_destroy(task1);
    ffrt_task_handle_destroy(task2);
    ffrt_task_handle_destroy(task3);
    ffrt_task_handle_destroy(task4);
    ffrt_task_handle_destroy(task5);
    return 0;
}

ffrt_queue_submit_h_f接口可以接收裸函数指针任务作为参数，如果任务存在前后处理可以参见ffrt_alloc_auto_managed_function_storage_base函数查看如何构造任务结构体。

接口说明

上述样例中涉及到主要的FFRT的接口如下，详情请参考Function Flow Runtime C API里的方法：

名称 描述
ffrt_queue_create 创建队列。
ffrt_queue_destroy 销毁队列。
ffrt_task_attr_set_queue_priority 设置队列任务优先级。
ffrt_queue_attr_set_max_concurrency 设置并发队列的并发度。
ffrt_queue_submit_h_f

向队列提交一个任务。

说明：从API version 20开始，支持该接口。
  • 如何使用FFRT C++ API详见：FFRT C++接口三方库使用指导。
  • 使用FFRT C接口或C++接口时，都可以通过FFRT C++接口三方库简化头文件包含，即使用#include "ffrt/ffrt.h"头文件包含语句。

约束限制

  1. ffrt_queue_attr_t必须先调用ffrt_queue_attr_init初始化后再设置/获取属性，不再使用后需要显式调用ffrt_queue_attr_destroy释放资源。
  2. ffrt_queue_t必须在进程退出前显式调用ffrt_queue_destroy释放资源。
  3. 并发队列最大并发度建议控制在合理范围内，配置过大超过Worker线程数没有意义，配置过小可能导致系统资源利用率不足。