[C++知识库]分布式消息队列kafka讲解二（C/C++librdkafka客户端10/21未完待续）

• 版本信息：

librdkafka版本0.11.6
Kafka版本2.11-2.3.00

1.生产者

1）流程

1.初始化配置
2.创建初始化kafka配置信息
3.根据broker修改kafka配置bootstrap.server
4.设置消息回调函数，这个消息回调函数的作用是显示将消息送到kafka是否成功或失败
5.创建生产者示例
6.创建主题（长生命周期）
7.CTRL+C会清空标准输入缓冲区
8.不断检测循环标准输入缓冲区数据

8.1如果有数据则去掉数据的换行符
8.2输入为空继续循环
8.3发送生产的消息（异步发送）
8.4若发送的消息放入消息队列失败，打印一句话并放入poll异步回调

9.回收工作

2）参数备注

1.bootstrap.servers：(注意：示例代码只给了一个broker)
1)该参数用来指定生产者客户端连接Kafka集群所需的broker地址清单，具体的内容格式为host1：port1，host2：port2，可以设置一个或多个地址，中间以逗号隔开，此参数的默认值为“”。
2)注意这里并非需要所有的broker地址，因为生产者会从给定的broker里查找到其他broker的信息。不过建议至少要设置两个以上的broker 地址信息，当其中任意一个宕机时，生产者仍然可以连接到 Kafka集群上
--------------------------------------
2.key.serializer 和 value.serializer：（这里是java的，java跟C/C++不同）
broker 端接收的消息必须以字节数组（byte[]）的形式存在。代码清单2-1中生产者使用的KafkaProducer＜String，String＞和ProducerRecord＜String，String＞中的泛型＜String，String＞对应的就是消息中key和value的类型
--------------------------------------
3.topic  主题
--------------------------------------
4.brokerID  brokerID

3）代码（c风格代码）

/**
 * Simple Apache Kafka producer
 * using the Kafka driver from librdkafka
 * (https://github.com/edenhill/librdkafka)
 */

#include <stdio.h>
#include <signal.h>
#include <string.h>


/* Typical include path would be <librdkafka/rdkafka.h>, but this program
 * is builtin from within the librdkafka source tree and thus differs. */
#include "rdkafka.h"


static int run = 1;

/**
 * @brief Signal termination of program
 */
static void stop (int sig) {
        run = 0;
        fclose(stdin); /* abort fgets() */
}


/**
 * @brief Message delivery report callback.
 *
 * This callback is called exactly once per message, indicating if
 * the message was succesfully delivered
 * (rkmessage->err == RD_KAFKA_RESP_ERR_NO_ERROR) or permanently
 * failed delivery (rkmessage->err != RD_KAFKA_RESP_ERR_NO_ERROR).
 *
 * The callback is triggered from rd_kafka_poll() and executes on
 * the application's thread.
 */
static void dr_msg_cb (rd_kafka_t *rk,
                       const rd_kafka_message_t *rkmessage, void *opaque) {
        if (rkmessage->err)
                fprintf(stderr, "%% Message delivery failed: %s\n",
                        rd_kafka_err2str(rkmessage->err));
        else
                fprintf(stderr,
                        "%% Message delivered (%zd bytes, "
                        "partition %"PRId32")\n",
                        rkmessage->len, rkmessage->partition);

        /* The rkmessage is destroyed automatically by librdkafka */
}


