Producer API：应用程序（生产者）基于此API将数据发送到Kafka的Topic。
Consumer API：应用程序（消费者）基于此API从Kafka的Topic获取数据进行消费。
Connection API：用于持续的从一些源系统输入数据到Kafka，或者从Kafka持续推送数据导目标系统，比如Hadoop、数据库等。
Streams API：用于从源Topic转化到目标Topic的数据流转换，作用跟Spark、Storm和Flink相似。
Admin API：用于管理Kafka，包括Broker、Topic等元数据管理，和上面介绍的Kafka自带脚本功能类似。Kafka自身没有提供可视化界面，基于Admin API可以搭建自己的管理控台。

PT2 PRODUCER API

Pt2.1 Producer参数

Kafka Producer支持以下参数配置。如果对配置还不是特别了解，建议先看一下后面的代码示例，有个初步的认识再了解每个配置的含义。

bootstrap.servers

该参数指定了一组host:port对，用于创建同Kafka broker服务器的连接，比如:kl:9092，k2:9092，k3:9092。

如果Kafka集群中机器数很多，那么只需要指定部分broker即可，不需要列出所有的机器。因为不管指定几台机器，producer都会通过该参数找到并发现集群中所有的broker；为该参数指定多台机器只是为了故障转移使用。这样即使某一台broker挂掉了，producer重启后依然可以通过该参数指定的其他broker连入Kafka集群。

