kafka C客户端librdkafka producer源码分析

简介

kafka网站上提供了C语言的客户端librdkafka，地址在这。

librdkafka是使用C语言根据apache kafka 协议实现的客户端。另外这个客户端还有简单的c++接口。客户端作者对这个客户端比较上心，经常会修改bug并提交新功能。

librdkafka的基本原理和我之前博客说的java版producer类似，一个线程向队列中加数据，另一个线程通过非阻塞的方式从队列中取出数据，并写入到broker。

源码分析

源码包含两个文件夹src和src-cpp

src是用c实现的源码，而src-cpp是在c接口上包装的一层c++类，实现了基本的功能。

代码运行流程如下

1、rd_kafka_conf_set设置全局配置

2、rd_kafka_topic_conf_set设置topic配置

3、rd_kafka_brokers_add设置broker地址，启动向broker发送消息的线程

4、rd_kafka_new启动kafka主线程

5、rd_kafka_topic_new建topic

6、rd_kafka_produce使用本函数发送消息

7、rd_kafka_poll调用回调函数

还是看发送一条消息的过程

入队列过程

 调用rd_kafka_produce可以将消息写到队列

int rd_kafka_produce (...) {
    //调用rd_kafka_msg_new
    return rd_kafka_msg_new(...);
}

首先先将消息包装成rd_kafka_msg_t类型，然后获取分区并相应的队列

int rd_kafka_msg_new (...) {
    ...
    //创建消息，将传入的参数转换为rkm
    rkm = rd_kafka_msg_new0(...);
    //分区并入队
    err = rd_kafka_msg_partitioner(rkt, rkm, 1);
    ...
    return -1;
}

int rd_kafka_msg_partitioner (...) {
     ...
     //获取分区号
     switch (rkt->rkt_state)
     {
         ...
     }
    //获取分区
    rktp_new = rd_kafka_toppar_get(rkt, partition, 0);
    ...
    //加入队列
    rd_kafka_toppar_enq_msg(rktp_new, rkm);
    return 0;
}

出队列过程

添加broker的过程中就启动了扫描队列的操作

static rd_kafka_broker_t *rd_kafka_broker_add (rd_kafka_t *rk,
                           rd_kafka_confsource_t source,
                           const char *name, uint16_t port,
                           int32_t nodeid) {
    ...
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    //启动向broker发送消息的主线程
    if ((err = pthread_create(&rkb->rkb_thread, &attr,
                  rd_kafka_broker_thread_main, rkb))) {
        ...
        return NULL;
    }
    //将broker加到broker队列中
    TAILQ_INSERT_TAIL(&rkb->rkb_rk->rk_brokers, rkb, rkb_link);
    (void)rd_atomic_add(&rkb->rkb_rk->rk_broker_cnt, 1);
    ...
    return rkb;
}

启动rd_kafka_broker_thread_main主线程

static void *rd_kafka_broker_thread_main (void *arg) {
    ...
    while (!rkb->rkb_rk->rk_terminate) {
        switch (rkb->rkb_state)
        {
        //如果broker连接未初始化，或中断，则不断重连broker
        case RD_KAFKA_BROKER_STATE_INIT:
        case RD_KAFKA_BROKER_STATE_DOWN:
            if (rd_kafka_broker_connect(rkb) == -1) {
                ...
            }
            break;
        //如果broker连接已经建立，则调用serve函数
        case RD_KAFKA_BROKER_STATE_UP:
            if (rkb->rkb_nodeid == RD_KAFKA_NODEID_UA)
                rd_kafka_broker_ua_idle(rkb);
            else if (rk->rk_type == RD_KAFKA_PRODUCER)
                rd_kafka_broker_producer_serve(rkb);
            else if (rk->rk_type == RD_KAFKA_CONSUMER)
                rd_kafka_broker_consumer_serve(rkb);
            break;
        }
    }
    ...
    return NULL;
}

只看producer的处理函数，该函数扫描消息并发送