//生产者：往固定的broker生产固定的topic
int main (int argc, char **argv) {
        //1.初始化配置
        rd_kafka_t *rk;         /* Producer instance handle *///生产者句柄
        rd_kafka_topic_t *rkt;  /* Topic object */
        rd_kafka_conf_t *conf;  /* Temporary configuration object */
        char errstr[512];       /* librdkafka API error reporting buffer */
        char buf[512];          /* Message value temporary buffer */
        const char *brokers;    /* Argument: broker list */
        const char *topic;      /* Argument: topic to produce to */
        //检查参数
        /*
         * Argument validation
         */
        if (argc != 3) {
                fprintf(stderr, "%% Usage: %s <broker> <topic>\n", argv[0]);
                return 1;
        }
        brokers = argv[1];      //获取固定的broker
        topic   = argv[2];      //获得主题参数



        //2.创建初始化kafka配置信息
        /*
         * Create Kafka client configuration place-holder
         */
        conf = rd_kafka_conf_new();


        //3.根据broker修改kafka配置bootstrap.server
        /* Set bootstrap broker(s) as a comma-separated list of
         * host or host:port (default port 9092).
         * librdkafka will use the bootstrap brokers to acquire the full
         * set of brokers from the cluster. */
        if (rd_kafka_conf_set(conf, "bootstrap.servers", brokers,
                              errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
                fprintf(stderr, "%s\n", errstr);
                return 1;
        }

        //4.设置消息回调函数，这个消息回调函数的作用是显示将消息送到kafka是否成功或失败
        /* Set the delivery report callback.
         * This callback will be called once per message to inform
         * the application if delivery succeeded or failed.
         * See dr_msg_cb() above. */
        rd_kafka_conf_set_dr_msg_cb(conf, dr_msg_cb);


        //5.创建生产者示例
        /*
         * Create producer instance.
         *
         * NOTE: rd_kafka_new() takes ownership of the conf object
         *       and the application must not reference it again after
         *       this call.
         */
        rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
        if (!rk) {
                fprintf(stderr,
                        "%% Failed to create new producer: %s\n", errstr);
                return 1;
        }

        //6.创建主题（长生命周期）
        /* Create topic object that will be reused for each message
         * produced.
         *
         * Both the producer instance (rd_kafka_t) and topic objects (topic_t)
         * are long-lived objects that should be reused as much as possible.
         */
        rkt = rd_kafka_topic_new(rk, topic, NULL);
        if (!rkt) {
                fprintf(stderr, "%% Failed to create topic object: %s\n",
                        rd_kafka_err2str(rd_kafka_last_error()));
                rd_kafka_destroy(rk);
                return 1;
        }

        //7.CTRL+C会清空标准输入缓冲区
        /* Signal handler for clean shutdown */
        signal(SIGINT, stop);

        fprintf(stderr,
                "%% Type some text and hit enter to produce message\n"
                "%% Or just hit enter to only serve delivery reports\n"
                "%% Press Ctrl-C or Ctrl-D to exit\n");

        //8.不断检测循环标准输入缓冲区数据
        while (run && fgets(buf, sizeof(buf), stdin)) {
                size_t len = strlen(buf);

                //8.1如果有数据则去掉数据的换行符
                if (buf[len-1] == '\n') /* Remove newline */
                        buf[--len] = '\0';

                //8.2输入为空继续循环
                if (len == 0) {
                        /* Empty line: only serve delivery reports */
                        rd_kafka_poll(rk, 0/*non-blocking */);
                        continue;
                }
                
                //8.3发送生产的消息（异步发送）
                //成功的化：会将消息放入消息队列，真正发消息到broker是守护进程分发消息的，发送成功或失败都会调用dr_msg_cb函数
                /*
                 * Send/Produce message.
                 * This is an asynchronous call, on success it will only
                 * enqueue the message on the internal producer queue.
                 * The actual delivery attempts to the broker are handled
                 * by background threads.
                 * The previously registered delivery report callback
                 * (dr_msg_cb) is used to signal back to the application
                 * when the message has been delivered (or failed).
                 */
        retry:
                //8.4若发送的消息放入消息队列失败，打印一句话并放入poll异步回调
                if (rd_kafka_produce(
                            /* Topic object */
                            rkt,                                                //传入主题
                            /* Use builtin partitioner to select partition*/
                            RD_KAFKA_PARTITION_UA,                              //
                            /* Make a copy of the payload. */
                            RD_KAFKA_MSG_F_COPY,
                            /* Message payload (value) and length */
                            buf, len,
                            /* Optional key and its length */
                            NULL, 0,
                            /* Message opaque, provided in
                             * delivery report callback as
                             * msg_opaque. */
                            NULL) == -1) {
                        /**
                         * Failed to *enqueue* message for producing.
                         */
                        fprintf(stderr,
                                "%% Failed to produce to topic %s: %s\n",
                                rd_kafka_topic_name(rkt),
                                rd_kafka_err2str(rd_kafka_last_error()));

                        /* Poll to handle delivery reports */
                        if (rd_kafka_last_error() ==
                            RD_KAFKA_RESP_ERR__QUEUE_FULL) {
                                /* If the internal queue is full, wait for
                                 * messages to be delivered and then retry.
                                 * The internal queue represents both
                                 * messages to be sent and messages that have
                                 * been sent or failed, awaiting their
                                 * delivery report callback to be called.
                                 *
                                 * The internal queue is limited by the
                                 * configuration property
                                 * queue.buffering.max.messages *///队列大小被queue.buffering.max.messages限制
                                rd_kafka_poll(rk, 1000/*block for max 1000ms*/);//设置非阻塞回调
                                goto retry;
                        }
                } 
                else 
                {
                        fprintf(stderr, "%% Enqueued message (%zd bytes) "
                                "for topic %s\n",
                                len, rd_kafka_topic_name(rkt));
                }//end of enqueue message


                /* A producer application should continually serve
                 * the delivery report queue by calling rd_kafka_poll()
                 * at frequent intervals.
                 * Either put the poll call in your main loop, or in a
                 * dedicated thread, or call it after every
                 * rd_kafka_produce() call.
                 * Just make sure that rd_kafka_poll() is still called
                 * during periods where you are not producing any messages
                 * to make sure previously produced messages have their
                 * delivery report callback served (and any other callbacks
                 * you register). */
                rd_kafka_poll(rk, 0/*non-blocking*/);
        }

        //9.回收工作
        /* Wait for final messages to be delivered or fail.
         * rd_kafka_flush() is an abstraction over rd_kafka_poll() which
         * waits for all messages to be delivered. */
        fprintf(stderr, "%% Flushing final messages..\n");
        //等待
        rd_kafka_flush(rk, 10*1000 /* wait for max 10 seconds */);

        /* Destroy topic object */
        rd_kafka_topic_destroy(rkt);

        /* Destroy the producer instance */
        rd_kafka_destroy(rk);

        return 0;
}

2.消费者

1）流程

2）参数备注

3）代码

3.主题和分区

4.日志存储

5.深入服务器

6.深入客户端

7.可靠性探究

8.kafka应用

9.kafka监控

10.kafka高级应用