用c语言应该如何编写

使用C语言编写Kafka客户端需要使用Kafka C语言API，这个API提供了一组函数，可以在C语言中使用Kafka的功能。

以下是一个简单的示例程序，使用Kafka C语言API从Kafka集群消费消息：

#include <stdio.h>
#include <stdlib.h>
#include <librdkafka/rdkafka.h>

int main(int argc, char **argv) {
    rd_kafka_t *rk;
    rd_kafka_conf_t *conf;
    char errstr[512];
    rd_kafka_resp_err_t err;
    rd_kafka_topic_partition_list_t *topics;
    rd_kafka_topic_partition_t *topic;
    rd_kafka_message_t *msg;

    /* Create Kafka configuration object */
    conf = rd_kafka_conf_new();

    /* Set bootstrap server */
    if (rd_kafka_conf_set(conf, "bootstrap.servers", "localhost:9092", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        fprintf(stderr, "Failed to set broker list: %s\n", errstr);
        exit(1);
    }

    /* Create Kafka producer instance */
    rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
    if (rk == NULL) {
        fprintf(stderr, "Failed to create producer: %s\n", errstr);
        exit(1);
    }

    /* Create topic partition list */
    topics = rd_kafka_topic_partition_list_new(1);

    /* Add topic partition to list */
    topic = rd_kafka_topic_partition_list_add(topics, "my-topic", RD_KAFKA_PARTITION_UA);
    if (topic == NULL) {
        fprintf(stderr, "Failed to add topic: %s\n", rd_kafka_err2str(err));
        rd_kafka_topic_partition_list_destroy(topics);
        rd_kafka_destroy(rk);
        exit(1);
    }

    /* Subscribe to topic */
    err = rd_kafka_subscribe(rk, topics);
    if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
        fprintf(stderr, "Failed to subscribe to topic: %s\n", rd_kafka_err2str(err));
        rd_kafka_topic_partition_list_destroy(topics);
        rd_kafka_destroy(rk);
        exit(1);
    }

    /* Consume messages */
    while (1) {
        msg = rd_kafka_consumer_poll(rk, 1000); // Wait up to 1000ms for a message
        if (msg == NULL) {
            continue;
        }
        if (msg->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
            printf("Received message: %.*s\n", (int)msg->len, (char *)msg->payload);
            rd_kafka_message_destroy(msg);
        }
    }

    /* Cleanup */
    rd_kafka_topic_partition_list_destroy(topics);
    rd_kafka_destroy(rk);

    return 0;
}

这个程序使用`rd_kafka_new()`函数创建了一个Kafka消费者实例，使用`rd_kafka_subscribe()`函数订阅了一个主题，然后使用`rd_kafka_consumer_poll()`函数从Kafka集群中消费消息。

程序中还使用了一些其他的函数，例如`rd_kafka_conf_set()`用于设置Kafka配置，`rd_kafka_topic_partition_list_new()`用于创建主题分区列表，`rd_kafka_topic_partition_list_add()`用于将主题分区添加到列表中等等。

当然，这只是一个简单的示例程序，实际使用中需要根据具体的需求进行更多的配置。