程序员经验分享-hwhale

kafka基础学习（六）：kafka 代码示例

2023-10-30

kafka 代码示例

生成者API在不同版本间无较大变动；消费者（Consumer）提供了两套API

  • 低版本（8.0版本及其以前）API
  • 高版本（8.0版本后）API

两种API的优缺点：

高版本Consumer API 优点：

  • 高级API写起来简单，易用。
    不需要自行去管理offset，API已经封装好了offset这块的东西，会通过zookeeper自行管理
    不需要管理分区，副本等情况，系统自动管理
    消费者断线后会自动根据上次记录在zookeeper中的offset接着消费消息。

高版本Consumer API 缺点：

  • 不能自行控制offset。
  • 不能自行管理分区，副本，zk等相关信息。

低版本Consumer API 优点：

  • 能够让开发者自己维护offset.想从哪里消费就从哪里消费
  • 自行控制连接分区，对分区自定义负载均衡
  • 对zookeeper的依赖性降低（如 offset 不一定要用zk来存储，可以存在缓存里或者内存中）

低版本Consumer API 缺点：

  • 过于复杂，需要自行控制offset，连接哪个分区，找分区leader等。

使用者可以根据自身情对两种API况进行选取。

一：Producer

消息发送流程

kafka 消息发布流程

Kafka的Producer发送消息采用的是异步发送的方式。在消息发送的过程中，涉及到了两个线程——main线程和Sender线程，以及一个线程共享变量——RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。

注意事项：参数说明

  • batch.size：只有数据积累到batch.size之后，sender才会发送数据。
  • linger.ms：如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。

代码示例

1. 导入依赖

<!-- kafka -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.11.0.0</version>
</dependency>

2. 代码部分

常用的类

  • KafkaProducer ：操作对象，用于数据发送。
  • ProducerRecord ：封装为ProducerRecord对象，进行传输。
  • ProducerConfig ：进行参数配置。

KafkaProducer 的构造方法

public KafkaProducer(Map<String, Object> configs) {
    this(new ProducerConfig(configs), null, null);
}
public KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
        this(new ProducerConfig(ProducerConfig.addSerializerToConfig(configs, keySerializer, valueSerializer)),
                keySerializer, valueSerializer);
}
public KafkaProducer(Properties properties) {
    this(new ProducerConfig(properties), null, null);
}
private KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
        try {
            log.trace("Starting the Kafka producer");
            Map<String, Object> userProvidedConfigs = config.originals();
            this.producerConfig = config;
            ......
}

上述源码可以看出，构造ProducerConfig来完成KafkaProducer的创建。而ProducerConfig的构造可以通过Map或者Properties来完成。本次代码示例使用Properties完成ProducerConfig的创建。

ProducerConfig 参数说明如下表

