使用tutorial-13-kafka_cli.cc里面的代码学着操作kafka，有四点疑问，内容如下

1、在回调函数中加while无限循环，没有实现循环生产数据，也没有实现循环消费数据
2、在消费者客户端中不使用set_offset_store时，总是消费到所有数据，类似kafkad消费主题数据命令时加了--from-beginning
3、在消费中设置了config.set_offset_store(0);尝试了很多设置，也没有实现实时读取kafka中主题的当前数据
4、在消费者设置了消费组模式时，没有消费到任何数据，kafka命令是可以的

望大佬指点下，workflow_kafka的使用。

```
#include <netdb.h>
#include <unistd.h>
#include <signal.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <string>
#include "workflow/WFKafkaClient.h"
#include "workflow/KafkaMessage.h"
#include "workflow/KafkaResult.h"
#include "workflow/WFTaskFactory.h"
#include "workflow/WFFacilities.h"
#include "workflow/WFGlobal.h"

using namespace protocol;

static WFFacilities::WaitGroup wait_group(1);

std::string url;
bool no_cgroup = false;
WFKafkaClient client;

void kafka_callback(WFKafkaTask *task)
{
    int index = 0;

int api_type = task->get_api_type();

WFKafkaTask *next_task = NULL;
    std::vector<std::vector<KafkaRecord *>> records;
    std::vector<KafkaToppar *> toppars;

while (true)
    {
        printf("$$$...%d###%d\n", index, api_type);

int state = task->get_state();
        int error = task->get_error();

if (state != WFT_STATE_SUCCESS)
        {
            fprintf(stderr, "error msg: %s\n", WFGlobal::get_error_string(state, error));
            fprintf(stderr, "Failed. Press Ctrl-C to exit.\n");
            client.deinit();
            wait_group.done();
            return;
        }

protocol::KafkaResult new_result;

switch (api_type)
        {
        case Kafka_Produce:
            task->get_result()->fetch_records(records);

for (const auto &v : records)
            {
                for (const auto &w : v)
                {
                    //设置偏移
                    w->set_offset(w->get_offset() + 1);
                    //设置值
                    w->set_value("333", 3);

const void *value;
                    size_t value_len;
                    w->get_value(&value, &value_len);

printf("produce\ttopic: %s, partition: %d, status: %d, offset: %lld, val_len: %zu\n",
                           w->get_topic(),
                           w->get_partition(),
                           w->get_status(),
                           w->get_offset(), value_len);
                }
            }

break;

case Kafka_Fetch:
            new_result = std::move(*task->get_result());
            new_result.fetch_records(records);

if (!records.empty())
            {
                if (!no_cgroup)
                    next_task = client.create_kafka_task("api=commit", 3, kafka_callback);

std::string out;

for (const auto &v : records)
                {
                    if (v.empty())
                        continue;

// 设置偏移量
                    v.back()->set_offset(v.back()->get_offset() + 1);

char fn[1024];
                    snprintf(fn, 1024, "./kafka.%s.%d.%llu",
                             v.back()->get_topic(), 
                             v.back()->get_partition(),
                             v.back()->get_offset());

FILE *fp = fopen(fn, "w+");
                    long long offset = 0;
                    int partition = 0;
                    std::string topic;

for (const auto &w : v)
                    {
                        const void *value;
                        size_t value_len;
                        w->get_value(&value, &value_len);
                        if (fp)
                            // fwrite(value, value_len, 1, fp);

printf("$: %s\n", (char *)value);

offset = w->get_offset();
                        partition = w->get_partition();
                        topic = w->get_topic();

if (!no_cgroup)
                            next_task->add_commit_record(*w);
                    }

if (!topic.empty())
                    {
                        out += "topic: " + topic;
                        out += ",partition: " + std::to_string(partition);
                        out += ",offset: " + std::to_string(offset) + ";";
                    }

if (fp)
                        fclose(fp);
                }

printf("fetch\t%s\n", out.c_str());

if (!no_cgroup)
                    series_of(task)->push_back(next_task);
            }
            else
            {
                printf("fetch_records is null...\n");
            }

break;

case Kafka_OffsetCommit:
            task->get_result()->fetch_toppars(toppars);

if (!toppars.empty())
            {
                for (const auto &v : toppars)
                {
                    printf("commit\ttopic: %s, partition: %d, offset: %llu, error: %d\n",
                           v->get_topic(), v->get_partition(),
                           v->get_offset(), v->get_error());
                }
            }

next_task = client.create_leavegroup_task(3, kafka_callback);

series_of(task)->push_back(next_task);

break;

case Kafka_LeaveGroup:
            printf("leavegroup callback\n");
            break;

default:
            break;
        }

if (!next_task)
        {
            client.deinit();
            wait_group.done();
        }

next_task = NULL;
        records.capacity();
        toppars.capacity();

index++;

sleep(1);
    }
}

void sig_handler(int signo) {}

int main(int argc, char *argv[])
{
    // if (argc < 3)
    // {
    //     fprintf(stderr, "USAGE: %s url <p/c> [compress_type/d]\n", argv[0]);
    //     exit(1);
    // }

signal(SIGINT, sig_handler);

// url = argv[1];
    // if (strncmp(argv[1], "kafka://", 8) != 0)
    //     url = "kafka://" + url;

url = "192.168.31.121:9092";

// char buf[512 * 1024];
    char buf[100] = "H-H-H";
    WFKafkaTask *task;

int _select = 2;
    if (_select == 1)
    {
        int compress_type = Kafka_NoCompress;

// if (argc > 3)
        //     compress_type = atoi(argv[3]);

if (compress_type > Kafka_Zstd)
            exit(1);

client.init(url);
        task = client.create_kafka_task("api=produce", 3, kafka_callback);
        KafkaConfig config;
        KafkaRecord record;

config.set_compress_type(compress_type);
        config.set_client_id("workflow");
        task->set_config(std::move(config));

// for (size_t i = 0; i < sizeof(buf); ++i)
        //     buf[i] = '1' + rand() % 128;

record.set_key("key1", strlen("key1"));
        record.set_value(buf, sizeof(buf));
        record.add_header_pair("hk1", 3, "hv1", 3);
        task->add_produce_record("workflow_test1", -1, std::move(record));

record.set_key("key2", strlen("key2"));
        record.set_value(buf, sizeof(buf));
        record.add_header_pair("hk2", 3, "hv2", 3);
        task->add_produce_record("workflow_test2", -1, std::move(record));
    }
    else if (_select == 2)
    {
        int cgroup = 1;
        if (cgroup == 0)
        {
            client.init(url);
            task = client.create_kafka_task("api=fetch", 3, kafka_callback);

KafkaToppar toppar;
            toppar.set_topic_partition("workflow_test1", 0);
            toppar.set_offset(0);
            task->add_toppar(toppar);

toppar.set_topic_partition("workflow_test2", 0);
            toppar.set_offset(1);
            task->add_toppar(toppar);

no_cgroup = true;
        }
        else
        {
            client.init(url, "workflow_group");
            task = client.create_kafka_task("topic=workflow_test1&topic=workflow_test2&api=fetch",
                                            3, kafka_callback);
        }

KafkaConfig config;
        config.set_client_id("workflow");

// 0：优先使用上次提交； 1：使用指定的offset
        config.set_offset_store(0);

task->set_config(std::move(config));
    }
    else
    {
        fprintf(stderr, "USAGE: %s url <p/c> [compress_type/d]\n", argv[0]);
        exit(1);
    }

task->start();

wait_group.wait();

while (true)
    {
        /* code */
    }

return 0;
}

```

GVP 搜狗开源/workflow

内容风险标识

评论 (10)

GVP搜狗开源/workflow

内容风险标识