另外，如果broker端没有显式配置listeners使用IP地址，那么最好将该参数也配置成主机名，而不是IP地址。因为Kafka内部使用的就是FQDN(Fully Qualified Domain Name，全限定域名）。

 # 以下配置可以在kafka目录下找到：/opt/kafka_2.13-2.7.0/config/server.properties
 # The address the socket server listens on. It will get the value returned from 
 # java.net.InetAddress.getCanonicalHostName() if not configured.
 #   FORMAT:
 #     listeners = listener_name://host_name:port
 #   EXAMPLE:
 #     listeners = PLAINTEXT://your.host.name:9092
 listeners=PLAINTEXT://0.0.0.0:9092

key.serializer

被发送到broker端的任何消息的格式都必须是字节数组，因此消息的各个组件必须先做序列化，才能发送到broker。该参数就是为消息的key做序列化之用的。示例参数指定的是实现 org.apache.kafka.common.serialization.Serializer 接口的类的全限定名称。

Kafka为大部分的初始类型（primitive type）默认提供了现成的序列化器。producer程序在发送消息时不指定key，key.serializer参数也是必须要设置的；否则程序会抛出ConfigException异常，提示“key.serializer”参数无默认值，必须要配置。

value.serialize

和key.serializer类似，只是它被用来对消息体（即消息value）部分做序列化，将消息value部分转换成字节数组。value.serializer 和 key. serializer 可以设置相同的值，也可以不同的值。只要消费端，消费数据的时候，保持一致就可以了。

需要注意的是，这两个参数都必须是全限定类名，如org.apache.kafka.common.serialization.Serializer。

acks

acks作用

acks 参数用于控制 producer 生产消息的持久性（durability）。对于 producer 而言， Kafka在乎的是“己提交”消息的持久性。一旦消息被成功提交，那么只要有任何一个保存了该消息的副本“存活”，这条消息就会被视为“不会丢失的” 。经常碰到抱怨Kafka的producer会丢消息，其实这里混淆了一个概念，即那些所谓的“己丢失”的消息其实并没有被成功写入 Kafka 。换句话说，它们井没有被成功提交，因此 Kafka 对这些消息的持久性不做任何保障。

Producer API 提供了回调机制供用户处理发送失败的情况。具体来说，当 producer 发送一条消息给 Kafka 集群时，这条消息会被发送到指定 topic 分区 leader 所在的 broker 上，producer 等待从该 leader broker 返回消息的写入结果（当然并不是无限等待，是有超时时间的）以确定消息被成功提交。这一切完成后 producer 可以继续发送新的消息。

Kafka 能够保证的是 consumer 永远不会读取到尚未提交完成的消息。所以 leader broker 何时发送写入结果返还给 producer 就非常关键，也会直接影响消息的持久性甚至是 producer 端的吞吐量（producer越快地接收到 leader broker响应，就能发送下一条消息），producer 端的 acks 参数就是用来控制做这件事情的。

acks配置

acks 指定了在给 producer 发送响应前， leader broker 必须要确保己成功写入该消息的副本数。当前 acks 有 3 个取值： 0、 1和 all 。

acks = 0：设置成 0 表示 producer 完全不理睬 leader broker 端的处理结果。此时，producer 发送消息后立即开启下一条消息的发送，根本不等待 leader broker 端返回结果。由于不接收发送结果，因此在这种情况下 producer.send 的回调也就完全失去了作用，即用户无法通过回调机制感知任何发送过程中的失败，所以 acks=0 时 producer 并不保证消息会被成功发送。

但凡事有利就有弊，由于不需要等待响应结果，通常这种设置下 producer 的吞吐量是最高的。

acks = all 或者-1：表示当发送消息时， leader broker 不仅会将消息写入本地日志（持久化），同时还会等待 ISR (in-sync replica set)中所有其他副本都成功写入它们各自的本地日志后，才发送响应结果给 producer。

显然当设置 acks=all 时，只要 ISR中至少有一个副本是处于“存活”状态的，那么这条消息就肯定不会丢失，因而可以达到最高的消息持久性，但通常这种设置下producer 的吞吐量也是最低的。

acks = 1 ：是 0 和 all 折中的方案，也是默认的参数值。producer发送消息后，leader broker仅将该消息写入本地日志，然后发送响应结果给 producer，而无须等待 ISR中其他副本写入该消息。那么此时只要该 leader broker 一直存活， Kafka 就能够保证这条消息不丢失。

这实际上是一种折中方案，既可以达到适当的消息持久性，同时也保证了 producer 端的吞吐量。

总结一下：acks 参数控制 producer 发送消息实现不同程度的消息持久性，它有 3 个取值，对应的优缺点以使用场景如表所示。

buffer.memory

该参数指定了 producer 端用于缓存消息的缓冲区大小，单位是字节，默认值是 33554432，即 32MB 。

如前所述，由于采用了异步发送消息的设计架构， Java 版本 producer 启动时会首先创建一块内存缓冲区用于保存待发送的消息，由另一个专属线程负责从缓冲区中读取消息执行真正的发送。这部分内存空间的大小即是由 buffer.memory 参数指定的。

若 producer 向缓冲区写消息的速度超过了专属 I/0 线程发送消息的速度，那么必然造成该缓冲区空间的不断增大。此时 producer 会停止手头的工作等待 I/0 线程追上来，若一段时间之后 I/0 线程还是无法追上 producer 的进度，就会抛出异常；若 producer 程序要给很多分区发送消息，那么就需要仔细地设置这个参数，以防止过小的内存缓冲区降低了producer 程序整体的吞吐量。

compression.type

设置 producer 端是否压缩消息，默认值是 none ，即不压缩消息。

Kafka 的 producer 端引入压缩后可以显著地降低网络 I/O 传输开销从而提升整体吞吐量，但也会增加 producer 端机器的 CPU 开销。另外，如果 broker 端的压缩参数设置得与 producer 不同， broker 端在写入消息时也会额外使用 CPU 资源对消息进行对应的解压缩－重新压缩操作。

目前 Kafka 支持 3 种压缩算法：GZIP、Snappy 和 LZ4。根据实际使用经验来看 producer 结合 LZ4 的性能是最好，即 LZ4 > Snappy > GZIP。

retries

Kafka broker 在处理写入请求时可能因为瞬时的故障（比如瞬时的leader选举或者网络抖动）导致消息发送失败。这种故障通常都是可以自行恢复的，如果把这些错误封装进回调函数的异常中返还给 producer，producer程序也并没有太多可以做的，只能简单地在回调函数中重新尝试发送消息。与其这样，还不如 producer 内部自动实现重试。因此 Java 版本 producer 在内部自动实现了重试，当然前提就是要设置retries参数。

该参数表示进行重试的次数，默认值是 0，表示不进行重试。

在实际使用过程中，设置重试可以很好地应对那些瞬时错误，因此推荐用户设置该参数为一个大于 0 的值。只不过在考虑retries的设置时，有两点需要着重注意。

1. 重试可能造成消息的重复发送

比如由于瞬时的网络抖动使得 broker 端己成功写入消息但没有成功发送响应给 producer，因此 producer 会认为消息发送失败，从而开启重试机制。为了应对这一风险，Kafka 要求用户在 consumer 端必须执行去重(幂等)处理。令人欣喜的是，社区己于 0.11.0.0 版本开始支持“精确一次”处理语义，从设计上避免了类似的问题。

2. 重试可能造成消息的乱序

当前 producer 会将多个消息发送请求（默认是 5 个）缓存在内存中；如果由于某种原因发生了消息发送的重试，就可能造成消息流的乱序。为了避免乱序发生，Java 版本 producer 提供了max.in.flight.requets.per.connection 参数。一旦用户将此参数设置成1，producer将确保某一时刻只能发送一个请求。

另外，producer两次重试之间会停顿一段时间，以防止频繁地重试对系统带来冲击。这段时间是可以配置的，由参数retry.backff.ms指定，默认是 100 毫秒。由于leader“换届选举”是最常见的瞬时错误，推荐用户通过测试来计算平均 leader选举时间并根据该时间来设定retries 和retry.backff.ms 的值。

batch.size

batch.size 是 producer 最重要的参数之一，它对于调优 producer 吞吐量和延时性能指标都有着非常重要的作用。

producer 会将发往同一分区的多条消息封装进一个 batch中，当 batch 满了的时候，producer 会发送 batch 中的所有消息。不过，producer并不总是等待batch满了才发送消息，很有可能当batch还有很多空闲空间时 producer 就发送该 batch。显然，batch 的大小就显得非常重要。

通常来说，一个小的 batch 中包含的消息数很少，因而一次发送请求能够写入的消息数也很少，所以 producer 的吞吐量会很低；一个 batch 非常之巨大，那么会给内存使用带来极大的压力，因为不管是否能够填满，producer 都会为该batch 分配固定大小的内存。

因此batch.size 参数的设置其实是一种时间与空间权衡的体现。batch.size 参数默认值是 16384，即 16KB。这其实是一个非常保守的数字。在实际使用过程中合理地增加该参数值，通常都会发现 producer 的吞吐量得到了相应的增加。

linger.ms

linger.ms 参数控制消息发送延时行为。该参数默认值是 0，表示消息需要被立即发送，无须关心 batch 是否己被填满。

大多数情况下这是合理的，毕竟我们总是希望消息被尽可能快地发送，不过这样做会拉低 producer 吞吐量，毕竟 producer 发送的每次请求中包含的消息数越多，producer就越能将发送请求的开销（和Broker建立连接，数据在网络上传送时间等）摊薄到更多的消息上从而提升吞吐量。

如果要设置这个参数，需要跟上述参数batch.size 配合使用，针对消息类型做出权衡考虑。比如，在大数据或者非实时场景中，可以适当增加 linger.ms 和 batch.size 来提升producer整体吞吐量；但是在解耦或者消峰等实时链路场景中，需要的是消息的及时响应，这里低延时优先级就要高于吞吐量。

max.request.size

该参数控制的是producer 端能够发送的最大消息大小。

由于请求有一些头部数据结构，因此包含一条消息的请求的大小要比消息本身大，不过姑且把它当作请求的最大尺寸是安全的。如果 producer 要发送尺寸很大的消息，那么这个参数就是要被设置的，默认的 1048576 字节（1MB）。

request.timeout.ms

当 producer 发送请求给broker后，broker 需要在规定的时间范围内将处理结果返还给producer。默认是 30 秒。

如果 broker 在 30 秒内都没有给 producer 发送响应，就会认为该请求超时了，回调函数中显式地抛出TimeoutException 异常。默认的 30 秒对于一般的情况是足够的，但如果 producer发送的负载很大，超时的情况就很容易碰到，此时就应该适当调整该参数值。

Pt2.2 代码示例

我安装的是wurstmeister/kafka:latest版本的Kafka，进入到运行的镜像，在/opt/kafka_2.13-2.7.0/libs/目录下可以看到安装的是Kafka2.7.0版本，代码中我们也同样引入2.7.0版本，保持一致。

第一步，引入kafka依赖。Producer API和Consumer API都需要依赖kafka-clients包，为了和服务端版本保持一致，我们使用2.7.0版本。

 <!-- 引入Kafka依赖包 -->
 <!-- Producer API和Consumer API都依赖kafka-clients包。版本号我们选择和服务端一致的2.7.0版本。 -->
 <dependency>
     <groupId>org.apache.kafka</groupId>
     <artifactId>kafka-clients</artifactId>
     <version>2.7.0</version>
 </dependency>

第二步，通过代码启动生产者。

 // Kafka生产者
 public class ProducerAPI 
 
   public static void main(String[] args) 
     Properties pros = new Properties();
 
     /** 1、参数配置 */
     // 配置服务端ip:port
     pros.put("bootstrap.servers", "121.114.133.125:9092");
     // key-value的序列化协议：Kafka在发送数据时都是需要序列化的。
     pros.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
     pros.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
     // Producer确认模式：0 发出去就确认,1 落盘就确认,all 所有follower同步完才确认
     pros.put("acks", "1");
     // 消息发送异常时(未被确认),发送方重试次数
     pros.put("retries", "3");
     // 多少条数据发送一次,默认16K。达到数量会触发消息发送
     pros.put("batch.size", "16385");
     // 批量发送的等待时间,时间到了会触发消息发送
     pros.put("linger.ms", "5");
     // 客户端缓冲区大小,默认32M,缓冲区满了也会触发消息发送
     pros.put("buffer.memory", 33554432);
     // 获取元数据时生产者的阻塞时间,超时后抛出异常
     pros.put("max.block.ms", 3000);
 
     /** 2、启动生产者客户端 */
     Producer<String, String> producer = new KafkaProducer<String, String>(pros);
     System.out.println("Apache Kafka Producer starting.");
 
     /** 3、构建并发送消息对象：只有Value场景 */
     ProducerRecord valueRecord = new ProducerRecord("first_topic", "这是一条只包含Value的消息。");
     producer.send(valueRecord);
     System.out.println("You have send a value message to topic:first_topic");
 
     /** 4、构建并发送消息对象：Key-Value场景 */
     ProducerRecord keyValueRecord = new ProducerRecord("first_topic", "firstMsg", "这是一条Key-Value的消息。");
     producer.send(keyValueRecord);
     System.out.println("You have send a kv message to topic:first_topic");
 
     /** 5、关闭生产者客户端 */
     producer.close();

PT3 CONSUMERAPI

Pt3.1 Consumer 参数

bootstrap.servers

和Java版本producer相同，这是必须要指定的参数。该参数指定了一组host:port对，用于创建与Kafka broker服务器的Socket连接。可以指定多组，使用逗号分隔，如kafka:9092，kafka2:9092，kafka3:9092。在实际生产环境中需要替换成线上broker列表。

另外，若Kafka集群中broker机器数很多，我们只需要指定部分broker即可，不需要列出完整的broker列表。这是因为不管指定了几台机器，consumer启动后都能通过这些机器找到完整的broker列表，因此为该参数指定多台机器通常只是为了常规的failover使用。这样即使某一台broker挂掉了，consumer重启后依然能够通过该参数指定的其他broker连接Kafka集群。

需要注意的是，如果broker端没有显式配置listeners（或advertised.listeners）使用IP地址的话，那么最好将bootstrap.servers配置成主机名而不要使用IP地址，因为Kafka内部使用的是全称域名（FQDN，Fully Qualified Domain Name）。倘若不统一，会出现无法获取元数据的异常，这点在Producer已经做了说明。

group.id

该参数指定的是consumer group的名字。它能够唯一标识一个consumer group，官网上该参数是有默认值的，即一个空字符串。但在开发consumer程序时我们依然要显式指定group.id，否则consumer端会抛出 InvalidGroupldException 异常。

通常为group.id设置一个有业务意义的名称。

key.deserializer

consumer代码从broker端获取的任何消息都是字节数组的格式，因此消息的每个组件都要执行相应的反序列化操作才能将数据“还原”成原来的对象格式。key.deserializer参数就是为消息key指定反序列化协议。

该参数值必须是实现org.apache.kafka.common.serialization.Deserializer接口的Java类的全限定名称。Kafka默认为绝大部分的初始类型（primitivetype）提供了现成的解序列化器。

示例代码中使用org.apache.kafka.common.serialization.StringDeserializer类。该类会将接收到的字节数组转换成UTF-8编码的字符串。consumer支持用户自定义deserializer，这通常都与producer端自定义serializer“遥相呼应”。值得注意的是，不论consumer消费的消息是否指定了key，consumer都必须要设置这个参数，否则程序会抛出ConfigException，提示“key.deserializer”没有默认值。

value.deserializer

与 value.deserializer 类似，该参数被用来对消息体（即消息 value ）进行反序列化，从而把消息“还原”回原来的对象类型。value.deserializer 可以设置成与key.deserializer 不同的值，前提是 key.serializer 与 value.serializer 设置了不同的值。

在使用过程中，我们一定要谨记 key.deserializer 和 value.deserializer 指定的是类的全限定名，单独指定类名是行不通的。

session.timeout.ms

session.timeout.ms 是 consumer group 检测组内成员发送崩溃的时间。

假设你设置该参数为5分钟，那么当某个group成员突然崩溃（比如被kill -9或宕机），管理group的Kafka组件（即消费者组协调者，也称group coordinator）有可能需要5分钟才能感知到这个崩溃。通常来说，我们想要缩短这个时间，让coordinator能够更快地检测到consumer失败。

这个参数还有另外一重含义：consumer 消息处理逻辑的最大时间。倘若consumer两次poll之间的间隔超过了该参数所设置的阑值，那么coordinator就会认为这个consumer己经追不上组内其他成员的消费进度，会将该consumer实例“踢出”组，该consumer负责的分区也会被分配给其他consumer。

在最好的情况下，这会导致不必要的rebalance，因为consumer需要重新加入group。更糟的是，对于那些在被踢出group后处理的消息，consumer都无法提交位移，这就意味着这些消息在rebalance之后会被重新消费一遍。如果一条消息或一组消息总是需要花费很长的时间处理，那么consumer甚至无法执行任何消费(不断重演被踢——加入——被踢的惨剧)，除非用户重新调整参数。

鉴于以上的“窘境”，Kafka社区于0.10.1.0版本对该参数的含义进行了拆分。在该版本及以后的版本中，session.timeout.ms参数被明确为“coordinator检测失败的时间”。

因此在实际使用中，用户可以为该参数设置一个比较小的值，让coordinator能够更快地检测consumer崩溃的情况，从而更快地开启rebalance，避免造成更大的消费滞后（consumer lag）。目前该参数的默认值是10秒。

max.poll.interval.ms

如前所述，session.timeout.ms中“consumer处理逻辑最大时间”的含义被剥离出来，Kafka单独开放了一个参数一一max.poll.interval.ms。

在一个典型的consumer使用场景中，用户对于消息的处理可能需要花费很长时间。这个参数就是用于设置消息处理逻辑的最大时间的。假设用户的业务场景中消息处理逻辑是把消息“落地”到远程数据库中，且这个过程平均处理时间是2分钟，那么用户仅需要将max.poll.interval.ms设置为稍稍大于2分钟的值即可，而不必为session.neout.ms也设置这么大的值。

通过将该参数设置成实际的逻辑处理时间再结合较低的session.timeout.ms参数值，consumer group既实现了快速的consumer崩溃检测，也保证了复杂的事件处理逻辑不会造成不必要的rebalance。

auto.offset. reset

指定了无位移信息或位移越界（即consumer要消费的消息的位移不在当前消息日志的合理区间范围）时Kafka的应对策略。特别要注意这里的无位移信息或位移越界，只有满足这两个条件中的任何一个时该参数才有效果。

关于这一点，我们举一个实际的例子来说明。假设你首次运行一个consumer group并且指定从头消费。显然该group会从头消费所有数据，因为此时该group还没有任何位移信息。一旦该group成功提交位移后，你重启了group，依然指定从头消费。此时你会发现该group并不会真的从头消费一一因为Kafka己经保存了该group的位移信息，因此它会无视auto.offset.reset的设置。

目前该参数有如下3个可能的取值：

earliest：指定从最早的位移开始消费。注意这里最早的位移不一定就是 0。
latest：指定从最新处位移开始消费。
none ：指定如果未发现位移信息或位移越界，则抛出异常。

enable.auto.commit

该参数指定 consumer 是否自动提交位移。

若设置为 true，则 consumer 在后台自动提交位移，否则，用户需要手动提交位移。

对于有较强“精确处理一次”语义需求的用户来说，最好将该参数设置为 false，由用户自行处理位移提交问题。

fetch.max.bytes

该参数指定了 consumer 端单次获取数据的最大字节数。若实际业务消息很大，则必须要设置该参数为一个较大的值，否则 consumer 将无法消费这些消息。

max.poll.records

该参数控制单次 poll 调用返回的最大消息数。

比较极端的做法是设置该参数为 1，那么每次 poll 只会返回 1 条消息。如果用户发现 consumer 端的瓶颈在 poll 速度太慢，可以适当地增加该参数的值。如果用户的消息处理逻辑很轻量，默认的 500 条消息通常不能满足实际的消息处理速度。

heartbeat.interval.ms

该参数和session.timeout.ms、max.poll.interval.ms参数是最难理解的consumer参数。前面己经讨论了后两个参数的含义，这里解析一下heartbeat.interval.ms的含义及用法。

从表面上看，该参数似乎是心跳的间隔时间，但既然己经有了上面的session.timeout.ms用于设置超时，为何还要引入这个参数呢？

这里的关键在于要搞清楚consumer group的其他成员，如何得知要开启新一轮rebalance；当coordinator决定开启新一轮rebalance时，它会将这个决定以REBALANCE_IN_PROGRESS异常的形式“塞进”consumer心跳请求的response中，这样其他成员拿到response后才能知道它需要重新加入group。显然这个过程越快越好，而heartbeat.interval.ms 就是用来做这件事情的。

比较推荐的做法是设置一个比较低的值，让group下的其他consumer成员能够更快地感知新一轮rebalance开启了。注意，该值必须小于session.timeout.ms！这很容易理解，毕竟如果consumer在session.timeout.ms这段时间内都不发送心跳，coordinator就会认为它已经dead，因此也就没有必要让它知晓coordinator的决定了。

connections.max.idle.ms

经常有用户抱怨在生产环境下周期性地观测到请求平均处理时间在飘升，这很有可能是因为 Kafka 会定期地关闭空闲 Socket 连接导致下次 consumer 处理请求时需要重新创建连向 broker 的 Socket 连接。

当前默认值是 9 分钟，如果用户实际环境中不在乎这些Socket资源开销，比较推荐设置该参数值为-1，即不要关闭这些空闲连接。

Pt3.2 代码示例

第一步，引入kafka依赖。Producer API和Consumer API都需要依赖kafka-clients包，为了和服务端版本保持一致，我们使用2.7.0版本。

 <!-- 引入Kafka依赖包 -->
 <!-- Producer API和Consumer API都依赖kafka-clients包。版本号我们选择和服务端一致的2.7.0版本。 -->
 <dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>2.7.0</version>
 </dependency>

第二步，调整输出日志。消费者在运行时有大量的DEBUG日志，非常影响我们调试程序和观察结果，加入logback配置调整日志的级别。

 <!-- 路径：src/main/resources/logback.xml -->
 <?xml version="1.0" encoding="UTF-8"?>
 <configuration>
   <logger name="org.apache.kafka.clients" level="info" />
 </configuration>

第三步，启动Consumer代码。

 public class ConsumerAPI 
 
   public static void main(String[] args) 
     Properties pros = new Properties();
 
     /** 1、参数配置 */
     // 配置服务端ip:port
     pros.put("bootstrap.servers", "121.114.133.125:9092");
     // 当前消费者所属消费者组
     pros.put("group.id", "first-group");
     // 是否自动提交偏移量,commit之后才更新消费组的offset
     pros.put("enable.auto.commit", "true");
     // 消费者自动提交的时间间隔,1000ms
     pros.put("auto.commit.interval.ms", "1000");
     // 消费的模式：
     //   earliest 当各分区下有已提交的offset时,从提交的offset开始消费；无提交的offset时,从头开始消费
     //   latest   当各分区下有已提交的offset时,从提交的offset开始消费；无提交的offset时,消费新产生的该分区下的数据
     //   none     topic各分区都存在已提交的offset时,从offset后开始消费；只要有一个分区不存在已提交的offset,则抛出异常
     pros.put("auto.offset.reset", "earliest");
     // 反序列化协议
     pros.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     pros.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 
     /** 2、启动消费者 */
     Consumer<String, String> consumer = new KafkaConsumer<String, String>(pros);
     System.out.println("Apache Kafka Consumer starting.");
 
     /** 3、订阅Topic */
     consumer.subscribe(Arrays.asList("first_topic"));
 
     /** 4、Consumer持续消费消息 */
     try 
       while (true) 
         // 获取当前批次的消息
         ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
         for (ConsumerRecord<String, String> r : records) 
           System.out.printf("offset=%d, key=%s, value=%s, partiton=%s%n", r.offset(), r.key(), r.value(),
                             r.partition());
         
       
      finally 
       consumer.close();

先启动Consumer，这时候是等待消息输入的状态。然后启动上面的Producer，观察Consumer的输出。

因为之前已经通过命令行的方式创建Consumer消费了一些消息，所以这里offset是从4开始的。

 offset=4， key=null， value=这是一条只包含Value的消息。， partiton=0
 offset=5， key=firstMsg， value=这是一条Key-Value的消息。， partiton=0

PT4 SPRING KAFKA

Pt4.1 Spring Kafka入门

Spring本质上是包装了JavaAPI，使得开发人员在使用时更加的简单，只要专注于业务开发即可，下面写个简单的例子。

1、新建SpringBoot项目，这步比较基础，细节省略。

2、引入依赖包

 <!-- Spring框架Kafka依赖（核心） -->
 <dependency>
   <groupId>org.springframework.kafka</groupId>
   <artifactId>spring-kafka</artifactId>
   <version>2.7.0</version>
 </dependency>
 <!-- 引入jackson-module-kotlin（日志输出会用，不引入也没关系，有个warn） -->
 <dependency>
   <groupId>com.fasterxml.jackson.module</groupId>
   <artifactId>jackson-module-kotlin</artifactId>
   <version>2.11.0</version>
 </dependency>

3、配置Kafka

在Spring配置文件中，配置Kafka信息。

 # Kafka服务端ip:port
 spring.kafka.bootstrap.servers=121.114.133.125:9092
 
 # Producer
 # key-value的序列化协议：Kafka在发送数据时都是需要序列化的。
 spring.kafka.producer.bootstrap-servers=121.114.133.125:9092
 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
 spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
 # Producer确认模式：0 发出去就确认，1 落盘就确认，all 所有follower同步完才确认
 spring.kafka.producer.acks=1
 # 消息发送异常时(未被确认)，发送方重试次数
 spring.kafka.producer.retries=3
 # 多少条数据发送一次，默认16K。达到数量会触发消息发送
 spring.kafka.producer.batch-size=16385
 # 批量发送的等待时间，时间到了会触发消息发送
 spring.kafka.producer.properties.linger.ms=5
 # 客户端缓冲区大小，默认32M，缓冲区满了也会触发消息发送
 spring.kafka.producer.buffer-memory=33554432
 
 
 # Consumer
 # 当前消费者所属消费者组
 pros.put("group.id", "first-group");
 spring.kafka.consumer.bootstrap-servers=121.114.133.125:9092
 # 是否自动提交偏移量，commit之后才更新消费组的offset
 spring.kafka.consumer.enable-auto-commit=true
 # 消费者自动提交的时间间隔，1000ms
 spring.kafka.consumer.auto-commit-interval=1000
 # 消费的模式：
 #   earliest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
 #   latest   当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
 #   none     topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
 spring.kafka.consumer.auto-offset-reset=earliest
 # 反序列化协议
 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
 spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

4、日志配置

配置logback日志，否则控台输出的trace信息太多，影响我们调试。

 <?xml version="1.0" encoding="UTF-8"?>
 <configuration>
   <logger name="org.apache.kafka.clients" level="info"/>
 </configuration>

5、编码Consumer

 @Component
 public class SpringConsumer 
   @KafkaListener(topics = "spring-kafka-first-topic", groupId = "spring-kafka")
   public void onMessage(String msg) 
     System.out.println("我是消费者，我已收到消息: " + msg);

6、编码Producer

 @Component
 public class SpringProducer 
   @Autowired
   private KafkaTemplate<String, Object> kafkaTemplate;
 
   public String send(String msg) 
     System.out.println("我是生产者，我准备发送消息。");
     kafkaTemplate.send("spring-kafka-first-topic", msg);
     System.out.println("我是生产者，消息已经发出请接收。");
     return "SUCCESS";

7、编写测试类

 @RestController
 public class TestController 
   @Autowired
   public SpringProducer producer;
 
   @RequestMapping("/first_spring")
   public void firstTest() 
     producer.send("这是一条Spring Kafka消息");

8、验证

启动SpringBoot，在浏览器输入http://localhost:8080/first_spring，查看测试结果。

   .   ____          _            __ _ _
  /\\\\ / ___'_ __ _ _(_)_ __  __ _ \\ \\ \\ \\
 ( ( )\\___ | '_ | '_| | '_ \\/ _` | \\ \\ \\ \\
  \\\\/  ___)| |_)| | | | | || (_| |  ) ) ) )
   '  |____| .__|_| |_|_| |_\\__, | / / / /
  =========|_|==============|___/=/_/_/_/
  :: Spring Boot ::                (v2.5.0)
 
 我是生产者，我准备发送消息。
 我是生产者，消息已经发出请接收。
 我是消费者，我已收到消息: 这是一条Spring Kafka消息