属性 描述 类型 默认值
bootstrap.servers 用于建立与kafka集群的连接，这个list仅仅影响用于初始化的hosts，来发现全部的servers。
格式：host1:port1,host2:port2,…，数量尽量不止一个，以防其中一个down了		 list
acks Server完成 producer request 前需要确认的数量。
acks=0时，producer不会等待确认，直接添加到socket等待发送；
acks=1时，等待leader写到local log就行；
acks=all或acks=-1时，等待isr中所有副本确认
（注意：确认都是 broker 接收到消息放入内存就直接返回确认，不是需要等待数据写入磁盘后才返回确认，这也是kafka快的原因）		 string 1
buffer.memory Producer可以用来缓存数据的内存大小。该值实际为RecordAccumulator类中的BufferPool，即Producer所管理的最大内存。
如果数据产生速度大于向broker发送的速度，producer会阻塞max.block.ms，超时则抛出异常		 long 33554432
compression.type Producer用于压缩数据的压缩类型，取值：none, gzip, snappy, or lz4 string none
batch.size Producer可以将发往同一个Partition的数据做成一个Produce Request发送请求，即Batch批处理，以减少请求次数，该值即为每次批处理的大小。
另外每个Request请求包含多个Batch，每个Batch对应一个Partition，且一个Request发送的目的Broker均为这些partition的leader副本。
若将该值设为0，则不会进行批处理		 int 16384
linger.ms Producer默认会把两次发送时间间隔内收集到的所有Requests进行一次聚合然后再发送，以此提高吞吐量，而linger.ms则更进一步，这个参数为每次发送增加一些delay，以此来聚合更多的Message。
官网解释翻译：producer会将request传输之间到达的所有records聚合到一个批请求。通常这个值发生在欠负载情况下，record到达速度快于发送。但是在某些场景下，client即使在正常负载下也期望减少请求数量。这个设置就是如此，通过人工添加少量时延，而不是立马发送一个record，producer会等待所给的时延，以让其他records发送出去，这样就会被聚合在一起。这个类似于TCP的Nagle算法。该设置给了batch的时延上限：当我们获得一个partition的batch.size大小的records，就会立即发送出去，而不管该设置；但是如果对于这个partition没有累积到足够的record，会linger指定的时间等待更多的records出现。该设置的默认值为0(无时延)。例如，设置linger.ms=5，会减少request发送的数量，但是在无负载下会增加5ms的发送时延。		 long 0
max.request.size 请求的最大字节数。这也是对最大消息大小的有效限制。注意：server具有自己对消息大小的限制，这些大小和这个设置不同。此项设置将会限制producer每次批量发送请求的数目，以防发出巨量的请求。 int 1048576
receive.buffer.bytes TCP的接收缓存 SO_RCVBUF 空间大小，用于读取数据 int 32768
request.timeout.ms client等待请求响应的最大时间,如果在这个时间内没有收到响应，客户端将重发请求，超过重试次数发送失败 int 30000
send.buffer.bytes TCP的发送缓存 SO_SNDBUF 空间大小，用于发送数据 int 131072
timeout.ms 指定server等待来自followers的确认的最大时间，根据acks的设置，超时则返回error int 30000
max.in.flight.requests.per.connection 在block前一个connection上允许最大未确认的requests数量。
当设为1时，即是消息保证有序模式，注意：这里的消息保证有序是指对于单个Partition的消息有顺序，因此若要保证全局消息有序，可以只使用一个Partition，当然也会降低性能		 int 5
metadata.fetch.timeout.ms 在第一次将数据发送到某topic时，需先fetch该topic的metadata，得知哪些服务器持有该topic的partition，该值为最长获取metadata时间 long 60000
reconnect.backoff.ms 连接失败时，当我们重新连接时的等待时间 long 50
retry.backoff.ms 在重试发送失败的request前的等待时间，防止若目的Broker完全挂掉的情况下Producer一直陷入死循环发送，折中的方法 long 100

概括下代码过程的大致过程：

1. 生产者通过KafkaProducer进行封装，通过该对象进行消息的生产。
2. KafkaProducer对象需要进行一些配置，而配置的创建依赖于ProducerConfig。
3. KafkaProducer生产的消息的载体为ProducerRecord。

异步发送（无回调函数）代码如下

有无回调函数的原因是kafkaProducer.send()方法重载。

public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
    return send(record, null);
}

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // intercept the record, which can be potentially modified; this method does not throw exceptions
    ProducerRecord<K, V> interceptedRecord = this.interceptors == null ? record : this.interceptors.onSend(record);
    return doSend(interceptedRecord, callback);
}

Properties properties = new Properties();
// bootstrap.servers 配置kafka集群中broker-list
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.115.157:9092");
// 配置消息确认（所有副本均确认后，返回确认消息）
properties.put(ProducerConfig.ACKS_CONFIG, "1");
// 重试次数
properties.put("retries", 1);
// 批次大小
properties.put("batch.size", 16384);
// 等待时间
properties.put("linger.ms", 1);
// RecordAccumulator缓冲区大小
properties.put("buffer.memory", 33554432);
// 设置生产者生产的消息的key值的数据类型
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 设置生产者生产的消息的value值的数据类型
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 构建KafkaProducer对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
// 模拟生产100消息
for (int i = 0; i < 100; i++) {
	kafkaProducer.send(new ProducerRecord<String, String>("test", "key" + i, "value" + i ));
	// 若想对返回值进行处理则可以加上Callback参数
	// kafkaProducer.send(new ProducerRecord<String, String>("test", "key" + i, "value" + i), (metadata, exception) -> {
	//      if (exception == null){
	//			 // 无异常，输出偏移量
	//		     System.out.println(metadata.offset());
	//		 } else {
	//			 exception.printStackTrace();
	//		}
	// });
}
// 关闭生产者
kafkaProducer.close();

