io.github.changebooks.kafka

类 KafkaBatchListener

java.lang.Object

io.github.changebooks.kafka.KafkaBatchListener

public class KafkaBatchListener
extends Object

消息分页，并行消费 总行数 / 线程数 = 每线程处理行数 可以整除时，线程数 == 实际线程数，如：总行数 = 9 和 线程数 = 3，实际结果：实际线程数 = 3，每线程处理行数 = 3 不能整除时，线程数 != 实际线程数，如：总行数 = 9 和 线程数 = 4，实际结果：实际线程数 = 3，每线程处理行数 = 3

作者:

changebooks@qq.com

构造器概要

构造器

构造器和说明
KafkaBatchListener(Executor executor, KafkaBatchConsumer consumer)

方法概要

限定符和类型	方法和说明
boolean	asyncConsume(List<List<org.apache.kafka.clients.consumer.ConsumerRecord<String,String>>> pageList, KafkaBatchContext context) 多线程消费
boolean	consume(List<org.apache.kafka.clients.consumer.ConsumerRecord<String,String>> records, KafkaBatchContext context) 执行消费
KafkaBatchConsumer	getConsumer()
Executor	getExecutor()
int	getThreadNum()
boolean	listen(List<org.apache.kafka.clients.consumer.ConsumerRecord<String,String>> records, KafkaBatchContext context) 计算分页，多线程消费
KafkaBatchListener	setThreadNum(int threadNum) 设置线程数 实际线程数 = CEIL(总行数 / CEIL(总行数 / 线程数))

从类继承的方法 java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

构造器详细资料

KafkaBatchListener

public KafkaBatchListener(Executor executor,
                          KafkaBatchConsumer consumer)

方法详细资料

listen

public boolean listen(List<org.apache.kafka.clients.consumer.ConsumerRecord<String,String>> records,
                      @Nullable
                      KafkaBatchContext context)

计算分页，多线程消费

参数:

records - 消息列表

context - 消费上下文

返回:

消费成功提交消息？否则，消费失败等待重试

asyncConsume

public boolean asyncConsume(@NonNull
                            List<List<org.apache.kafka.clients.consumer.ConsumerRecord<String,String>>> pageList,
                            @Nullable
                            KafkaBatchContext context)

多线程消费

参数:

pageList - 分页列表

context - 消费上下文

返回:

全部线程消费成功提交消息？否则，任一线程消费失败等待重试

consume

public boolean consume(List<org.apache.kafka.clients.consumer.ConsumerRecord<String,String>> records,
                       @Nullable
                       KafkaBatchContext context)

执行消费

参数:

records - 消息列表

context - 消费上下文

返回:

消费成功提交消息？否则，消费失败等待重试

getExecutor

public Executor getExecutor()

getConsumer

public KafkaBatchConsumer getConsumer()

getThreadNum

public int getThreadNum()

setThreadNum

public KafkaBatchListener setThreadNum(int threadNum)

设置线程数 实际线程数 = CEIL(总行数 / CEIL(总行数 / 线程数))

参数:

threadNum - 线程数，线程数 小于 (线程池最大线程数 - 1)

返回:

the KafkaBatchListener instance