static void rd_kafka_broker_producer_serve (rd_kafka_broker_t *rkb) {
    ...
    while (!rkb->rkb_rk->rk_terminate &&
           rkb->rkb_state == RD_KAFKA_BROKER_STATE_UP) {
        ...
        do {
            cnt = 0;
            ...
            //扫描所有的topic-partitions，并发送消息
            TAILQ_FOREACH(rktp, &rkb->rkb_toppars, rktp_rkblink) {
                ...
                //将入队过程中的队列rktp_msgq加到rktp_xmit_msgq中
                if (rktp->rktp_msgq.rkmq_msg_cnt > 0)
                    rd_kafka_msgq_concat(&rktp->
                                 rktp_xmit_msgq,
                                 &rktp->rktp_msgq);
                rd_kafka_toppar_unlock(rktp);
                //扫描消息队列中数据是否超时
                if (unlikely(do_timeout_scan))
                    rd_kafka_msgq_age_scan(&rktp->
                                   rktp_xmit_msgq,
                                   &timedout,
                                   now);
                //队列为空则从头继续
                if (rktp->rktp_xmit_msgq.rkmq_msg_cnt == 0)
                    continue;
                
                //如果没有超时，或者没达到处理消息数量的阈值，则从头继续，这样批处理可以提高性能
                if (rktp->rktp_ts_last_xmit +
                    (rkb->rkb_rk->rk_conf.
                     buffering_max_ms * 1000) > now &&
                    rktp->rktp_xmit_msgq.rkmq_msg_cnt <
                    rkb->rkb_rk->rk_conf.
                    batch_num_messages) {
                    /* Wait for more messages */
                    continue;
                }

                rktp->rktp_ts_last_xmit = now;

                //按协议转换并填充数据到rkb中
                while (rktp->rktp_xmit_msgq.rkmq_msg_cnt > 0) {
                    int r = rd_kafka_broker_produce_toppar(
                        rkb, rktp);
                    if (likely(r > 0))
                        cnt += r;
                    else
                        break;
                }
            }

        } while (cnt);

        //触发数据发送情况的回调函数，将发送失败的写到一个操作结果队列中
        if (unlikely(isrfailed.rkmq_msg_cnt > 0))
            rd_kafka_dr_msgq(rkb->rkb_rk, &isrfailed,
                     RD_KAFKA_RESP_ERR__ISR_INSUFF);

        if (unlikely(timedout.rkmq_msg_cnt > 0))
            rd_kafka_dr_msgq(rkb->rkb_rk, &timedout,
                     RD_KAFKA_RESP_ERR__MSG_TIMED_OUT);

        rd_kafka_broker_toppars_unlock(rkb);

        /* Check and move retry buffers */
        if (unlikely(rkb->rkb_retrybufs.rkbq_cnt) > 0)
            rd_kafka_broker_retry_bufs_move(rkb);

        rd_kafka_broker_unlock(rkb);

        //开始在网络上发送数据
                rd_kafka_broker_io_serve(rkb);

        /* Scan wait-response queue
         * Note: 'now' may be a bit outdated by now. */
        if (do_timeout_scan)
            rd_kafka_broker_waitresp_timeout_scan(rkb, now);

        rd_kafka_broker_lock(rkb);
    }

    rd_kafka_broker_unlock(rkb);
} 

通过poll处理网络事件，将消息从网络发送到broker

static void rd_kafka_broker_io_serve (rd_kafka_broker_t *rkb) {
    rd_kafka_op_t *rko;
    rd_ts_t now = rd_clock();
    //处理broker操作
    if (unlikely(rd_kafka_q_len(&rkb->rkb_ops) > 0))
        while ((rko = rd_kafka_q_pop(&rkb->rkb_ops, RD_POLL_NOWAIT)))
            rd_kafka_broker_op_serve(rkb, rko);
    //请求metadata
    if (unlikely(now >= rkb->rkb_ts_metadata_poll))
        rd_kafka_broker_metadata_req(rkb, 1 /* all topics */, NULL,
                                             NULL, "periodic refresh");
    //如果有消息，手动增加写事件
    if (rkb->rkb_outbufs.rkbq_cnt > 0)
        rkb->rkb_pfd.events |= POLLOUT;
    else
        rkb->rkb_pfd.events &= ~POLLOUT;
    if (poll(&rkb->rkb_pfd, 1,
         rkb->rkb_rk->rk_conf.buffering_max_ms) <= 0)
        return;
    //poll函数，处理各种事件，发送消息时，只处理写事件，当请求metadata时，处理读事件
    if (rkb->rkb_pfd.revents & POLLIN)
        while (rd_kafka_recv(rkb) > 0)
            ;
    if (rkb->rkb_pfd.revents & POLLHUP)
        return rd_kafka_broker_fail(rkb, RD_KAFKA_RESP_ERR__TRANSPORT,
                        "Connection closed");
    if (rkb->rkb_pfd.revents & POLLOUT)
        while (rd_kafka_send(rkb) > 0)
            ;
}

问题

librdkafka不像java客户端那样，可以通过future.get()实现同步发送。所以，如果broker不能连通的话，send方法还是可以正常将消息放入队列。这会导致两个问题

1、我们的客户端是不会知道broker已经挂掉了，因而不能对这种情况作出及时处理，导致消息全部堆积在内存中，如果此时不幸，我们的客户端也挂掉了，那这部分消息就全部丢失了。

2、如果broker一直没有恢复，而我们一直向队列中写数据的话，producer中有一个选项message.timeout.ms，如果超过了设定的消息超时时间，那么会有线程清理队列中的数据，导致消息丢失，而如果将时间设置为0（永不超时）的话，将导致客户端内存撑满。

上面这个问题可以通过如下方法实现的同步发送来解决

void dr_cb (...err, , void *msg_opaque) {
     int *produce_statusp = (int *)msg_opaque;

     /* set sync_produce()'s produce_status value to the error code (which can be NO_ERROR) */
     *produce_statusp = err;
}

int sync_produce (rkt, msg..) {
   int produce_status = -100000; /* or some other magic value that is not proper value in rd_kafka_resp_err_t */

   rd_kafka_produce(rkt, ..msg, .., &produce_status /* msg_opaque */);

   do {
     /* poll dr and error callbacks. */
     rd_kafka_poll(rk, 1000);
    /* wait for dr_cb to be called and setting produce_status to the error value. */
   } while (produce_status == -100000);

  if (produce_status == RD_KAFKA_RESP_ERR_NO_ERROR)
   return SUCCESS!;
  else
   return FAILURE;
}