Kafka消费者特性

//这个方法会在再均衡开始之前和消费者停止读取消息之后被调用。
//可以通过这个回调方法来处理消费位移的提交，以此来避免一些不必要的重复消费现象的发生。
//参数 partitions 表示再均衡前所分配到的分区。
void onPartitionsRevoked(Collection partitions);
//这个方法会在重新分配分区之后和消费者开始读取消费之前被调用。
//参数 partitions 表示再均衡后所分配到的分区。
void onPartitionsAssigned(Collection partitions); 


//配合外部存储使用的再平衡监听器
consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        //store offset in DB （storeOffsetToDB）
    }
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        for(TopicPartition tp: partitions){
            consumer.seek(tp, getOffsetFromDB(tp));//从DB中读取消费位移
        }
    }
});

//在 poll() 方法返回之前调用拦截器的 onConsume() 方法来对消息进行相应的定制化操作
//比如修改返回的消息内容、按照某种规则过滤消息（可能会减少 poll() 方法返回的消息的个数）。
//如果 onConsume() 方法中抛出异常，那么会被捕获并记录到日志中，但是异常不会再向上传递。
public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records)；


//在提交完消费位移之后调用拦截器的 onCommit() 方法
//使用这个方法来记录跟踪所提交的位移信息
//比如当消费者使用commitSync的无参方法时，我们不知道提交的消费位移的具体细节，而使用拦截器的onCommit()方法却可以做到这一点
public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets)；
public void close();

public class ConsumerInterceptorTTL implements ConsumerInterceptor<String, String> {
    private static final long EXPIRE_INTERVAL = 10 * 1000;


    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        long now = System.currentTimeMillis();
        Map<TopicPartition, List<ConsumerRecord<String, String>>> newRecords = new HashMap<>();
        for (TopicPartition tp : records.partitions()) {
            List<ConsumerRecord<String, String>> tpRecords = records.records(tp);
            List<ConsumerRecord<String, String>> newTpRecords = new ArrayList<>();
            for (ConsumerRecord<String, String> record : tpRecords) {
                //判断消息的在有效期内,在使用ProducerRecord时可指定定义消息的时间戳
                if (now - record.timestamp() < EXPIRE_INTERVAL) {
                    newTpRecords.add(record);
                }
            }
            if (!newTpRecords.isEmpty()) {
                newRecords.put(tp, newTpRecords);
            }
        }
        return new ConsumerRecords<>(newRecords);
    }


    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        offsets.forEach((tp, offset) -> 
                System.out.println(tp + ":" + offset.offset()));
    }


    @Override
    public void close() {}


    @Override
    public void configure(Map<String, ?> configs) {}
}

props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
ConsumerInterceptorTTL.class.getName());