Pt4.2 Spring Transaction

基于Spring实现Kafka生产者事务案例。

我们直接基于上面示例(1)中的配置，来编写事务相关处理逻辑。

1、增加事务相关配置

 # Kafka服务端ip:port
 spring.kafka.bootstrap.servers=121.114.133.125:9092
 
 
 # Producer
 # key-value的序列化协议：Kafka在发送数据时都是需要序列化的。
 spring.kafka.producer.bootstrap-servers=121.114.133.125:9092
 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
 spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
 # Producer确认模式：0 发出去就确认，1 落盘就确认，all 所有follower同步完才确认
 spring.kafka.producer.acks=all
 # 消息发送异常时(未被确认)，发送方重试次数
 spring.kafka.producer.retries=3
 # 多少条数据发送一次，默认16K。达到数量会触发消息发送
 spring.kafka.producer.batch-size=16385
 # 批量发送的等待时间，时间到了会触发消息发送
 spring.kafka.producer.properties.linger.ms=5
 # 客户端缓冲区大小，默认32M，缓冲区满了也会触发消息发送
 spring.kafka.producer.buffer-memory=33554432
 # 设置事务前缀，开启事务。
 spring.kafka.producer.transaction-id-prefix=tx-
 
 # Consumer
 # 当前消费者所属消费者组
 pros.put("group.id", "first-group");
 spring.kafka.consumer.bootstrap-servers=121.114.133.125:9092
 # 是否自动提交偏移量，commit之后才更新消费组的offset
 spring.kafka.consumer.enable-auto-commit=true
 # 消费者自动提交的时间间隔，1000ms
 spring.kafka.consumer.auto-commit-interval=1000
 # 消费的模式：
 #   earliest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
 #   latest   当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
 #   none     topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
 spring.kafka.consumer.auto-offset-reset=earliest
 # 反序列化协议
 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
 spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

