使用librdkafka开发一个producer的步骤：
librdkafka:
1. conf 设置
kafka conf:
rd_kafka_conf_new(): rd_kafka_conf_set()
topic conf:
rd_kafka_topic_conf_new(): rd_kafka_topic_conf_set()

1. 设置conf回调，消息发送成功或者失败都会调用
   rd_kafka_conf_set_dr_cb()
   rd_kafka_conf_set_dr_msg_cb()

2. 创建kafka
   rd_kafka_new()
   设置系统日志
   rd_kafka_set_logger()
   rd_kafka_set_log_level()
   添加下游brokers:
   rd_kafka_brokers_add()

3. 创建新的topic
   rd_kafka_topic_new()

4. producer:
   rd_kafka_produce()
   发送后，设置时间观察，第二个参数是阻塞等待时间，一般设置为0，rd_kafka_poll()

5. 销毁操作
   rd_kafka_topic_destroy()
   rd_kafka_destroy()
   rd_kafka_wait_destroyed(2000)

一些数据结构的解释

• Brokers
  librdkafka 只需要一份最初的brokers列表（至少包含一个broker）。它将连接所有”metadata.broker.list”或者是rd_kafka_brokers_add()函数添加的brokers，然后向每个brokers申请一些元数据信息：包含brokers的完整列表、topic、partitions以及它们在Kafka 集群中的leaders broker信息。

Brokers名字的形式为：host：port； 其中port是可选的，默认是9092，host是任何一个可以解析的hostname或者ipv4或者ipv6地址。如果host是多个地址，librdkafka将会在每一次连接尝试中循环连接这些地址。包含所有broker 地址的DNS记录可以用来提供可靠的bootstrap broker。

• rd_kafka_t
  实际应用中，需要创建一个top-level的对象 rd_kafka_t， 这个对象是基本的容器，它提供了全局性配置属性以及共享状态信息，它由rd_kafka_new()函数创建。

• rd_kafka_topic_t
  同时也需要创建一个或者多个topics对象rd_kafka_topic_t，给produer以及consumer使用。 topic对象具有topic特定的配置属性，同时还包含了所有可用partitions与leader brokers映射关系。它通过调用rd_kafka_topic_new()函数创建。

  注意：实际应用中，可能会创建多个rd_kafka_t对象，它们并没有共享状态信息
  注意：rd_kafka_topic_t对象只能由创建它的对象rd_kafka_t使用。

• 线程和回调函数

  librdkafka 内部将会有多个线程，以充分利用硬件资源。API的实现是完全线程安全的，实际应用中可以在任何时候任何线程中调用任何API函数而不用担心线程安全。

  一个以轮询为基础的API用来给实际应用提供信号反馈，实际应用应当按照固定时间间隔调用rd_kafka_poll()函数。这个轮询的API将会调用以下可的回调（都是可选的）：

  消息发送报告回调：报告消息发送失败。这将允许实际应用采取措施应对发送失败，并释放消息发送过程中占有的资源。

  错误回调：报告错误；错误一般是信息化方面的，例如连接broker失败，实际应用通常不需要采取任何措施。错误的数据类型是通过rd_kafka_resp_err_t enum类型数据，可以描述本地错误和远程broker错误。

  不是poll函数引起的可选回调函数，可能是由任意线程引发的：

  logging 回调：实际应用中，用于发送librdkafka产生的log消息。

  partitioner 回调：实际应用提供消息的partitioner。partitioner可能被任何线程任何时候调用，它可能由于同一个key而被调用多次。Partitioner 函数有以下限制：

  一定不能调用rd_kafka_*()等函数

  一定不能阻塞或延长执行

  一定要返回一个0到partition_cnt-1之间的值，或者是在partitioning不能执行的时候返回特定RD_KAFKA_PARTITION_UA值。
  -rd_kafka_message_t对象成员：
  err：错误返回值。非0值表示出现错误，err是rd_kafka_resp_err_t类型数据。如果是0则表示进行了适当的消息抓取，并且payload中包含了message。
  rkt，partition：topic和partition信息
  payload，len：消息的payload数据或者错误的消息（err！=0）
  key，key_len：可选参数，主要是用来获取特定的消息。
  offset：消息的偏移地址

一些函数

* rd_kafka_consume_start()函数的参数：

rkt： 进行consume的topic， 由前面rd_kafka_topic_new()创建

partition：进行consume的partition

offset：开始consume的消息偏移。这个偏移可能是一个绝对消息偏移，或者是RD_KAKFA_OFFSET_STORED来使用存储的offset，也可能是两个特定偏移之一：RD_KAFKA_OFFSET_BEGINNING，从partition消息队列的开始进行consume；RD_KAFKA_OFFSET_END：从partition中的将要produce的下一条信息开始（忽略即当前所有的消息）。


在topic+partition的consumer启动之后，librdkafka将尝试使本地消息队列中的消息数目保持在queued.min.messages，一方反复的从broker获取消息。


本地消息队列将通过以下三种不同的consum  APIs进行consume：

rd_kafka_consume()：每次consume一条消息

rd_kafka_consume_batch()：批处理consume，一条或多条

rd_kafka_consume_callback()：consume本地消息队列中的所有消息，并调用回调函数处理每条消息


上述三种方式按照性能排列的，rd_kafka_consume()是最慢的，rd_kafka_consume_callback()最快。不同的需求可以选择不同的实现方式。


一条consumed消息，由每一个consume函数提供或返回，具体是由rd_kafka_messag_t类型对象保存。</big>

sparkstreaming 部分

（编辑：安卓应用网_ASP源码网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!