与原生KafkaConsumer接口适配说明

Set<TopicPartition> assignment()

接口

支持

获取consumer消费的通道与分区信息

Set<String> subscription()

接口

支持

获取consumer已订阅的通道名称

void assign(Collection<TopicPartition> var1)

接口

支持

分配指定的分区

void subscribe(Collection<String> var1)

接口

支持

订阅指定的通道

void subscribe(Collection<String> var1, ConsumerRebalanceListener var2)

接口

支持

订阅指定的通道并支持ConsumerRebalanceListener回调

void subscribe(Pattern var1, ConsumerRebalanceListener var2)

接口

支持

订阅所有匹配通配符的通道并支持ConsumerRebalanceListener回调

void unsubscribe()

接口

支持

取消所有订阅

ConsumerRecords<K, V> poll(long var1)

接口

支持

获取消息，但消息当中未实现

checksum(消息的CRC32校验值)、serializedKeySize(key序列化后的字节长度)、serializedValueSize(key序列化后的字节长度)。

void commitSync()

接口

支持

同步提交当前消费的offset

void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets)

接口

支持

同步提交指定的offset

void commitAsync()

接口

支持

异步提交当前消费的offset

public void commitAsync(OffsetCommitCallback callback)

接口

支持

异步提交当前消费的offset并支持OffsetCommitCallback回调

void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)

接口

支持

异步提交指定的offset并支持OffsetCommitCallback回调

void seek(TopicPartition partition, long offset)

接口

支持

给分区设置指定的offset

void seekToBeginning(Collection<TopicPartition> partitions)

接口

支持

分区的offset设置为最旧的值

void seekToEnd(Collection<TopicPartition> partitions)

接口

支持

分区的offset设置为最新的值

long position(TopicPartition partition)

接口

支持

获取分区当前已消费数据的offset

OffsetAndMetadata committed(TopicPartition partition)

接口

支持

获取分区已提交的offset

List<PartitionInfo> partitionsFor(String topic)

接口

支持

获取通道的分区信息，但PartitionInfo里面的leader, replicas, inSyncReplicas未实现。

Map<String, List<PartitionInfo>> listTopics()

接口

支持

获取所有的通道信息，但PartitionInfo里面的leader, replicas, inSyncReplicas未实现。

void pause(Collection<TopicPartition> partitions)

接口

支持

暂停消费分区

void resume(Collection<TopicPartition> partitions)

接口

支持

恢复消费分区

Set<TopicPartition> paused()

接口

支持

获取所有已暂停消费的分区

close()

接口

支持

关闭consumer

Map<MetricName, ? extends Metric> metrics()

接口

不支持

获取统计信息

wakeup()

接口

不支持

内部实现原理不一样，不需要。

group.id

参数

支持

消费组ID

client.id

参数

支持

每个consumer的client.id必须唯一，如果不配置client.id, dis kafka consumer会生成一个uuid作为client.id。

key.deserializer

参数

支持

含义与kafka设置相同，但默认值为StringDeserializer (kafka必须配置)。

value.deserializer

参数

支持

含义与kafka设置相同，但默认值为StringDeserializer (kafka必须配置)。

enable.auto.commit

参数

支持

同kafka的默认设置，默认为true

• true表示启用自动提交，每隔${auto.commit.interval.ms}的时间提交一次offset；
• false表示不自动提交offset

auto.commit.interval.ms

参数

支持

自动提交offset的周期(毫秒)，默认值5000。

auto.offset.reset

参数

支持

同Kafka的默认配置，默认为latest。

此值用于没有初始偏移量或者偏移量不正确的情况下，自动设置offset位置：

• earliest 将偏移量自动重置为最旧的值;
• latest将偏移量自动重置为最新的值;
• none 如果在消费者组中没有发现前一个偏移量，就向消费者抛出一个异常;

其他参数

参数

不支持

-