同步发送代码如下

同步发送的特点是，一条消息发送后，会阻塞当前线程，直至返回ack为止。

利用又返回值的KafkaProducer的send方法，返回的Future对象来实现同步发送的效果。

即kafkaProducer.send().get();

get();方法是获取返回对象，而该方法会阻塞主线程，直至获取到返回对象为止。

......
for (int i = 0; i < 100; i++) {
	RecordMetadata test = kafkaProducer.send(new ProducerRecord<String, String>("test", "key" + i, "value" + i), (metadata, exception) -> {
		if (exception == null) {
			// 无异常，输出偏移量
			System.out.println(metadata.offset());
		} else {
			exception.printStackTrace();
		}
	}).get();
	System.out.println(test.offset());
}

二：Consumer

消费者业务逻辑的关键在于维护offset；

两种方式

  • 手动提交（低版本【0.8版本及其以前】的kafka消费者API）
  • 自动提交（高版本【0.8版本以后】的kafka消费者API）

与Producer相似，有以下常用类

  • KafkaConsumer：需要创建一个消费者对象，用来消费数据
  • ConsumerConfig：获取所需的一系列配置参数
  • ConsuemrRecord：每条数据都要封装成一个ConsumerRecord对象
NAME DESCRIPTION TYPE DEFAULT
bootstrap.servers host/port,用于和kafka集群建立初始化连接。因为这些服务器地址仅用于初始化连接，并通过现有配置的来发现全部的kafka集群成员（集群随时会变化），所以此列表不需要包含完整的集群地址（但尽量多配置几个，以防止配置的服务器宕机）。 list
key.deserializer key的解析序列化接口实现类（Deserializer）。 class
value.deserializer value的解析序列化接口实现类（Deserializer） class
fetch.min.bytes 服务器哦拉取请求返回的最小数据量，如果数据不足，请求将等待数据积累。默认设置为1字节，表示只要单个字节的数据可用或者读取等待请求超时，就会应答读取请求。将此值设置的越大将导致服务器等待数据累积的越长，这可能以一些额外延迟为代价提高服务器吞吐量。 int 1
group.id 此消费者所属消费者组的唯一标识。如果消费者用于订阅或offset管理策略的组管理功能，则此属性是必须的。 string ""
heartbeat.interval.ms 当使用Kafka的分组管理功能时，心跳到消费者协调器之间的预计时间。心跳用于确保消费者的会话保持活动状态，并当有新消费者加入或离开组时方便重新平衡。该值必须必比session.timeout.ms小，通常不高于1/3。它可以调整的更低，以控制正常重新平衡的预期时间。 int 3000
max.partition.fetch.bytes 服务器将返回每个分区的最大数据量。如果拉取的第一个非空分区中第一个消息大于此限制，则仍然会返回消息，以确保消费者可以正常的工作。broker接受的最大消息大小通过message.max.bytes（broker config）或max.message.bytes (topic config)定义。参阅fetch.max.bytes以限制消费者请求大小。 int 1048576
session.timeout.ms 用于发现消费者故障的超时时间。消费者周期性的发送心跳到broker，表示其还活着。如果会话超时期满之前没有收到心跳，那么broker将从分组中移除消费者，并启动重新平衡。请注意，该值必须在broker配置的group.min.session.timeout.ms和group.max.session.timeout.ms允许的范围内。 int 10000
ssl.key.password 密钥存储文件中的私钥的密码。 客户端可选 password null
ssl.keystore.location 密钥存储文件的位置， 这对于客户端是可选的，并且可以用于客户端的双向认证。 string null
ssl.keystore.password 密钥仓库文件的仓库密码。客户端可选，只有ssl.keystore.location配置了才需要。 password null
ssl.truststore.location 信任仓库文件的位置 string null
ssl.truststore.password 信任仓库文件的密码 password null
auto.offset.reset 当Kafka中没有初始offset或如果当前的offset不存在时（例如，该数据被删除了），该怎么办。
最早：自动将偏移重置为最早的偏移
 最新：自动将偏移重置为最新偏移
none：如果消费者组找到之前的offset，则向消费者抛出异常
 其他：抛出异常给消费者。		 string latest