2、事务生产者

 @Component
 public class TransactionProducer 
   @Autowired
   private KafkaTemplate<String, Object> kafkaTemplate;
 
   public String send() 
     try 
       kafkaTemplate.executeInTransaction(operations -> 
         System.out.println("我是生产者，我准备发送3条消息。");
         operations.send("spring-kafka-first-topic", "这是第1条消息。");
         operations.send("spring-kafka-first-topic", "这是第2条消息。");
         operations.send("spring-kafka-first-topic", "这是第3条消息。");
         System.out.println("我是生产者，消息已经发出请接收。");
         return null;
       );
      catch (Exception ex) 
       ex.printStackTrace();
     
     return "SUCCESS";

3、事务测试类

 @RestController
 public class TestTransactionController 
   @Autowired
   public TransactionProducer producer;
 
   @RequestMapping("/spring_transaction")
   public void firstTest() 
     producer.send();

运行后，结果如下：

 我是生产者，我准备发送3条消息。
 我是生产者，消息已经发出请接收。
 我是消费者，我已收到消息: 这是第1条消息。
 我是消费者，我已收到消息: 这是第2条消息。
 我是消费者，我已收到消息: 这是第3条消息。

4、重新修改生产者类

 @Component
 public class TransactionProducer 
   @Autowired
   private KafkaTemplate<String, Object> kafkaTemplate;
 
   public String send() 
     try 
       kafkaTemplate.executeInTransaction(operations -> 
         System.out.println("我是生产者，我准备发送3条消息。");
         operations.send("spring-kafka-first-topic", "这是第1条消息。");
         operations.send("spring-kafka-first-topic", "这是第2条消息。");
 
         // 抛出异常将导致事务失败
         System.out.println(1/0);
 
         operations.send("spring-kafka-first-topic", "这是第3条消息。");
         System.out.println("我是生产者，消息已经发出请接收。");
         return null;
       );
      catch (Exception ex) 
       ex.printStackTrace();
     
     return "SUCCESS";
   
 j

