SpringCloud 處理非反序列化異常

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

使用高級(jí)DSL仍然難以進(jìn)行魯棒的錯(cuò)誤處理。Kafka Streams本身還不支持錯(cuò)誤處理。

但是，當(dāng)您在應(yīng)用程序中使用低級(jí)處理器API時(shí)，有一些選項(xiàng)可以控制此行為。見下文。

@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());
    			    }
    			}

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