connections.max.idle.ms 指定在多少毫秒之后关闭闲置的连接 long 540000
enable.auto.commit 如果为true，消费者的offset将在后台周期性的提交 boolean true
exclude.internal.topics 内部topic的记录（如偏移量）是否应向消费者公开。如果设置为true，则从内部topic接受记录的唯一方法是订阅它。 boolean true
fetch.max.bytes 服务器为拉取请求返回的最大数据值。这不是绝对的最大值，如果在第一次非空分区拉取的第一条消息大于该值，该消息将仍然返回，以确保消费者继续工作。接收的最大消息大小通过message.max.bytes (broker config) 或 max.message.bytes (topic config)定义。注意，消费者是并行执行多个提取的。 int 52428800
max.poll.interval.ms 使用消费者组管理时poll()调用之间的最大延迟。消费者在获取更多记录之前可以空闲的时间量的上限。如果此超时时间期满之前poll()没有调用，则消费者被视为失败，并且分组将重新平衡，以便将分区重新分配给别的成员。 int 300000
max.poll.records 在单次调用poll()中返回的最大记录数。 int 500
partition.assignment.strategy 当使用组管理时，客户端将使用分区分配策略的类名来分配消费者实例之间的分区所有权 list class org.apache.kafka
.clients.consumer
.RangeAssignor
receive.buffer.bytes 读取数据时使用的TCP接收缓冲区（SO_RCVBUF）的大小。 如果值为-1，则将使用OS默认值。 int 65536
request.timeout.ms 配置控制客户端等待请求响应的最长时间。 如果在超时之前未收到响应，客户端将在必要时重新发送请求，如果重试耗尽则客户端将重新发送请求。 int 305000
sasl.jaas.config JAAS配置文件中SASL连接登录上下文参数。 这里描述JAAS配置文件格式。 该值的格式为： '(=)*;' password null
sasl.kerberos.service.name Kafka运行Kerberos principal名。可以在Kafka的JAAS配置文件或在Kafka的配置文件中定义。 string null
sasl.mechanism 用于客户端连接的SASL机制。安全提供者可用的机制。GSSAPI是默认机制。 string GSSAPI
security.protocol 用于与broker通讯的协议。 有效值为：PLAINTEXT，SSL，SASL_PLAINTEXT，SASL_SSL。 string PLAINTEXT
send.buffer.bytes 发送数据时要使用的TCP发送缓冲区（SO_SNDBUF）的大小。 如果值为-1，则将使用OS默认值。 int 131072
ssl.enabled.protocols 启用SSL连接的协议列表。 list TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type key仓库文件的文件格式，客户端可选。 string JKS
ssl.protocol 用于生成SSLContext的SSL协议。 默认设置是TLS，这对大多数情况都是适用的。 最新的JVM中的允许值为TLS，TLSv1.1和TLSv1.2。 较旧的JVM可能支持SSL，SSLv2和SSLv3，但由于已知的安全漏洞，不建议使用SSL。 string TLS
ssl.provider 用于SSL连接的安全提供程序的名称。 默认值是JVM的默认安全提供程序。 string null
ssl.truststore.type 信任存储文件的文件格式。 string JKS
auto.commit.interval.ms 如果enable.auto.commit设置为true，则消费者偏移量自动提交给Kafka的频率（以毫秒为单位）。 int 5000
check.crcs 自动检查CRC32记录的消耗。 这样可以确保消息发生时不会在线或磁盘损坏。 此检查增加了一些开销，因此在寻求极致性能的情况下可能会被禁用。 boolean true
client.id 在发出请求时传递给服务器的id字符串。 这样做的目的是通过允许将逻辑应用程序名称包含在服务器端请求日志记录中，来跟踪ip/port的请求源。 string ""
fetch.max.wait.ms 如果没有足够的数据满足fetch.min.bytes，服务器将在接收到提取请求之前阻止的最大时间。 int 500
interceptor.classes 用作拦截器的类的列表。 你可实现ConsumerInterceptor接口以允许拦截（也可能变化）消费者接收的记录。 默认情况下，没有拦截器。 list null
metadata.max.age.ms 在一定时间段之后（以毫秒为单位的），强制更新元数据，即使没有任何分区领导变化，任何新的broker或分区。 long 300000
metric.reporters 用作度量记录员类的列表。实现MetricReporter接口以允许插入通知新的度量创建的类。JmxReporter始终包含在注册JMX统计信息中。 list ""
metrics.num.samples 保持的样本数以计算度量。 int 2
metrics.recording.level 最高的记录级别。 string INFO
metrics.sample.window.ms The window of time a metrics sample is computed over. long 30000
reconnect.backoff.ms 尝试重新连接指定主机之前等待的时间，避免频繁的连接主机，这种机制适用于消费者向broker发送的所有请求。 long 50
retry.backoff.ms 尝试重新发送失败的请求到指定topic分区之前的等待时间。避免在某些故障情况下，频繁的重复发送。 long 100
sasl.kerberos.kinit.cmd Kerberos kinit命令路径。 string /usr/bin/kinit
sasl.kerberos.min.time.before.relogin 尝试/恢复之间的登录线程的休眠时间。 long 60000
sasl.kerberos.ticket.renew.jitter 添加到更新时间的随机抖动百分比。 double 0.05
sasl.kerberos.ticket.renew.window.factor 登录线程将休眠，直到从上次刷新到ticket的指定的时间窗口因子到期，此时将尝试续订ticket。 double 0.8
ssl.cipher.suites 密码套件列表，用于TLS或SSL网络协议的安全设置，认证，加密，MAC和密钥交换算法的明明组合。默认情况下，支持所有可用的密码套件。 list null
ssl.endpoint.identification.algorithm 使用服务器证书验证服务器主机名的端点识别算法。 string null
ssl.keymanager.algorithm 密钥管理器工厂用于SSL连接的算法。 默认值是为Java虚拟机配置的密钥管理器工厂算法。 string SunX509
ssl.secure.random.implementation 用于SSL加密操作的SecureRandom PRNG实现。 string null
ssl.trustmanager.algorithm 信任管理器工厂用于SSL连接的算法。 默认值是为Java虚拟机配置的信任管理器工厂算法。 string PKIX

