Kafka C++客户端库librdkafka笔记

系统管理员 2022-05-26 13:58 512阅读 0赞

# 目录 #

目录 1

1. 前言 2

2. 缩略语 2

3. 配置和主题 3

3.1. 配置和主题结构 3

3.1.1. Conf 3

3.1.2. ConfImpl 3

3.1.3. Topic 3

3.1.4. TopicImpl 3

4. 线程 4

5. 消费者 5

5.1. 消费者结构 5

5.1.1. Handle 5

5.1.2. HandleImpl 5

5.1.3. ConsumeCb 6

5.1.4. EventCb 6

5.1.5. Consumer 7

5.1.6. KafkaConsumer 7

5.1.7. KafkaConsumerImpl 7

5.1.8. rd\_kafka\_message\_t 7

5.1.9. rd\_kafka\_msg\_s 7

5.1.10. rd\_kafka\_msgq\_t 8

5.1.11. rd\_kafka\_toppar\_t 8

6. 生产者 10

6.1. 生产者结构 10

6.1.1. DeliveryReportCb 11

6.1.2. PartitionerCb 11

6.1.3. Producer 11

6.1.4. ProduceImpl 11

6.2. 生产者启动过程1 11

6.3. 生产者启动过程2 12

6.4. 生产者生产过程 14

7. poll过程 15

# 1. 前言 #

librdkafka提供的异步的生产接口，异步的消费接口和同步的消息接口，没有同步的生产接口。

# 2. 缩略语 #

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;"> 缩略语 </td> 
 <td style="vertical-align:top;"> 缩略语全称 </td> 
 <td style="vertical-align:top;"> 示例或说明 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> rd </td> 
 <td style="vertical-align:top;"> Rapid&nbsp;Development </td> 
 <td style="vertical-align:top;"> rd.h </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> rk </td> 
 <td style="vertical-align:top;"> RdKafka </td> 
 <td style="vertical-align:top;"> &nbsp; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> toppar </td> 
 <td style="vertical-align:top;"> Topic&nbsp;Partition </td> 
 <td style="vertical-align:top;"> struct&nbsp;rd_kafka_toppar_t { }; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> rep </td> 
 <td style="vertical-align:top;"> Reply， </td> 
 <td style="vertical-align:top;"> struct&nbsp;rd_kafka_t&nbsp;{ &nbsp;&nbsp;rd_kafka_q_t&nbsp;*rk_rep }; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> msgq </td> 
 <td style="vertical-align:top;"> Message&nbsp;Queue </td> 
 <td style="vertical-align:top;"> struct&nbsp;rd_kafka_msgq_t&nbsp;{ }; </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> rkb </td> 
 <td style="vertical-align:top;"> RdKafka&nbsp;Broker </td> 
 <td style="vertical-align:top;"> Kafka代理 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> rko </td> 
 <td style="vertical-align:top;"> RdKafka&nbsp;Operation </td> 
 <td style="vertical-align:top;"> Kafka操作 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> rkm </td> 
 <td style="vertical-align:top;"> RdKafka&nbsp;Message </td> 
 <td style="vertical-align:top;"> Kafka消息 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> payload </td> 
 <td style="vertical-align:top;"> &nbsp; </td> 
 <td style="vertical-align:top;"> 存在Kafka上的消息（或叫Log) </td> 
 </tr> 
 </tbody> 
</table>

# 3. 配置和主题 #

## 3.1. 配置和主题结构 ##

![20180503191504128][]

### 3.1.1. Conf ###

配置接口，配置分两种：全局的和主题的。

### 3.1.2. ConfImpl ###

配置的实现。

### 3.1.3. Topic ###

主题接口。

### 3.1.4. TopicImpl ###

主题的实现。

# 4. 线程 #

RdKafka编程涉及到三类线程：

1) 应用线程，业务代码的实现

2) Kafka Broker线程rd\_kafka\_broker\_thread\_main，负责与Broker通讯，多个

3) Kafka Handler线程rd\_kafka\_thread\_main，每创建一个consumer或producer即会创建一个Handler线程。

![20180503191513559][]

# 5. 消费者 #

## 5.1. 消费者结构 ##

![20180503191522896][]

### 5.1.1. Handle ###

定义了poll等接口，它的实现者为HandleImpl。

### 5.1.2. HandleImpl ###

实现了消费者和生产者均使用的poll等，其中poll的作用为：

1) 为生产者回调消息发送结果；

2) 为生产者和消费者回调事件。

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

