前言在日常的微服务开发中我们经常需要将核心接口的调用数据如入参、返参、耗时等异步上报到 Kafka供下游进行数据同步、审计或数据计算。如果在每个业务方法里手动编写 Kafka 发送逻辑不仅代码冗余度高而且严重侵入业务逻辑。本文将分享一种基于自定义注解 AOP 拦截的优雅方案实现接口调用数据的无侵入式、异步、防重上送并支持线程池隔离与动态配置。一、核心诉求与设计目标在设计这套消息上送组件时我们主要考虑了以下几个核心诉求诉求说明低侵入业务代码尽量不感知 Kafka 上送逻辑做到声明式调用异步化上送过程不能阻塞主业务线程保障接口响应速度防重复同一请求如同一分页数据在同一天内避免重复上送可配置Topic、线程池参数等可通过配置中心动态调整链路追踪异步线程需透传 TraceID保证日志链路完整二、整体架构设计整体数据流转分为四层业务层、AOP 拦截层、异步执行层和 Kafka 发送层。┌─────────────────────────────────────────────────────────────┐ │ 业务 Service 层 │ │ PushToKafka(scene, type) │ │ ↓ 方法调用 │ ├─────────────────────────────────────────────────────────────┤ │ AOP 拦截层 │ │ PushToKafkaAspect.around() │ │ ├─ 记录方法入参 / 返参 / 耗时 / 异常 │ │ ├─ Redis 防重锁校验 │ │ └─ 提交异步任务到线程池 │ ├─────────────────────────────────────────────────────────────┤ │ KafkaPushExecutorProvider │ │ ↓ 获取独立的 kafkaPushThreadPool │ ├─────────────────────────────────────────────────────────────┤ │ KafkaMessagePushProducerService │ │ ├─ 构造 KafkaMessageDto │ │ ├─ 注入公共字段 (如业务标识、接口名) │ │ └─ KafkaTemplate.send(ProducerRecord) │ ├─────────────────────────────────────────────────────────────┤ │ Kafka Broker │ └─────────────────────────────────────────────────────────────┘三、核心组件代码详解3.1 自定义注解PushToKafka通过自定义注解标记需要上送 Kafka 的方法实现业务代码零侵入。packagecom.example.kafka.annotation;importjava.lang.annotation.*;/** * Kafka 推送注解 * p * 用于标记需要推送请求入参和接口返参到 Kafka 的方法。 * 通过 AOP 拦截该注解异步将方法调用信息发送到 Kafka。 */Target(ElementType.METHOD)Retention(RetentionPolicy.RUNTIME)publicinterfacePushToKafka{/** * 业务场景描述 */Stringscene()default;/** * 业务类型标识 */Stringtype()default;}设计要点Target(ElementType.METHOD)仅作用于方法级别。Retention(RetentionPolicy.RUNTIME)运行时保留供 AOP 反射读取。type字段用于区分不同业务场景最终写入 Kafka 消息体供下游分类消费。3.2 AOP 拦截器PushToKafkaAspect环绕通知拦截注解标记的方法在方法执行完成后异步触发 Kafka 上送。packagecom.example.kafka.aspect;importcom.example.common.model.dto.JsonResultDto;importcom.example.kafka.annotation.PushToKafka;importcom.example.kafka.config.KafkaPushExecutorProvider;importcom.example.kafka.service.KafkaMessagePushProducerService;importlombok.extern.slf4j.Slf4j;importorg.aspectj.lang.ProceedingJoinPoint;importorg.aspectj.lang.annotation.Around;importorg.aspectj.lang.annotation.Aspect;importorg.aspectj.lang.reflect.MethodSignature;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.beans.factory.annotation.Value;importorg.springframework.stereotype.Component;importjava.lang.reflect.Method;Slf4jAspectComponentpublicclassPushToKafkaAspect{AutowiredprivateKafkaMessagePushProducerServicekafkaProducerService;AutowiredprivateKafkaPushExecutorProviderkafkaPushExecutorProvider;Value(${app.kafka.topic.push-data:datasync-public-logs})privateStringkafkaTopic;/** * 环绕通知拦截 PushToKafka 注解标记的方法 */Around(annotation(com.example.kafka.annotation.PushToKafka))publicObjectaround(ProceedingJoinPointjoinPoint)throwsThrowable{longstartTimeSystem.currentTimeMillis();Objectresultnull;Throwableerrornull;try{resultjoinPoint.proceed();returnresult;}catch(Throwablet){errort;throwt;}finally{longcostTimeSystem.currentTimeMillis()-startTime;asyncPushToKafka(joinPoint,result,error,costTime);}}/** * 异步推送数据到 Kafka */privatevoidasyncPushToKafka(ProceedingJoinPointjoinPoint,Objectresult,Throwableerror,longcostTime){try{MethodSignaturesignature(MethodSignature)joinPoint.getSignature();Methodmethodsignature.getMethod();PushToKafkapushToKafkamethod.getAnnotation(PushToKafka.class);if(pushToKafkanull)return;Object[]argsjoinPoint.getArgs();// 此处可根据实际业务校验入参和返参类型例如// if (args null || args.length 0 || !(args[0] instanceof BaseReq)) return;// if (result null || !(result instanceof JsonResultDto)) return;// 获取业务参数StringinterfaceNamejoinPoint.getSignature().getName();StringtypepushToKafka.type();log.info(开始异步推送 Kafka 数据type: {}, interfaceName: {}, costTime: {}ms,type,interfaceName,costTime);// 提交异步任务到独立线程池kafkaPushExecutorProvider.getExecutor().execute(()-kafkaProducerService.sendMessage(kafkaTopic,type,interfaceName,result));}catch(Exceptione){log.error(异步推送 Kafka 数据异常,e);}}}核心逻辑拆解环绕拦截Around拦截目标方法记录startTime。执行原方法joinPoint.proceed()异常原样抛出不影响主流程。异步提交通过专用线程池执行 Kafka 发送避免阻塞主线程。3.3 Kafka 生产者服务与 Redis 防重锁负责消息序列化、字段注入、防重校验及最终发送到 Kafka Broker。packagecom.example.kafka.service;importcom.alibaba.fastjson.JSON;importlombok.extern.slf4j.Slf4j;importorg.apache.kafka.clients.producer.ProducerRecord;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.data.redis.core.RedisTemplate;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.kafka.support.SendResult;importorg.springframework.stereotype.Service;importorg.springframework.util.concurrent.ListenableFuture;importorg.springframework.util.concurrent.ListenableFutureCallback;importjava.util.Calendar;importjava.util.concurrent.TimeUnit;Slf4jServicepublicclassKafkaMessagePushProducerService{AutowiredprivateKafkaTemplateString,StringkafkaTemplate;AutowiredprivateRedisTemplateString,StringredisTemplate;/** * 发送消息到 Kafka */publicvoidsendMessage(Stringtopic,Stringtype,StringinterfaceName,Objectresult){// 1. 构造消息体并序列化StringmessageJsonJSON.toJSONString(result);// 2. 构建 ProducerRecordProducerRecordString,StringrecordnewProducerRecord(topic,messageJson);// 3. 异步发送并监听回调ListenableFutureSendResultString,StringfuturekafkaTemplate.send(record);future.addCallback(newListenableFutureCallbackSendResultString,String(){OverridepublicvoidonSuccess(SendResultString,Stringresult){log.debug(Kafka 消息发送成功, topic{},topic);}OverridepublicvoidonFailure(Throwableex){log.error(Kafka 消息发送失败, topic{},topic,ex);}});}/** * 检查 Redis 锁是否存在 (防重机制) */publicbooleanisLock(StringlockKey){returnBoolean.TRUE.equals(redisTemplate.hasKey(lockKey));}/** * 设置 Redis 锁过期时间为当天最后一分钟 */publicvoidlock(StringlockKey){CalendarnowCalendar.getInstance();CalendarendOfDayCalendar.getInstance();endOfDay.set(Calendar.HOUR_OF_DAY,23);endOfDay.set(Calendar.MINUTE,59);endOfDay.set(Calendar.SECOND,59);longexpireMinutesTimeUnit.MILLISECONDS.toMinutes(endOfDay.getTimeInMillis()-now.getTimeInMillis())1;if(expireMinutes0)expireMinutes1;redisTemplate.opsForValue().set(lockKey,1,expireMinutes,TimeUnit.MINUTES);}}防重锁特性粒度细以接口名 业务主键 分页参数为维度避免同一分页数据重复上送。当天有效过期时间动态计算到当天23:59:59次日自动失效并可重新上送。3.4 独立线程池配置Kafka 上送采用独立线程池与业务线程隔离。推荐使用ThreadPoolTaskExecutor并结合链路追踪组件如 Sleuth/Micrometer透传 TraceID。# application.yml (支持配置中心动态下发)app:trace:thread:configList:-threadNameBean:kafkaPushThreadPoolcorePoolSize:4maxPoolSize:16queueCapacity:200keepAliveSeconds:60threadNamePrefix:kafka-push-# 队列满时由调用线程自己执行避免丢数据rejectedExecutionHandler:java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy四、使用方式在需要上送 Kafka 的 Service 方法上添加PushToKafka注解即可真正做到了声明式编程ServicepublicclassDataService{PushToKafka(scene核心业务数据同步,typebusiness_data_sync)publicJsonResultDtoDataVOqueryData(QueryReqreq){// 1. 纯业务逻辑...DataVOdatadoBusinessLogic(req);returnJsonResultDto.success(data);}}零侵入体现❌ 无需修改方法内部逻辑❌ 无需手动构造 Kafka 消息❌ 无需关注线程池和异常处理五、关键设计要点总结设计点方案收益注解驱动自定义PushToKafka AOP 拦截业务代码零侵入声明式使用异步发送独立线程池kafkaPushThreadPool不阻塞主业务线程保障接口 RT防重机制Redis 分布式锁当天维度去重避免重复数据降低 Kafka 压力失败感知ListenableFutureCallback回调发送失败可记录日志便于排查动态配置Topic、线程池均走 Nacos/Apollo无需发版即可调整参数链路追踪支持 TraceID 透传的线程池包装TraceID 在线程池间透传日志可串联六、生产环境注意事项切面执行顺序若方法上存在其他切面如Transactional、Cacheable需关注Order优先级确保 Kafka 切面在最外层能捕获到最终的完整结果。序列化兼容性消息体建议统一转为 JSON 字符串并与下游消费方约定好字段结构避免反序列化失败。Redis 锁失效兜底若 Redis 出现网络抖动导致锁未正常写入可能会产生少量重复消息。必须在 Kafka 消费端做好幂等性处理如基于唯一键去重。线程池监控建议对kafkaPushThreadPool的活跃线程数、队列积压量配置 Prometheus 监控与告警防止队列打满触发拒绝策略。原创不易如果这篇文章对你有帮助欢迎点赞、收藏、关注你的支持是我持续创作的动力。