从上表提取出一些项目中常用的配置

fetch.min.bytes

这个参数允许消费者指定从broker读取消息时最小的数据量。当消费者从broker读取消息时，如果数据量小于这个阈值，broker会等待直到有足够的数据，然后才返回给消费者。对于写入量不高的主题来说，这个参数可以减少broker和消费者的压力，因为减少了往返的时间。而对于有大量消费者的主题来说，则可以明显减轻broker压力。

fetch.max.wait.ms

上面的fetch.min.bytes参数指定了消费者读取的最小数据量，而这个参数则指定了消费者读取时最长等待时间，从而避免长时间阻塞。这个参数默认为500ms。

max.partition.fetch.bytes

这个参数指定了每个分区返回的最多字节数，默认为1M。也就是说，KafkaConsumer.poll()返回记录列表时，每个分区的记录字节数最多为1M。如果一个主题有20个分区，同时有5个消费者，那么每个消费者需要4M的空间来处理消息。实际情况中，我们需要设置更多的空间，这样当存在消费者宕机时，其他消费者可以承担更多的分区。

需要注意的是，max.partition.fetch.bytes必须要比broker能够接收的最大的消息（由max.message.size设置）大，否则会导致消费者消费不了消息。另外，在上面的样例可以看到，我们通常循环调用poll方法来读取消息，如果max.partition.fetch.bytes设置过大，那么消费者需要更长的时间来处理，可能会导致没有及时poll而会话过期。对于这种情况，要么减小max.partition.fetch.bytes，要么加长会话时间。

session.timeout.ms

这个参数设置消费者会话过期时间，默认为3秒。也就是说，如果消费者在这段时间内没有发送心跳，那么broker将会认为会话过期而进行分区重平衡。这个参数与heartbeat.interval.ms有关，heartbeat.interval.ms控制KafkaConsumer的poll()方法多长时间发送一次心跳，这个值需要比session.timeout.ms小，一般为1/3，也就是1秒。更小的session.timeout.ms可以让Kafka快速发现故障进行重平衡，但也加大了误判的概率（比如消费者可能只是处理消息慢了而不是宕机）。

auto.offset.reset

这个参数指定了当消费者第一次读取分区或者上一次的位置太老（比如消费者下线时间太久）时的行为，可以取值为latest（从最新的消息开始消费）或者earliest（从最老的消息开始消费）。

enable.auto.commit