### 5.1.3. ConsumeCb ###

只针对消费者的Callback。

### 5.1.4. RebalanceCb ###

只针对消费者的Callback。

### 5.1.5. EventCb ###

消费者和生产者均可设置EventCb，如：\_global\_conf->set("event\_cb", &\_event\_cb, errmsg);。

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

### 5.1.6. Consumer ###

简单消息者，一般不使用，而是使用KafkaConsumer。

### 5.1.7. KafkaConsumer ###

消费者和生产者均采用多重继承方式，其中KafkaConsumer为消费者接口，KafkaConsumerImpl为消费者实现。

### 5.1.8. KafkaConsumerImpl ###

KafkaConsumerImpl为消费者实现。

### 5.1.9. rd\_kafka\_message\_t ###

消息结构。

### 5.1.10. rd\_kafka\_msg\_s ###

消息结构，但消息数据实际存储在rd\_kafka\_message\_t，结构大致如下：

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;"> struct&nbsp;rd_kafka_msg_s { &nbsp;&nbsp;rd_kafka_message_t&nbsp;rkm_rkmessage; &nbsp;&nbsp;struct &nbsp;&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_msg_s*&nbsp;tqe_next; &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_msg_s**&nbsp;tqe_prev; &nbsp;&nbsp;&nbsp;&nbsp;int64_t&nbsp;rkm_timestamp; &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_timestamp_type_t&nbsp;rkm_tstype; &nbsp;&nbsp;}rkm_link; }; </td> 
 </tr> 
 </tbody> 
</table>

### 5.1.11. rd\_kafka\_msgq\_t ###

存储消息的消息队列，生产者生产的消息并不直接socket发送到brokers，而是放入了这个队列，结构大致如下：

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;"> struct&nbsp;rd_kafka_msgq_t { &nbsp;&nbsp;struct &nbsp;&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_msg_s*&nbsp;tqh_first;&nbsp;//&nbsp;队首 &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_msg_s*&nbsp;tqh_last;&nbsp;&nbsp;//&nbsp;队尾 &nbsp;&nbsp;}; &nbsp;&nbsp; &nbsp;&nbsp;//&nbsp;消息个数 &nbsp;&nbsp;rd_atomic32_t&nbsp;rkmq_msg_cnt; &nbsp;&nbsp;//&nbsp;所有消息加起来的字节数 &nbsp;&nbsp;rd_atomic64_t&nbsp;rkmq_msg_bytes; }; </td> 
 </tr> 
 </tbody> 
</table>

### 5.1.12. rd\_kafka\_toppar\_t ###

Topic-Partition队列，很复杂的一个结构，部分内容如下：

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;"> //&nbsp;Topic&nbsp;+&nbsp;Partition&nbsp;combination typedef&nbsp;struct&nbsp;rd_kafka_toppar_s { &nbsp;&nbsp;struct &nbsp;&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s*&nbsp;tqe_next; &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s**&nbsp;tqe_prev; &nbsp;&nbsp;}rktp_rklink; &nbsp; &nbsp;&nbsp;struct &nbsp;&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s*&nbsp;tqe_next; &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s**&nbsp;tqe_prev; &nbsp;&nbsp;}rktp_rkblink; &nbsp;&nbsp; &nbsp;&nbsp;struct &nbsp;&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s*&nbsp;cqe_next; &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s*&nbsp;cqe_prev; &nbsp;&nbsp;}rktp_fetchlink; &nbsp;&nbsp; &nbsp;&nbsp;struct &nbsp;&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s*&nbsp;tqe_next; &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s**&nbsp;tqe_prev; &nbsp;&nbsp;}rktp_rktlink; &nbsp;&nbsp; &nbsp;&nbsp;struct &nbsp;&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s*&nbsp;tqe_next; &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_toppar_s**&nbsp;tqe_prev; &nbsp;&nbsp;}rktp_cgrplink; &nbsp;&nbsp; &nbsp;&nbsp;rd_kafka_itopic_t*&nbsp;rktp_rkt; &nbsp;&nbsp;int32_t&nbsp;rktp_partition; &nbsp;&nbsp;int32_t&nbsp;rktp_leader_id; &nbsp;&nbsp;rd_kafka_broker_t*&nbsp;rktp_leader; &nbsp;&nbsp;rd_kafka_broker_t*&nbsp;rktp_next_leader; &nbsp;&nbsp;rd_refcnt_t&nbsp;rktp_refcnt; &nbsp;&nbsp;rd_kafka_msgq_t&nbsp;rktp_msgq;&nbsp;//&nbsp;application-&gt;rdkafka&nbsp;queue }rd_kafka_toppar_t; </td> 
 </tr> 
 </tbody> 
