C++ Workflow：高性能异步任务调度引擎

icy 昨天 14 抢沙发

默认

摘要： C++ Workflow：高性能异步任务调度引擎项目概述 C++ Workflow 是搜狗公司开源的一款高性能异步任务调度引擎，专为现代C++开发设计。该项目提供了一套完整的异步...

C++ Workflow：高性能异步任务调度引擎

项目概述

C++ Workflow 是搜狗公司开源的一款高性能异步任务调度引擎，专为现代C++开发设计。该项目提供了一套完整的异步编程框架，让开发者能够轻松构建高性能、高并发的网络服务和应用。

核心特性

1. 统一的任务模型

Workflow 将各种计算任务抽象为统一的模型，包括： - 网络请求（HTTP、Redis、MySQL等） - 文件IO操作 - 定时器任务 - 计算密集型任务

2. 基于任务的异步调度

text

#include "workflow/WFTaskFactory.h"
#include "workflow/WFFacilities.h"

void http_callback(WFHttpTask *task) {
    if (task->get_state() == WFT_STATE_SUCCESS) {
        const void *body;
        size_t body_len;
        task->get_resp()->get_parsed_body(&body, &body_len);
        printf("Response: %.*s\n", (int)body_len, (const char *)body);
    }
}

int main() {
    // 创建HTTP任务
    WFHttpTask *task = WFTaskFactory::create_http_task(
        "https://httpbin.org/get", 
        3,  // 重试次数
        2,  // 重试间隔
        http_callback
    );
    
    // 启动任务
    task->start();
    
    // 等待任务完成
    WFFacilities::WaitGroup wait_group(1);
    wait_group.wait();
    
    return 0;
}

3. 串并联任务流

Workflow 支持复杂的任务依赖关系：

text

// 创建并行任务组
WFGoTask *task1 = WFTaskFactory::create_go_task("task1", []() {
    printf("Task1 running\n");
});

WFGoTask *task2 = WFTaskFactory::create_go_task("task2", []() {
    printf("Task2 running\n");
});

// 创建串行任务
SeriesWork *series = Workflow::create_series_work(task1, nullptr);
series->push_back(task2);

// 添加回调
series->set_callback([](const SeriesWork *series) {
    printf("All tasks completed\n");
});

// 启动任务流
series->start();

4. 完整的协议支持

HTTP/1.1, HTTP/2
Redis协议
MySQL协议
WebSocket
DNS解析

实际应用示例

示例1：并发HTTP请求

text

#include <vector>
#include "workflow/WFTaskFactory.h"
#include "workflow/WFFacilities.h"

int main() {
    const int n = 10;
    WFFacilities::WaitGroup wait_group(n);
    
    for (int i = 0; i < n; i++) {
        // 创建多个并发HTTP任务
        WFHttpTask *task = WFTaskFactory::create_http_task(
            "https://api.example.com/data",
            0, 0,
            [&wait_group](WFHttpTask *task) {
                if (task->get_state() == WFT_STATE_SUCCESS) {
                    printf("Request succeeded\n");
                }
                wait_group.done();
            }
        );
        
        task->start();
    }
    
    wait_group.wait();
    return 0;
}

示例2：构建微服务

text

// 服务端示例
#include "workflow/WFHttpServer.h"

int main() {
    WFHttpServer server([](WFHttpTask *task) {
        // 处理HTTP请求
        task->get_resp()->append_output_body("Hello from Workflow!");
    });
    
    if (server.start(8888) == 0) {
        getchar();  // 按回车键停止服务器
        server.stop();
    }
    
    return 0;
}

示例3：定时任务

text

#include "workflow/WFTaskFactory.h"

void timer_callback(WFTimerTask *task) {
    printf("Timer fired! User data: %s\n", 
           (const char *)task->user_data);
}

int main() {
    // 创建3秒后执行的定时器
    WFTimerTask *timer = WFTaskFactory::create_timer_task(
        3,      // 延迟秒数
        0,      // 重复间隔（0表示不重复）
        timer_callback
    );
    
    timer->user_data = (void *)"Custom Data";
    timer->start();
    
    getchar();  // 等待定时器触发
    return 0;
}