这个参数指定了消费者是否自动提交消费位移，默认为true。如果需要减少重复消费或者数据丢失，你可以设置为false。如果为true，你可能需要关注自动提交的时间间隔，该间隔由auto.commit.interval.ms设置。

partition.assignment.strategy

我们已经知道当消费组存在多个消费者时，主题的分区需要按照一定策略分配给消费者。这个策略由PartitionAssignor类决定，默认有两种策略：

  • 范围（Range）：对于每个主题，每个消费者负责一定的连续范围分区。假如消费者C1和消费者C2订阅了两个主题，这两个主题都有3个分区，那么使用这个策略会导致消费者C1负责每个主题的分区0和分区1（下标基于0开始），消费者C2负责分区2。可以看到，如果消费者数量不能整除分区数，那么第一个消费者会多出几个分区（由主题数决定）。
  • 轮询（RoundRobin）：对于所有订阅的主题分区，按顺序一一的分配给消费者。用上面的例子来说，消费者C1负责第一个主题的分区0、分区2，以及第二个主题的分区1；其他分区则由消费者C2负责。可以看到，这种策略更加均衡，所有消费者之间的分区数的差值最多为1。
    partition.assignment.strategy设置了分配策略，默认为org.apache.kafka.clients.consumer.RangeAssignor（使用范围策略），你可以设置为org.apache.kafka.clients.consumer.RoundRobinAssignor（使用轮询策略），或者自己实现一个分配策略然后将partition.assignment.strategy指向该实现类。

client.id

这个参数可以为任意值，用来指明消息从哪个客户端发出，一般会在打印日志、衡量指标、分配配额时使用。

max.poll.records

这个参数控制一个poll()调用返回的记录数，这个可以用来控制应用在拉取循环中的处理数据量。

receive.buffer.bytes、send.buffer.bytes

这两个参数控制读写数据时的TCP缓冲区，设置为-1则使用系统的默认值。如果消费者与broker在不同的数据中心，可以一定程度加大缓冲区，因为数据中心间一般的延迟都比较大。

大致过程

  • 构建KafkaConsumer
  • 调用消费方法：KafkaConsumer.poll();

自动提交和手动提交

  • 提交的内容为offset：意义在于标记消息读取到的位置+1，即下次消费消息的开始位置。
  • 当提交设置不合理，可能发生消息的重复消费和消息缺失的情况。具体情况后续会详细分析！

简单的kafka消费者代码示例

public static void main(String[] args) {
	Properties properties = new Properties();
	// 配置kafka broker-list
	properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.115.157:9092");
	// 设置消费者消费的消息的key值的数据类型
	properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
	// 设置消费者消费的消息的value值的数据类型
	properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
	// 配置消费者组，相同groupId为同一个消费组
	properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group_test");
	// 自动提交offset关闭
	properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
	// 下次消费的位置
	// 可以取值为latest（从最新的消息开始消费）或者earliest（从最老的消息开始消费）。
	properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
	// 创建KafkaConsumer对象
	KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
	// 订阅消费的Topic - Topic名为test
	kafkaConsumer.subscribe(Arrays.asList("test"));
	// 消费数据
	try {
		while (true) {
			// 每100毫秒消费一次消息
			ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(100);
			consumerRecords.forEach(consumerRecord -> {
				System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n",
						consumerRecord.topic(), consumerRecord.partition(), consumerRecord.offset(),
						consumerRecord.key(), consumerRecord.value());
			});
			// 异步提交offset
			kafkaConsumer.commitSync();
		}
	} catch(CommitFailedException e){
		e.printStackTrace();
	} finally {
		// 关闭消费者
		kafkaConsumer.close();
	}
}

自动提交

  • 无需kafkaConsumer.commitSync();进行offset提交，自动提交，会在消费者消费消息后自动提交。
  • properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, “false”); 配置为true（默认为true）。
  • properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, “earliest”); 配置下次读取的开始位置。

异步提交与同步提交

手动提交offset的方法有两种：分别是commitSync（同步提交）和commitAsync（异步提交）。

  • 两者的相同点是，都会将本次poll的一批数据最高的偏移量提交；
  • 不同点是，commitSync会失败重试，一直到提交成功（如果由于不可恢复原因导致，也会提交失败）；而commitAsync则没有失败重试机制，故有可能提交失败。