</table>

# 6. 生产者 #

## 6.1. 生产者结构 ##

![20180503191534114][]

### 6.1.1. DeliveryReportCb ###

消息已经成功递送到Broker时回调，只针对生产者有效。

### 6.1.2. PartitionerCb ###

计算分区号回调函数，只针对生产者有效。

### 6.1.3. Producer ###

Producer为生产者接口，它的实现者为ProducerImpl。

### 6.1.4. ProduceImpl ###

ProducerImpl为生产者的实现。

## 6.2. 生产者启动过程1 ##

启动时会创建两组线程：一组Broker线程（rd\_kafka\_broker\_thread\_main，多个），实为与Broker间的网络IO线程；一组Handler线程（rd\_kafka\_thread\_main，单个），每调用一次RdKafka::Producer::create或rd\_kafka\_new即创建一Handler线程。

![20180503191545836][]  
Handler线程调用栈：

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

## 6.3. 生产者启动过程2 ##

创建网络IO线程，消费者启动过程类似，只是一个调用rd\_kafka\_broker\_producer\_serve(rkb)，另一个调用rd\_kafka\_broker\_consumer\_serve(rkb)。

IO线程负责消息的收和发，发送底层调用的是sendmsg，收调用的是recvmsg（但MSVC平台调用send和recv）。

![20180503191552108][]

## 6.4. 生产者生产过程 ##

![20180503191609558][]

生产者生产的消息并不直接socket发送到brokers，而是放入队列rd\_kafka\_msgq\_t中。Broker线程（rd\_kafka\_broker\_thread\_main）消费这个队列。

Broker线程同时监控与Broker间的网络连接，又要监控队列中是否有数据，如何实现的？这个队列和管道绑定在一起的，绑定的是管道写端（rktp->rktp\_msgq\_wakeup\_fd = rkb->rkb\_toppar\_wakeup\_fd; rkb->rkb\_toppar\_wakeup\_fd=rkb->rkb\_wakeup\_fd\[1\]）。

