kafka可插拔增強如何實現？

image.png

導彈攔截，精準防禦。

背景

攔截器：在不修改應用程式業務邏輯的情況下，一組基於事件的可插拔的邏輯處理鏈；
類比springMVC的攔截器：

image.png

file

這些都是通過配置攔截器，插入到應用程式中，實現可插拔的修改業務邏輯；

kafka在0.10.0.0版本中開始引入攔截器。分為生產者攔截器和消費者攔截器，類似責任鏈的方式編排多個攔截器為一個大攔截器。

配置方法：配置參數


Properties props = new Properties();
List<String> interceptors = new ArrayList<>();
interceptors.add("com.yourcompany.kafkaproject.interceptors.AddTimestampInterceptor"); // 攔截器1
interceptors.add("com.yourcompany.kafkaproject.interceptors.UpdateCounterInterceptor"); // 攔截器2
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
……

file

注意： 配置攔截器需要制定攔截器的全限定名，並且保證生產者或者消費者客戶端能夠正確載入到配置的攔截器；

file

通過攔截器實現，強制讓所有的生產者，消費者配置該攔截器，實現消息審計的功能； |

生產者攔截器

攔截器需要實現org.apache.kafka.clients.producer.ProducerInterceptor

file

消費者攔截器

org.apache.kafka.clients.consumer.ConsumerInterceptor

file

實操

實現端到端的性能監控：

處理過程：

file

生產者程式碼：

public class AvgLatencyProducerInterceptor implements ProducerInterceptor<String, String> {


    private Jedis jedis; // 省略Jedis初始化


    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        jedis.incr("totalSentMessage");
        return record;
    }


    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    }


    @Override
    public void close() {
    }


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

消費者程式碼：

public class AvgLatencyConsumerInterceptor implements ConsumerInterceptor<String, String> {


    private Jedis jedis; //省略Jedis初始化


    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        long lantency = 0L;
        for (ConsumerRecord<String, String> record : records) {
            lantency += (System.currentTimeMillis() - record.timestamp());
        }
        jedis.incrBy("totalLatency", lantency);
        long totalLatency = Long.parseLong(jedis.get("totalLatency"));
        long totalSentMsgs = Long.parseLong(jedis.get("totalSentMessage"));
        jedis.set("avgLatency", String.valueOf(totalLatency / totalSentMsgs));
        return records;
    }


    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
    }


    @Override
    public void close() {
    }


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

配置到攔截器到對應的生產者和消費者對象，即簡單的實現了平均消息延時的端到端性能統計。

小結

類比AOP是Spring提供的核心功能，即面向切面編程，可以把跟業務邏輯無關的安全，審計，性能相關功能放到切面增強中實現。
對Kafka進行一些可插拔的功能增強可以通過攔截器實現。

本篇介紹了kafka的攔截器的使用方法，以及通過實例展示了具體的用法，希望對團隊使用的kafka做一些增強功能的時候可以利用這個點去擴展。

image.png

原創不易，關注誠可貴，轉發價更高！轉載請註明出處，讓我們互通有無，共同進步，歡迎溝通交流。
我會持續分享Java軟體編程知識和程式設計師發展職業之路，歡迎關注，我整理了這些年編程學習的各種資源，關注公眾號『李福春持續輸出』，發送'學習資料'分享給你！

Tags:

VirMach 便宜 VPS