Producer攔截器(interceptor)是個相當新的功能，它和consumer端interceptor是在Kafka 0.10版本被引入的，主要用於實現clients端的定制化控制邏輯。
對於producer而言，interceptor使得用戶在消息發送前以及producer回調邏輯前有機會對消息做一些定制化需求，比如修改消息等。同時，producer允許用戶指定多個interceptor按序作用於同一條消息從而形成一個攔截鏈(interceptor chain)。Intercetpor的實現接口是org.apache.kafka.clients.producer.ProducerInterceptor，其定義的方法包括：
- onSend(ProducerRecord)：該方法封裝進KafkaProducer.send方法中，即它運行在用戶主線程中的。Producer確保在消息被序列化以計算分區前調用該方法。用戶可以在該方法中對消息做任何操作，但最好保證不要修改消息所屬的topic和分區，否則會影響目標分區的計算
- onAcknowledgement(RecordMetadata, Exception)：該方法會在消息被應答之前或消息發送失敗時調用，並且通常都是在producer回調邏輯觸發之前。onAcknowledgement運行在producer的IO線程中，因此不要在該方法中放入很重的邏輯，否則會拖慢producer的消息發送效率
- close：關閉interceptor，主要用於執行一些資源清理工作
如前所述，interceptor可能被運行在多個線程中，因此在具體實現時用戶需要自行確保線程安全。另外倘若指定了多個interceptor，則producer將按照指定順序調用它們，並僅僅是捕獲每個interceptor可能拋出的異常記錄到錯誤日志中而非在向上傳遞。這在使用過程中要特別留意。
本文實現一個簡單的雙interceptor組成的攔截鏈。第一個interceptor會在消息發送前將時間戳信息加到消息value的最前部；第二個interceptor會在消息發送后更新成功發送消息數或失敗發送消息數。這兩個interceptor實現的邏輯其實都很簡單，特別是第一個interceptor做的事情實際上並無太多的實際意義，只是為了演示如何使用interceptor以及連接鏈。
TimeStampPrependerInterceptor類
public class TimeStampPrependerInterceptor implements ProducerInterceptor<String, String> {
@Override
public void configure(Map<String, ?> configs) {
}
@Override
public ProducerRecord onSend(ProducerRecord record) {
return new ProducerRecord(
record.topic(), record.partition(), record.timestamp(), record.key(), System.currentTimeMillis() + "," + record.value().toString());
}
@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
}
@Override
public void close() {
}
}
上面代碼關鍵的部分在於onSend方法中我們會創建一個新的record，把時間戳寫入消息體的最前部。下面我們定義第二個interceptor：CounterInterceptor，該interceptor會在消息發送后更新"發送成功消息數"和"發送失敗消息數"兩個計數器，並在producer關閉時打印這兩個計數器。完整代碼如下所示：
CounterInterceptor類
public class CounterInterceptor implements ProducerInterceptor<String, String> {
private int errorCounter = 0;
private int successCounter = 0;
@Override
public void configure(Map<String, ?> configs) {
}
@Override
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
return record;
}
@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
if (exception == null) {
successCounter++;
} else {
errorCounter++;
}
}
@Override
public void close() {
// 保存結果
System.out.println("Successful sent: " + successCounter);
System.out.println("Failed sent: " + errorCounter);
}
}
定義好interceptor之后，我們需要在producer主程序中指定它們，代碼如下：
Properties props = new Properties();
props.put(...);
// 構建攔截鏈
List<String> interceptors = new ArrayList<>();
interceptors.add("huxi.test.producer.TimeStampPrependerInterceptor"); // interceptor 1
interceptors.add("huxi.test.producer.CounterInterceptor"); // interceptor 2
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
...
String topic = "test-topic";
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 10; i++) {
ProducerRecord<String, String> record = new ProducerRecord<>(topic, "message" + i);
producer.send(record).get();
}
// 一定要關閉producer，這樣才會調用interceptor的close方法
producer.close();
上面代碼發送10條消息到Kafka，我們利用kafka-console-consumer工具來測試下兩個interceptor是否工作正常。輸出如下所示：
可見所有消息的value前面都被加上了發送時的時間戳，這就說明TimeStampPrependerInterceptor是正常工作的。而運行producer之后的輸出如下：
Successful sent: 10
Failed sent: 0
此輸出說明了CounterInterceptor也是能夠正常工作的。我們的雙interceptor構成的連接鏈生效了。