这样Broker线程即可同时监听网络数据和管道数据。

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;"> //&nbsp;int&nbsp;rd_kafka_msg_partitioner(rd_kafka_itopic_t&nbsp;*rkt,&nbsp;rd_kafka_msg_t&nbsp;*rkm,int&nbsp;do_lock) (gdb)&nbsp;p&nbsp;*rkm $7&nbsp;=&nbsp;{rkm_rkmessage&nbsp;=&nbsp;{err&nbsp;=&nbsp;RD_KAFKA_RESP_ERR_NO_ERROR,&nbsp;rkt&nbsp;=&nbsp;0x1590c10,&nbsp;partition&nbsp;=&nbsp;1,&nbsp;payload&nbsp;=&nbsp;0x7f48c4001260,&nbsp;len&nbsp;=&nbsp;203,&nbsp;key&nbsp;=&nbsp;0x7f48c400132b,&nbsp;key_len&nbsp;=&nbsp;14,&nbsp;offset&nbsp;=&nbsp;0,&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;_private&nbsp;=&nbsp;0x0},&nbsp;rkm_link&nbsp;=&nbsp;{tqe_next&nbsp;=&nbsp;0x5b5d47554245445b,&nbsp;tqe_prev&nbsp;=&nbsp;0x6361667265746e69},&nbsp;rkm_flags&nbsp;=&nbsp;196610,&nbsp;rkm_timestamp&nbsp;=&nbsp;1524829399009,&nbsp; &nbsp;&nbsp;rkm_tstype&nbsp;=&nbsp;RD_KAFKA_TIMESTAMP_CREATE_TIME,&nbsp;rkm_u&nbsp;=&nbsp;{producer&nbsp;=&nbsp;{ts_timeout&nbsp;=&nbsp;16074575505526,&nbsp;ts_enq&nbsp;=&nbsp;16074275505526}}} (gdb)&nbsp;p&nbsp;rkm-&gt;rkm_rkmessage $8&nbsp;=&nbsp;{err&nbsp;=&nbsp;RD_KAFKA_RESP_ERR_NO_ERROR,&nbsp;rkt&nbsp;=&nbsp;0x1590c10,&nbsp;partition&nbsp;=&nbsp;1,&nbsp;payload&nbsp;=&nbsp;0x7f48c4001260,&nbsp;len&nbsp;=&nbsp;203,&nbsp;key&nbsp;=&nbsp;0x7f48c400132b,&nbsp;key_len&nbsp;=&nbsp;14,&nbsp;offset&nbsp;=&nbsp;0,&nbsp;_private&nbsp;=&nbsp;0x0} (gdb)&nbsp;p&nbsp;rkm-&gt;rkm_rkmessage-&gt;payload $9&nbsp;=&nbsp;(void&nbsp;*)&nbsp;0x7f48c4001260 (gdb)&nbsp;p&nbsp;(char*)rkm-&gt;rkm_rkmessage-&gt;payload $10&nbsp;=&nbsp;0x7f48c4001260&nbsp;"{\"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"... </td> 
 </tr> 
 </tbody> 
</table>

# 7. poll过程 #

poll的作用是触发回调，生产者即使不调用poll，消息也会发送出去，但是如果不通过poll触发回调，则不能确定消息发送状态（成功或失败等）。

消费队列rd\_kafka\_t->rk\_rep，rk\_rep为响应队列，类型为rd\_kafka\_q\_t或rd\_kafka\_q\_s：

![20180503191602295][]

相关源代码摘要：

<table style="width:426.1pt;"> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:426.1pt;"> //&nbsp;rdkafkacpp_int.h class&nbsp;HandleImpl&nbsp;:&nbsp;virtual&nbsp;public&nbsp;Handle&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;int&nbsp;poll&nbsp;(int&nbsp;timeout_ms)&nbsp;{&nbsp;return&nbsp;rd_kafka_poll(rk_,&nbsp;timeout_ms);&nbsp;}; }; &nbsp; //&nbsp;rdkafka.c int&nbsp;rd_kafka_poll&nbsp;(rd_kafka_t&nbsp;*rk,&nbsp;int&nbsp;timeout_ms)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;rd_kafka_q_serve(rk-&gt;rk_rep,&nbsp;timeout_ms,&nbsp;0, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;RD_KAFKA_Q_CB_CALLBACK,&nbsp;rd_kafka_poll_cb,&nbsp;NULL); } &nbsp; /** &nbsp;*&nbsp;rdkafka_queue.c&nbsp; &nbsp;* &nbsp;*&nbsp;Pop&nbsp;all&nbsp;available&nbsp;ops&nbsp;from&nbsp;a&nbsp;queue&nbsp;and&nbsp;call&nbsp;the&nbsp;provided&nbsp; &nbsp;*&nbsp;callback&nbsp;for&nbsp;each&nbsp;op. &nbsp;*&nbsp;`max_cnt`&nbsp;limits&nbsp;the&nbsp;number&nbsp;of&nbsp;ops&nbsp;served,&nbsp;0&nbsp;=&nbsp;no&nbsp;limit. &nbsp;* &nbsp;*&nbsp;Returns&nbsp;the&nbsp;number&nbsp;of&nbsp;ops&nbsp;served. &nbsp;* &nbsp;*&nbsp;Locality:&nbsp;any&nbsp;thread. &nbsp;*/ int&nbsp;rd_kafka_q_serve&nbsp;(rd_kafka_q_t&nbsp;*rkq,&nbsp;int&nbsp;timeout_ms, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;int&nbsp;max_cnt,&nbsp;rd_kafka_q_cb_type_t&nbsp;cb_type, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_q_serve_cb_t&nbsp;*callback,&nbsp;void&nbsp;*opaque)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 &nbsp;&nbsp;&nbsp;&nbsp;mtx_lock(&amp;rkq-&gt;rkq_lock); &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 &nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;从列队取数据，超时时长为timeout_ms &nbsp;&nbsp;&nbsp;&nbsp;while&nbsp;(!(rko&nbsp;=&nbsp;TAILQ_FIRST(&amp;rkq-&gt;rkq_q))&nbsp;&amp;&amp;&nbsp;timeout_ms&nbsp;!=&nbsp;0)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(cnd_timedwait_ms(&amp;rkq-&gt;rkq_cond, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&amp;rkq-&gt;rkq_lock,&nbsp;timeout_ms)&nbsp;!=&nbsp;thrd_success) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;break; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;timeout_ms&nbsp;=&nbsp;0; &nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(!rko)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;mtx_unlock(&amp;rkq-&gt;rkq_lock); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;0; &nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 } </td> 
 </tr> 
 </tbody> 
</table>

# **1. consume过程** #

<table style="width:426.1pt;"> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;width:426.1pt;"> //&nbsp;ConsumerImpl.cpp class&nbsp;ConsumerImpl&nbsp;:&nbsp;virtual&nbsp;public&nbsp;Consumer,&nbsp;virtual&nbsp;public&nbsp;HandleImpl&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;Message&nbsp;*consume&nbsp;(Topic&nbsp;*topic,&nbsp;int32_t&nbsp;partition,&nbsp;int&nbsp;timeout_ms)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rkmessage&nbsp;=&nbsp;rd_kafka_consume(topicimpl-&gt;rkt_,&nbsp;partition,&nbsp;timeout_ms); &nbsp;&nbsp;&nbsp;&nbsp;} }; &nbsp; //&nbsp;rdkafka.c rd_kafka_message_t&nbsp;*rd_kafka_consume&nbsp;( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_topic_t&nbsp;*app_rkt, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;int32_t&nbsp;partition, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;int&nbsp;timeout_ms)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 &nbsp;&nbsp;&nbsp;&nbsp;rkmessage&nbsp;=&nbsp;rd_kafka_consume0( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rkt-&gt;rkt_rk, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rktp-&gt;rktp_fetchq, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;timeout_ms); &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 } &nbsp; /** &nbsp;*&nbsp;rdkafka.c &nbsp;* &nbsp;*&nbsp;Serve&nbsp;queue&nbsp;'rkq'&nbsp;and&nbsp;return&nbsp;one&nbsp;message. &nbsp;*&nbsp;By&nbsp;serving&nbsp;the&nbsp;queue&nbsp;it&nbsp;will&nbsp;also&nbsp;call&nbsp;any&nbsp;registered&nbsp;callbacks &nbsp;*&nbsp;registered&nbsp;for&nbsp;matching&nbsp;events,&nbsp;this&nbsp;includes&nbsp;consumer_cb() &nbsp;*&nbsp;in&nbsp;which&nbsp;case&nbsp;no&nbsp;message&nbsp;will&nbsp;be&nbsp;returned. &nbsp;*/ static&nbsp;rd_kafka_message_t&nbsp;*rd_kafka_consume0&nbsp;( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_t&nbsp;*rk, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_q_t&nbsp;*rkq, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;int&nbsp;timeout_ms)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 &nbsp;&nbsp;&nbsp;&nbsp;rd_ts_t&nbsp;abs_timeout&nbsp;=&nbsp;rd_timeout_init(timeout_ms); &nbsp;&nbsp;&nbsp;&nbsp;while&nbsp;((rko&nbsp;=&nbsp;rd_kafka_q_pop( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rkq,&nbsp;rd_timeout_remains(abs_timeout),&nbsp;0)))&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 &nbsp;&nbsp;&nbsp;&nbsp;/*&nbsp;Get&nbsp;rkmessage&nbsp;from&nbsp;rko&nbsp;*/ &nbsp;&nbsp;&nbsp;&nbsp;rkmessage&nbsp;=&nbsp;rd_kafka_message_get(rko); &nbsp;&nbsp;&nbsp;&nbsp;/*&nbsp;Store&nbsp;offset&nbsp;*/ &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_op_offset_store(rk,&nbsp;rko,&nbsp;rkmessage); &nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_set_last_error(0,&nbsp;0); &nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;rkmessage; } &nbsp; //&nbsp;rdkafka_queue.c rd_kafka_op_t&nbsp;*rd_kafka_q_pop&nbsp;( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_q_t&nbsp;*rkq, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;int&nbsp;timeout_ms, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;int32_t&nbsp;version)&nbsp;{ return&nbsp;rd_kafka_q_pop_serve( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rkq,&nbsp;timeout_ms,&nbsp;version, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;RD_KAFKA_Q_CB_RETURN, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;NULL,&nbsp;NULL); } &nbsp; /** &nbsp;*&nbsp;rdkafka_queue.c &nbsp;* &nbsp;*&nbsp;Serve&nbsp;q&nbsp;like&nbsp;rd_kafka_q_serve()&nbsp;until&nbsp;an&nbsp;op&nbsp;is&nbsp;found&nbsp;that&nbsp;can&nbsp;be&nbsp;returned &nbsp;*&nbsp;as&nbsp;an&nbsp;event&nbsp;to&nbsp;the&nbsp;application. &nbsp;* &nbsp;*&nbsp;@returns&nbsp;the&nbsp;first&nbsp;event:able&nbsp;op,&nbsp;or&nbsp;NULL&nbsp;on&nbsp;timeout. &nbsp;* &nbsp;*&nbsp;Locality:&nbsp;any&nbsp;thread &nbsp;*/ rd_kafka_op_t&nbsp;*rd_kafka_q_pop_serve&nbsp;( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_q_t&nbsp;*rkq,&nbsp;int&nbsp;timeout_ms, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;int32_t&nbsp;version, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_q_cb_type_t&nbsp;cb_type, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_q_serve_cb_t&nbsp;*callback, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;void&nbsp;*opaque)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 &nbsp;&nbsp;&nbsp;&nbsp;mtx_lock(&amp;rkq-&gt;rkq_lock); &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 &nbsp;&nbsp;&nbsp;&nbsp;do&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;retry: &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;while&nbsp;((rko&nbsp;=&nbsp;TAILQ_FIRST(&amp;rkq-&gt;rkq_q))&nbsp;&amp;&amp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;!(rko&nbsp;=&nbsp;rd_kafka_op_filter(rkq,&nbsp;rko,&nbsp;version))) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(rko)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;/*&nbsp;Proper&nbsp;versioned&nbsp;op&nbsp;*/ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rd_kafka_q_deq0(rkq,&nbsp;rko); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;/*&nbsp;Ops&nbsp;with&nbsp;callbacks&nbsp;are&nbsp;considered&nbsp;handled &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;*&nbsp;and&nbsp;we&nbsp;move&nbsp;on&nbsp;to&nbsp;the&nbsp;next&nbsp;op,&nbsp;if&nbsp;any. &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;*&nbsp;Ops&nbsp;w/o&nbsp;callbacks&nbsp;are&nbsp;returned&nbsp;immediately&nbsp;*/ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(res&nbsp;==&nbsp;RD_KAFKA_OP_RES_HANDLED) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;goto&nbsp;retry;&nbsp;/*&nbsp;Next&nbsp;op&nbsp;*/ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;else&nbsp;if&nbsp;(unlikely(res&nbsp;==&nbsp;RD_KAFKA_OP_RES_YIELD))&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;/*&nbsp;Callback&nbsp;yielded,&nbsp;unroll&nbsp;*/ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;mtx_unlock(&amp;rkq-&gt;rkq_lock); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;NULL; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;res&nbsp;=&nbsp;rd_kafka_op_handle( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;rkq-&gt;rkq_rk,&nbsp;rkq,&nbsp;rko, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;cb_type,&nbsp;opaque,&nbsp;callback); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;/*&nbsp;No&nbsp;op,&nbsp;wait&nbsp;for&nbsp;one&nbsp;*/ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;pre&nbsp;=&nbsp;rd_clock(); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(cnd_timedwait_ms( &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&amp;rkq-&gt;rkq_cond, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&amp;rkq-&gt;rkq_lock, &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;timeout_ms)&nbsp;==&nbsp;thrd_timedout)&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;mtx_unlock(&amp;rkq-&gt;rkq_lock); &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;NULL; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;/*&nbsp;Remove&nbsp;spent&nbsp;time&nbsp;*/ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;timeout_ms&nbsp;-=&nbsp;(int)&nbsp;(rd_clock()-pre)&nbsp;/&nbsp;1000; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(timeout_ms&nbsp;&lt;&nbsp;0) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;timeout_ms&nbsp;=&nbsp;RD_POLL_NOWAIT; &nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;while&nbsp;(timeout_ms&nbsp;!=&nbsp;RD_POLL_NOWAIT); &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 &nbsp;&nbsp;&nbsp;&nbsp;。。。。。。 } </td> 
 </tr> 
 </tbody> 
</table>

[20180503191504128]: /images/20220526/5ee18680cf89494f8ba09f0b54157f3e.png
[20180503191513559]: /images/20220526/a5eef4b6b6034b37b9e99e3fba8ced40.png
[20180503191522896]: /images/20220526/9f428b02c1b54527a8a506d9b22538df.png
[20180503191534114]: /images/20220526/e049a17b4c954473a33bdee4e3f7d68b.png
[20180503191545836]: /images/20220526/c0fceae30735430eaab9418c1bd703e1.png
[20180503191552108]: /images/20220526/544c7c243a7e4a2683096b5650fc48e3.png
[20180503191609558]: /images/20220526/22f6a997ec374d6e9bf0954aa748ffc5.png
[20180503191602295]: /images/20220526/e1cda9e930f443739cb99c6c3cf81d86.png