咱們在寫Kafka消費者的時候，有沒有發(fā)現(xiàn)一個很麻煩的事：消費消息前每次都要手動解析Kafka消息，轉(zhuǎn)換成自己想要的類型，再進行業(yè)務(wù)操作
大家好，我是Jensen。一個想和大家一起打怪升級的程序員朋友。
咱們在寫Kafka消費者的時候，有沒有發(fā)現(xiàn)一個很麻煩的事：消費消息前每次都要手動解析Kafka消息，轉(zhuǎn)換成自己想要的類型，再進行業(yè)務(wù)操作，比如：
/**
* 訂單支付成功通知
* @Author 公眾號：架構(gòu)師修行錄
*/
@KafkaListener(topics = "oms.orderPaySuccess", groupId = "fms")
public void orderPaySuccess(ConsumerRecord<String, String> consumerRecord) {
// 解析Kafka消息
OrderPaySuccessEvent event = JSON.parseObject(consumerRecord.value(), OrderPaySuccessEvent.class);
// TODO 完成解析成功后的業(yè)務(wù)操作
}
對代碼有潔癖的同志就比較難受了，每次解析的操作都差不多，但這又不是業(yè)務(wù)代碼……于是你就想：有沒有一個辦法讓系統(tǒng)能夠自動解析成自己想要的參數(shù)對象呢？
不過也發(fā)現(xiàn)了一些難以解決的問題，比如：共用了一個groupId，使用Java線程池來管理，這樣工程會存在性能瓶頸。
專業(yè)的事還是交給專業(yè)的“人”來做吧，用Kafka組件本身來管理不同分組消費會更靠譜。
Spring官方也發(fā)現(xiàn)這個問題，并且對此提出了解決方案——spring-messaging包下的HandlerMethodArgumentResolver接口。
我們從官方文檔中看到，Spring-kafka在2.4.2版本支持了對kafka消息進行方法參數(shù)級別轉(zhuǎn)換。
其實，spring-web包下也有這個同名接口，用于自動解析Controller方法上的參數(shù)，這塊網(wǎng)上資料比較多我就不詳細展開了，而spring-messaging包下面這個接口的非官方資料比較少，我這里給大家總結(jié)一下用法。
HandlerMethodArgumentResolver方法參數(shù)處理器接口很簡單，只有兩個方法：
public interface HandlerMethodArgumentResolver {
boolean supportsParameter(MethodParameter var1);
@Nullable
Object resolveArgument(MethodParameter var1, Message<?> var2) throws Exception;
}supportsParameter方法返回是否支持參數(shù)自動解析，resolveArgument方法就是具體的解析邏輯，把MQ傳遞參數(shù)轉(zhuǎn)換為具體類型參數(shù)。
我們來看一下實際案例。
首先實現(xiàn)HandlerMethodArgumentResolver接口，定義為一個Spring的Component：
@Component
public class KafkaListenerMethodArgumentResolver implements HandlerMethodArgumentResolver {
@Override
public boolean supportsParameter(@NonNull MethodParameter parameter) {
// 默認以com.xxx開頭的類，這樣可以不用在在參數(shù)前加@Payload注解
return parameter.getParameterType().getName().startsWith("com.xxx") || parameter.hasParameterAnnotation(Payload.class);
}
@Override
public Object resolveArgument(@NonNull MethodParameter parameter, @NonNull Message<?> message) {
Class<?> parameterType = parameter.getParameterType();
String messageContent = (String) message.getPayload();
Object body;
try {
// 這里定義自己的解析方法
body = JsonUtils.fromJson(messageContent, parameterType);
Objects.requireNonNull(body);
} catch (Throwable cause) {
throw new KafkaException("kafka 消息解析失敗: 非法JSON字符串", cause);
}
// 可選，定義解析后的參數(shù)校驗邏輯
validate(parameter, body);
return body;
}
private void validate(MethodParameter parameter, Object target) {
for (Annotation ann : parameter.getParameterAnnotations()) {
Validated validatedAnn = AnnotationUtils.getAnnotation(ann, Validated.class);
if (Objects.nonNull(validatedAnn) || ann.annotationType().getSimpleName().startsWith("Valid")) {
ValidationUtils.valid(target);
}
}
}
}
到這里，一個類就把參數(shù)自動解析搞定了，接下來咱們看看怎么用。
/**
* 訂單支付成功通知
* @Author 公眾號：架構(gòu)師修行錄
*/
@KafkaListener(topics = "oms.orderPaySuccess", groupId = "fms")
public void orderPaySuccess(@Payload OrderPaySuccessEvent orderPaySuccessEvent) {
// 已經(jīng)自動解析Kafka消息為orderPaySuccessEvent參數(shù)
// TODO 完成解析成功后的業(yè)務(wù)操作
}
怎么樣，代碼是不是比原來簡潔多了，香不香！
如果你對技術(shù)有追求，不想一直寫業(yè)務(wù)代碼，不妨把項目中所有需要手動解析參數(shù)的地方，替換成自定義方法參數(shù)解析器來實現(xiàn)~