5、再次运行测试类

 我是生产者，我准备发送3条消息。
 java.lang.ArithmeticException: / by zero
 at com.example.kafkademo.transaction.TransactionProducer.lambda$send$0(TransactionProducer.java:21)
 at org.springframework.kafka.core.KafkaTemplate.executeInTransaction(KafkaTemplate.java:466)
 at com.example.kafkademo.transaction.TransactionProducer.send(TransactionProducer.java:15)
 at com.example.kafkademo.transaction.TestTransactionController.firstTest(TestTransactionController.java:15)
 at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
 at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
 at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
 at java.base/java.lang.reflect.Method.invoke(Method.java:566)
 at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:197)
 at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:141)
 at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:106)
 at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:894)
 at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:808)
 at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87)
 at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1063)
 at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:963)
 at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006)
 at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:898)
 at javax.servlet.http.HttpServlet.service(HttpServlet.java:626)
 at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:883)
 at javax.servlet.http.HttpServlet.service(HttpServlet.java:733)
 at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:227)
 at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)
 at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:53)
 at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189)
 at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)
 at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:100)
 at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
 at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189)
 at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)
 at org.springframework.web.filter.FormContentFilter.doFilterInternal(FormContentFilter.java:93)
 at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
 at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189)
 at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)
 at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:201)
 at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
 at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:189)
 at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:162)
 at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:202)
 at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:97)
 at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:542)
 at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:143)
 at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:92)
 at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:78)
 at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:357)
 at org.apache.coyote.http11.Http11Processor.service(Http11Processor.java:374)
 at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:65)
 at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:893)
 at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1707)
 at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)
 at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
 at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
 at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
 at java.base/java.lang.Thread.run(Thread.java:834)

以上是关于08 APACHE KAFKA实战的主要内容，如果未能解决你的问题，请参考以下文章

技术实战 |《Apache Kafka实战》作者胡夕：Apache Kafka监控与调优

《深入浅出》Apache Kafka实战

Flink实战系列Lorg/apache/flink/kafka/shaded/org/apache/kafka/clients/consumer/ConsumerRecord；)Ljava/

kafka安装到实战教程(Python版)

电子书丨《Apache Kafka实战》

Apache Kafka 在 vivo 的实战