kafka拦截器实现队列插队效果

@Slf4j
@Service
public class MyProducerInterceptor implements ProducerInterceptor {

    //在发送broke之前的一个操作，可以对数据进行加工处理或者进行topic pationtion重新指向     
    @Override
    public ProducerRecord onSend(ProducerRecord producerRecord) {
        return null;
    }
    /**
    * KafkaProducer 会在消息被应答（Acknowledgement）之前或消息发送失败时调用生产者拦截器的 onAcknowledgement（）方法
    * 
    *
    **/
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    //close（）方法主要用于在关闭拦截器时执行一些资源的清理工作。
    @Override
    public void close() {

    }

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

    }
}

    服务端应答的模式可以通过acks进行设置: 1 0 -1
1 : 在生产者发送消息之后，从节点保存完数据，就会进行响应，如果消息无法写入leader副本，比如在leader 副本崩溃、重新选举新的 leader 副本的过程中，那么生产者就会收到一个错误的响应，为了避免消息丢失，生产者可以选择重发消息。如果消息写入leader副本并返回成功响应给生产者，且在被其他follower副本拉取之前leader副本崩溃，那么此时消息还是会丢失，因为新选举的leader副本中并没有这条对应的消息。acks设置为1，是消息可靠性和吞吐量之间的折中方案。

0 : acks=0。生产者发送消息之后不需要等待任何服务端的响应。如果在消息从发送到写入Kafka的过程中出现某些异常，导致Kafka并没有收到这条消息，那么生产者也无从得知，消息也就丢失了。在其他配置环境相同的情况下，acks 设置为 0 可以达到最大的吞吐量。

-1: acks=-1或acks=all。生产者在消息发送之后，需要等待ISR中的所有副本都成功写入消息之后才能够收到来自服务端的成功响应。在其他配置环境相同的情况下，acks 设置为-1（all）可以达到最强的可靠性。但这并不意味着消息就一定可靠，因为ISR中可能只有leader副本，这样就退化成了acks=1的情况

    AtomicInteger atomicInteger=new AtomicInteger(0);
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> consumerRecords) {
        
        return new ConsumerRecords( );
    }

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

    private String getJsonTrackingMessage(ConsumerRecord<String, String> record) {
        return record.value();
    }

    //拦截器关闭做一些操作
    @Override
    public void close() {

    }

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

        int mark = atomicInteger.incrementAndGet();
        Map<TopicPartition, List<ConsumerRecord<String, String>>> newRecords
                = new HashMap<>();
        ConsumerRecord consumerRecord=
                new ConsumerRecord("xpp_test",0,0,null,1000);
        for (TopicPartition tp : records.partitions()) {
            List<ConsumerRecord<String, String>> tpRecords =
                    records.records(tp);
            List<ConsumerRecord<String, String>> newTpRecords = new ArrayList<>();
            newTpRecords.add(consumerRecord);
            newTpRecords.addAll(tpRecords);
            if (!newTpRecords.isEmpty()) {
                newRecords.put(tp, newTpRecords);
            }
        }
        return new ConsumerRecords<>(newRecords);