C++异步网络库workflow系列教程(3)Series串联任务流

news2024/11/14 3:09:34

往期教程

如果觉得写的可以,请给一个点赞+关注支持一下

观看之前请先看,往期的两篇博客教程,否则这篇博客没办法看懂

workFlow c++异步网络库编译教程与简介
C++异步网络库workflow入门教程(1)HTTP任务
C++异步网络库workflow系列教程(2)redis任务

简介

首先,workflow是任务流的意思,在workflow中万物皆为任务流.任务流分为以下两种

串联:就像链表一样,前面的任务执行完了后执行后面的
并联:可以理解为开了多个线程,并发执行任务

无标题

示例代码

首先,还是老规矩,先看代码示例,我会将所有新出现的成员一一进行介绍

#include <workflow/WFFacilities.h>
void seriesCallback(const SeriesWork *series){
    fprintf(stderr,"series callback , free pkey\n");
    std::string *pkey = static_cast<std::string *>(series->get_context());
    delete pkey;
}
void callback(WFRedisTask *redisTask){
        protocol::RedisRequest *req = redisTask->get_req();
        protocol::RedisResponse *resp = redisTask->get_resp();
        int state = redisTask->get_state();
        int error = redisTask->get_error();
        protocol::RedisValue value;//value对象专门用来存储redis任务的结果
        switch (state){
        case WFT_STATE_SYS_ERROR:
            fprintf(stderr,"system error: %s\n", strerror(error));
            break;
        case WFT_STATE_DNS_ERROR:
            fprintf(stderr,"dns error: %s\n", gai_strerror(error));
            break;
        case WFT_STATE_SUCCESS:
            resp->get_result(value);
            if(value.is_error()){
                fprintf(stderr,"redis error\n");
                state = WFT_STATE_TASK_ERROR;
            }
            break;
        }

        if (state != WFT_STATE_SUCCESS){
            fprintf(stderr, "Failed\n");
            return;
        }
        else{
            fprintf(stderr, "Success!\n");
        }

        std::string cmd;
        req->get_command(cmd);
        if(cmd=="SET"){
            //firstTask的基本工作做完了
            //创建新任务，把新任务加入到本序列的末尾
            fprintf(stderr,"first task callback begins\n");
            std::string *pkey = static_cast<std::string *>(redisTask->user_data);
            WFRedisTask *secondTask = WFTaskFactory::create_redis_task("redis://127.0.0.1:6379",0,callback);
            protocol::RedisRequest *req = secondTask->get_req();
            req->set_request("GET",{*pkey});
            SeriesWork* series = series_of(redisTask);
            series->set_context(static_cast<void *>(pkey));
            series->set_callback(seriesCallback);
            series->push_back(secondTask);
            fprintf(stderr,"first task callback ends\n");
        }
        else{
            //secondTask的基本工作做完了
            fprintf(stderr,"second task callback begins\n");
            fprintf(stderr, "redis request, cmd = %s\n", cmd.c_str());
            if (value.is_string()){
                fprintf(stderr, "value is a string, value = %s\n", value.string_value().c_str());
            }
            else if (value.is_array()){
                fprintf(stderr, "value is string array\n");
                for (size_t i = 0; i < value.arr_size(); ++i){
                    fprintf(stderr, "value at %lu = %s\n", i, value.arr_at(i).string_value().c_str());
                }
            }
            fprintf(stderr,"second task callback ends\n");
        }
    }
int main(){
    //创建redis任务
    //std::string key = "43key1";
    std::string * pkey = new std::string("43key2");
    WFRedisTask *firstTask = WFTaskFactory::create_redis_task("redis://127.0.0.1:6379",0,callback);
    //设置redis任务的属性
    protocol::RedisRequest *req = firstTask->get_req();
    req->set_request("SET",{*pkey, "200"});
    firstTask->user_data = static_cast<void *>(pkey);
    //启动redis任务
    firstTask->start();
}

示例剖析

首先我们先看见main函数中第77行firstTask->user_data,如下是该成员的源码实现截图,为一个void*指针,通过变量名我们可以知道这是用来存储用户上下文的变量,可以在响应回调函数中将数据取出来进行使用(如43行中取出使用)

任务流

series_of(redisTask)函数可以通过任务流节点获取到该任务节点所属的任务流的指针(在main函数中调用的firstTask->start();便是创建了一个任务流),
set_context()与上面的user_data类似,不过set_context设置的生存周期在整个任务流的生存周期,而user_data的生存周期只在所属的任务流节点的生存周期
set_callback()用来设置该任务流的清理回调函数,调用时机在该任务流所有任务节点全部执行完毕之后
series->push_back(secondTask):将secondTask这个任务添加在任务流的末尾节点