获取分区定位以及重新分区

在消费者订阅主题中有个重载方法可以定义获取的分区以及重新分区

// 订阅消费的Topic - Topic名为test
kafkaConsumer.subscribe(Arrays.asList("test"),
		new ConsumerRebalanceListener() {
			// 定义了重新获取分区（消费者组中新增了消费者后，消费者对应的分区会根据分区规则进行重新分区）
			@Override
			public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
				System.out.println("====重新分区====");
				partitions.forEach(partition->{
					System.out.printf("Topic = %s , partition = %i", partition.topic(), partition.partition());
				});
			}
			// 定位新分配分区的offset
			@Override
			public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
				System.out.println("====初次获取分区====");
				partitions.forEach(partition->{
					System.out.printf("Topic = %s , partition = %i", partition.topic(), partition.partition());
					// 获取offset （伪代码，从ZK或者存储offset的位置中获取对应TopicPartition的offset）
					long offset = getoffset(partition);
					// 定位offset，后续消费从offset后消费消息
					kafkaConsumer.seek(partition, offset + 1);
				});


			}
		}
);
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

面试

kafka

消息中间件

生产者与消费者

代码示例

kafka基础学习（六）：kafka 代码示例 的相关文章

  • 未转变者怎么调服务器难度,Unturned——作弊模式下的各项数值微调【较实用的已详细描述】...

    您尚未登录 立即登录享受更好的浏览体验 您需要 登录 才可以下载或查看 没有帐号 注册 register x 本帖最后由 Crazy Zombie 于 2017 8 11 10 31 编辑 如标题所示 在下发一个关于Unturned模式下各

  • 区块链与哈希函数

    目录 哈希函数 定义 性质 发展 常见攻击方法 1 穷举攻击 2 生日攻击 3 其他攻击 构造方法 1 利用对称密码体制来设计哈希函数 2 直接设计哈希函数 编辑 常用哈希函数简介 1 SHA 256算法 编辑 2 Keccak算法 3 S

  • Rust 学习心得＜3＞：无栈协程

    Rust 学习心得 lt 3 gt 无栈协程 有栈协程 vs 无栈协程 Go 有栈协程 Rust 协程 绿色线程 GreenThread 无栈协程 协程解决的问题 Rust作为一门新兴语言 主打系统编程 提供了多种编写代码的模式 Rust在

  • 虚幻引擎入门_框架

    虚幻引擎所提供的GamePlay框架可谓是虚幻引擎最为重要的一部分内容也不为过 虚幻引擎的设计希望开发人员在使用引擎之前是准备好了的 并且有充足的能力去理解游戏设计意图 在此之上为我们提供了一套开发规则 我们称之为游戏框架 GamePlay

