SpringCloud 處理非反序列化異常

對于Kafka Streams活頁夾中的常規(guī)錯誤處理，最終用戶應用程序要處理應用程序級錯誤。作為為反序列化異常處理程序提供DLQ的副作用，Kafka Streams綁定器提供了一種訪問直接從應用程序發(fā)送bean的DLQ的方法。一旦可以訪問該bean，就可以以編程方式將任何異常記錄從應用程序發(fā)送到DLQ。

使用高級DSL仍然難以進行魯棒的錯誤處理。Kafka Streams本身還不支持錯誤處理。

但是，當您在應用程序中使用低級處理器API時，有一些選項可以控制此行為。見下文。

@Autowired
private SendToDlqAndContinue dlqHandler;

@StreamListener("input")
@SendTo("output")
public KStream<?, WordCount> process(KStream<Object, String> input) {

    input.process(() -> new Processor() {
    			ProcessorContext context;

    			@Override
    			public void init(ProcessorContext context) {
    				this.context = context;
    			}

    			@Override
    			public void process(Object o, Object o2) {

    			    try {
    			        .....
    			        .....
    			    }
    			    catch(Exception e) {
    			        //explicitly provide the kafka topic corresponding to the input binding as the first argument.
                        //DLQ handler will correctly map to the dlq topic from the actual incoming destination.
                        dlqHandler.sendToDlq("topic-name", (byte[]) o1, (byte[]) o2, context.partition());
    			    }
    			}

    			.....
    			.....
    });
}