Kafka C++客户端库librdkafka笔记

缩略语

缩略语全称

示例或说明

class Handle {   /**    * @brief Polls the provided kafka handle for events.    *    * Events will trigger application provided callbacks to be called.    *    * The \p timeout_ms argument specifies the maximum amount of time    * (in milliseconds) that the call will block waiting for events.    * For non-blocking calls, provide 0 as \p timeout_ms.    * To wait indefinately for events, provide -1.    *    * Events:    *   - delivery report callbacks (if an RdKafka::DeliveryCb is configured) [producer]    *   - event callbacks (if an RdKafka::EventCb is configured) [producer & consumer]    *    * @remark  An application should make sure to call poll() at regular    *          intervals to serve any queued callbacks waiting to be called.    *    * @warning This method MUST NOT be used with the RdKafka::KafkaConsumer,    *          use its RdKafka::KafkaConsumer::consume() instead.    *    * @returns the number of events served.    */   virtual int poll(int timeout_ms) = 0; };

/**  * @brief Event callback class  *  * Events are a generic interface for propagating errors, statistics, logs, etc  * from librdkafka to the application.  *  * @sa RdKafka::Event  */ class RD_EXPORT EventCb {  public:   /**    * @brief Event callback    *    * @sa RdKafka::Event    */   virtual void event_cb (Event &event) = 0;     virtual ~EventCb() { } };   /**  * @brief Event object class as passed to the EventCb callback.  */ class RD_EXPORT Event {  public:   /** @brief Event type */   enum Type {     EVENT_ERROR,     /**< Event is an error condition */     EVENT_STATS,     /**< Event is a statistics JSON document */     EVENT_LOG,       /**< Event is a log message */     EVENT_THROTTLE   /**< Event is a throttle level signaling from the broker */   }; };

struct rd_kafka_msg_s {   rd_kafka_message_t rkm_rkmessage;   struct   {     rd_kafka_msg_s* tqe_next;     rd_kafka_msg_s** tqe_prev;     int64_t rkm_timestamp;     rd_kafka_timestamp_type_t rkm_tstype;   }rkm_link; };

struct rd_kafka_msgq_t {   struct   {     rd_kafka_msg_s* tqh_first; // 队首     rd_kafka_msg_s* tqh_last;  // 队尾   };      // 消息个数   rd_atomic32_t rkmq_msg_cnt;   // 所有消息加起来的字节数   rd_atomic64_t rkmq_msg_bytes; };

// Topic + Partition combination typedef struct rd_kafka_toppar_s {   struct   {     rd_kafka_toppar_s* tqe_next;     rd_kafka_toppar_s** tqe_prev;   }rktp_rklink;     struct   {     rd_kafka_toppar_s* tqe_next;     rd_kafka_toppar_s** tqe_prev;   }rktp_rkblink;      struct   {     rd_kafka_toppar_s* cqe_next;     rd_kafka_toppar_s* cqe_prev;   }rktp_fetchlink;      struct   {     rd_kafka_toppar_s* tqe_next;     rd_kafka_toppar_s** tqe_prev;   }rktp_rktlink;      struct   {     rd_kafka_toppar_s* tqe_next;     rd_kafka_toppar_s** tqe_prev;   }rktp_cgrplink;      rd_kafka_itopic_t* rktp_rkt;   int32_t rktp_partition;   int32_t rktp_leader_id;   rd_kafka_broker_t* rktp_leader;   rd_kafka_broker_t* rktp_next_leader;   rd_refcnt_t rktp_refcnt;   rd_kafka_msgq_t rktp_msgq; // application->rdkafka queue }rd_kafka_toppar_t;

(gdb) t 17 [Switching to thread 17 (Thread 0x7ff7059d3700 (LWP 16765))] #0  0x00007ff7091e6cf2 in pthread_cond_timedwait@@GLIBC_2.3.2 () from /lib64/libpthread.so.0 (gdb) bt #0  0x00007ff7091e6cf2 in pthread_cond_timedwait@@GLIBC_2.3.2 () from /lib64/libpthread.so.0 #1  0x00000000005b4d2f in cnd_timedwait_ms (cnd=0x1517748, mtx=0x1517720, timeout_ms=898) at tinycthread.c:501 #2  0x0000000000580e16 in rd_kafka_q_serve (rkq=0x1517720, timeout_ms=898, max_cnt=0, cb_type=RD_KAFKA_Q_CB_CALLBACK, callback=0x0, opaque=0x0) at rdkafka_queue.c:440 #3  0x000000000054ee9b in rd_kafka_thread_main (arg=0x1516df0) at rdkafka.c:1227 #4  0x00000000005b4e0f in _thrd_wrapper_function (aArg=0x15179d0) at tinycthread.c:624 #5  0x00007ff7091e2e25 in start_thread () from /lib64/libpthread.so.0 #6  0x00007ff7082d135d in clone () from /lib64/libc.so.6

// int rd_kafka_msg_partitioner(rd_kafka_itopic_t *rkt, rd_kafka_msg_t *rkm,int do_lock) (gdb) p *rkm $7 = {rkm_rkmessage = {err = RD_KAFKA_RESP_ERR_NO_ERROR, rkt = 0x1590c10, partition = 1, payload = 0x7f48c4001260, len = 203, key = 0x7f48c400132b, key_len = 14, offset = 0,      _private = 0x0}, rkm_link = {tqe_next = 0x5b5d47554245445b, tqe_prev = 0x6361667265746e69}, rkm_flags = 196610, rkm_timestamp = 1524829399009,    rkm_tstype = RD_KAFKA_TIMESTAMP_CREATE_TIME, rkm_u = {producer = {ts_timeout = 16074575505526, ts_enq = 16074275505526}}} (gdb) p rkm->rkm_rkmessage $8 = {err = RD_KAFKA_RESP_ERR_NO_ERROR, rkt = 0x1590c10, partition = 1, payload = 0x7f48c4001260, len = 203, key = 0x7f48c400132b, key_len = 14, offset = 0, _private = 0x0} (gdb) p rkm->rkm_rkmessage->payload $9 = (void *) 0x7f48c4001260 (gdb) p (char*)rkm->rkm_rkmessage->payload $10 = 0x7f48c4001260 "{\"p\":\"f\",\"o\":1,\"d\":\"m\",\"d\":\"m\",\"i\":\"f2\",\"ip\":\"127.0.0.1\",\"pt\":2018,\"sc\":0,\"fc\":1,\"tc\":0,\"acc\":395,\"mcc\":395,\"cd\":\"test\",\"cmd\":\"tester\",\"cf\":\"main\",\"cp\":\"1.49.16.9"...