随机推荐

  • 给定一个字符串，验证它是否是回文串，只考虑字母和数字字符，可以忽略字母的大小写。

    1 给定一个字符串 验证它是否是回文串 只考虑字母和数字字符 可以忽略字母的大小写 说明 本题中 我们将空字符串定义为有效的回文串 示例 1 输入 A man a plan a canal Panama 输出 true 示例 2 输入 ra

  • 环形缓冲器

    环形缓冲器 ringr buffer 也称作圆形队列 circular queue 循环缓冲区 cyclic buffer 圆形缓冲区 circula buffer 是一种用于表示一个固定尺寸 头尾相连的缓冲区的数据结构 适合缓存数据流 中

  • 编译出错总结

    1 incompatible types in assignment of char to char 64 可以使用strcpy 解决 2 error assigning to an array from an initializer li

  • Unity PlayerPrefs记录时间

    static public void SaveDateTimeInPlayerPrefs string key DateTime dt string timeStr dt ToString yyyy MM dd HH mm ss Playe

  • 时序预测

    时序预测 MATLAB实现基于BP神经网络的时间序列预测 递归预测未来 多指标评价 目录 时序预测 MATLAB实现基于BP神经网络的时间序列预测 递归预测未来 多指标评价 预测结果 基本介绍 程序设计 参考资料 预测结果 基本介绍 Mat

  • IOS7手势识别

    添加一个view 给view加上各种手势查看效果 UIView gestureTestView UIView alloc initWithFrame CGRectMake 100 100 200 200 gestureTestView ba

  • AI开源项目分享：pytorch增强图像数据、Tensorflow 2.0实现...

    文末免费送电子书 七月在线干货组最新 升级的 名企AI面试100题 免费送 项目一 igel 无需编写代码即可训练 测试和使用模型 一个令人愉悦的机器学习工具 可让您无需编写代码即可训练 拟合 测试和使用模型 该项目的目标是为技术用户和非技

  • Windows server 2008 r2关于SMB签名未配置的漏洞修复

    一 漏洞描述 SMB签名未配置漏洞是指 远程SMB服务器上未配置需要签名 这将导致未经身份验证的远程攻击者可以利用此漏洞对SMB服务器进行中间人攻击 SMB是一个协议名 全称是Server Message Block 服务器消息快协议 用于

  • 第二十三篇：UE Ceisum开发之如何加载本地地形及经纬度、空间坐标相互转化

    本篇是基于Cesium for Unreal这个插件 基本入门我这里就不多说了 不会的人可以先跟着这个教程学习一下 Cesium for Unreal快速入门 本篇重点讲述一下如何加载本地地形 1 准备工作 安装最新的Cesiumlab软件

  • JavaScript中的事件循环（event loop）机制

    聚沙成塔 每天进步一点点 专栏简介 调用栈 Call Stack 消息队列 Message Queue 事件循环 Event Loop 宏任务和微任务 示例 写在最后 专栏简介 前端入门之旅 探索Web开发的奇妙世界 记得点击上方或者右侧链

  • C#报错：线程间操作无效: 从不是创建控件“XXXX”的线程访问它

    看到一篇很好的博文 如果只是测试用的 不需要上位机多稳定 直接用第一种方法 非常nice 在zhihu上看到的一篇文章 C 线程间操作无效 从不是创建控件 dataGridView 的线程访问它 知乎 https zhuanlan zhih

  • Nginx启动时提示nginx: [emerg] still could not bind()

    今天跟着视频学习Nginx时 启动Nginx报这个nginx emerg still could not bind 于是在网上找了解决方法在这里分享出来 报错界面 解决方法 1 根据Nginx配置文件查看配置的端口默认的是80端口 然后我们

  • PTA 浙大版《C语言程序设计（第3版）》题目集 练习5-2

    本题要求对两个整数a和b 输出其中较大的数 函数接口定义 int max int a int b 其中a和b是用户传入的参数 函数返回的是两者中较大的数 裁判测试程序样例 include

  • 小博客练习

    博客 简单的综合运用 一 了解博客需求 1 主要的分为前台和后台 后台 注册登录 登录 登录 分类管理 显示分类 编辑分类 添加分类 删除分类 文章管理 显示文章 编辑文章 添加文章 删除文章 用户管理 标签管理 等等功能 前台 首页面 列

  • linux常用命令总结

    linux常用命令总结 1 在根目录下查找xxx文件 2 列出程序启动时搜索库路径的过程 3 查看程序或so的依赖 1 查看Test程序依赖的so文件 2 也可以查看so依赖的文件 4 SSH相关 5 查看当前登录的用户 6 查看重启 登录

  • Java面试题整理-高并发篇

    1 synchronized的实现原理以及锁优化 synchronized原理分析 遇见更好的自己 的博客 CSDN博客Java 高并发专题之synchronized关键字1 synchronized作为jvm关键字有三个作用域synchr

  • rancher部署

    Rancher 管理 Kubernetes 集群 Rancher 简介 Rancher 是一个开源的企业级多集群 Kubernetes 管理平台 实现了 Kubernetes 集群在混合云 本地数据中心的集中部署与管理 以确保集群的安全性

  • 在脚本中 使用数组 获取列表中选中的复选框id + 获取选中的单选框的值

    var chk value 定义一个数组 input name communicateTeacher claIds checked each function 遍历每一个名字为interest的复选框 其中选中的执行函数 chk value

  • Pandas提取指定行列

    取列 1 通过列名称来提取指定列 推荐 列名 irline sentiment gold name negativereason gold retweet count text get data data airline sentiment

  • kafka基础学习（六）：kafka 代码示例

    kafka 代码示例 生成者API在不同版本间无较大变动 消费者 Consumer 提供了两套API 低版本 8 0版本及其以前 API 高版本 8 0版本后 API 两种API的优缺点 高版本Consumer API 优点 高级API写起

热门标签