TP-Link 478+ 升级秘密武器：高效固件包等你来下载！

admin 管理员组

文章数量: 1184232

ZIP文件

资源目录

相关推荐

核心逻辑： * 1. 若DLQ未启用，直接调用原始处理器； * 2. 若启用，按配置重试处理事件； * 3. 重试耗尽后发送事件到DLQ。 * * @param event 待处理的事件 */ @Override public void handleEvent(Event event) { if (!dlqConfig.isEnabled()) { delegate.handleEvent(event); return; } int retryCount = event.getRetryCount(); // 达到阈值直接进入死信 if (retryCount > dlqConfig.getMaxRetries()) { sendToDLQ(event, retryCount, lastException, mainTopic + "_dlq_topic"); return; } try { delegate.handleEvent(event); } catch (Exception e) { int newRetryCount = retryCount + 1; event.setRetryCount(newRetryCount); if (newRetryCount > dlqConfig.getMaxRetries()) { sendToDLQ(event, newRetryCount, e.getMessage(), mainTopic + "_dlq_topic"); } else { long delay = dlqConfig.getRetryStrategy().getNextDelay(newRetryCount); eventCenter.sendDelay(mainTopic, event, delay, new FlexibleMatchStrategy(DelayQueueStrategy.ApproximationMode.UP)); log.info("事件延时 {}ms 后重试", delay); } } } /** * 将事件发送至死信队列 */ private void sendToDLQ(Event event,int retryCount,String lastException,String dlqTopic) { // 构造自定义EventFuture回调，捕获发送结果 EventFuture dlqEventFuture = new EventFuture() { @Override public void onSuccess(EventCenterSendResult result) { log.info("发送至死信队列！原始事件: {},时间戳：{}，主题: {}, 分区: {}, 已重试次数: {},最后一次异常信息：{}", event.toString(),result.getTimestamp(),result.getTopic(), result.getPartition(), retryCount - 1 , lastException); } @Override public void onFailure(Throwable throwable) { // 关键：记录重试次数和发送异常 log.error("死信消息发送失败！原始事件: {}, 已重试次数: {}, 异常信息: {}", event.toString(), retryCount - 1 , throwable.getMessage(), throwable); } }; try { // 调用带回调的send方法（匹配参数：topic, event, eventFuture） eventCenter.send(dlqTopic, event, dlqEventFuture); } catch (Exception e) { // 处理send方法本身的异常（如参数错误） log.error("调用死信发送方法时发生异常", e); } } } 2025-10-17 10:04:47.425 INFO 19516 --- [nPool-worker-11] c.t.n.d.b.DLQGenericValidationApp : 处理EventV2事件: Hello EventV2! 2025-10-17 10:04:47.426 INFO 19516 --- [nPool-worker-11] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=1, 到期时间expirationTime=1760666688426 2025-10-17 10:04:47.426 INFO 19516 --- [onPool-worker-4] c.t.n.d.b.DLQGenericValidationApp : 处理EventV2事件: Hello EventV2! 2025-10-17 10:04:47.426 INFO 19516 --- [onPool-worker-4] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=3, 到期时间expirationTime=1760666697426 2025-10-17 10:04:47.446 INFO 19516 --- [nPool-worker-11] c.t.n.d.b.DLQGenericValidationApp : 处理CustomEvent事件 2025-10-17 10:04:47.447 ERROR 19516 --- [nPool-worker-11] c.t.s.e.core.GenericDataProcessor : Fail to process event, handler:DLQGenericEventHandlerWrapper java.lang.RuntimeException: retryCount值未更新！ at com.tplink.smb.eventcenter.port.kafka.deadletter.DLQGenericEventHandlerWrapper.handleEvent(DLQGenericEventHandlerWrapper.java:56) ~[eventcenter.kafka-1.4.5002-test-SNAPSHOT.jar:1.4.5002-test-SNAPSHOT] at com.tplink.smb.eventcenter.port.kafka.deadletter.DLQGenericEventHandlerWrapper.handleEvent(DLQGenericEventHandlerWrapper.java:24) ~[eventcenter.kafka-1.4.5002-test-SNAPSHOT.jar:1.4.5002-test-SNAPSHOT] at com.tplink.smb.eventcenter.core.GenericDataProcessor.run(GenericDataProcessor.java:32) ~[eventcenter.core-1.4.5002-test-SNAPSHOT.jar:1.4.5002-test-SNAPSHOT] at java.util.concurrent.ForkJoinTask$AdaptedRunnableAction.exec(ForkJoinTask.java:1386) [na:1.8.0_462-462] at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) [na:1.8.0_462-462] at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056) [na:1.8.0_462-462] at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) [na:1.8.0_462-462] at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175) [na:1.8.0_462-462] 2025-10-17 10:04:47.450 INFO 19516 --- [nPool-worker-11] c.t.n.d.b.DLQGenericValidationApp : 处理Event事件: {"retryCount":1,"filterKey":"key1","timeStamp":0,"message":"Hello Event!"} 2025-10-17 10:04:47.450 INFO 19516 --- [nPool-worker-11] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=1, 到期时间expirationTime=1760666688450 2025-10-17 10:04:47.451 INFO 19516 --- [r_group_level_1] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际暂停成功，topic: delay_topic_level_1 2025-10-17 10:04:47.452 INFO 19516 --- [nPool-worker-22] c.t.n.d.b.DLQGenericValidationApp : 处理Event事件: {"retryCount":0,"filterKey":"key1","timeStamp":0,"message":"Hello Event!"} 2025-10-17 10:04:47.452 INFO 19516 --- [nPool-worker-22] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=3, 到期时间expirationTime=1760666697452 2025-10-17 10:04:47.454 INFO 19516 --- [r_group_level_3] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际暂停成功，topic: delay_topic_level_3 2025-10-17 10:04:48.443 INFO 19516 --- [pool-2-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_1, currentTime=1760666688443 2025-10-17 10:04:48.474 INFO 19516 --- [r_group_level_1] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际恢复成功，topic: delay_topic_level_1 2025-10-17 10:04:48.475 INFO 19516 --- [pool-2-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-eventv2-topic 2025-10-17 10:04:48.478 INFO 19516 --- [nPool-worker-25] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-event-topic 2025-10-17 10:04:48.479 INFO 19516 --- [nPool-worker-25] c.t.n.d.b.DLQGenericValidationApp : 处理EventV2事件: Hello EventV2! 2025-10-17 10:04:48.479 INFO 19516 --- [nPool-worker-25] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=2, 到期时间expirationTime=1760666693479 2025-10-17 10:04:48.479 INFO 19516 --- [nPool-worker-22] c.t.n.d.b.DLQGenericValidationApp : 处理Event事件: {"retryCount":2,"filterKey":"key1","timeStamp":0,"message":"Hello Event!"} 2025-10-17 10:04:48.479 INFO 19516 --- [nPool-worker-22] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=2, 到期时间expirationTime=1760666693479 2025-10-17 10:04:48.493 INFO 19516 --- [r_group_level_2] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际暂停成功，topic: delay_topic_level_2 2025-10-17 10:04:53.482 INFO 19516 --- [pool-5-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_2, currentTime=1760666693482 2025-10-17 10:04:53.482 INFO 19516 --- [pool-5-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-eventv2-topic 2025-10-17 10:04:53.482 INFO 19516 --- [pool-5-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_2, currentTime=1760666693482 2025-10-17 10:04:53.500 INFO 19516 --- [nPool-worker-22] c.t.n.d.b.DLQGenericValidationApp : 处理EventV2事件: Hello EventV2! 2025-10-17 10:04:53.500 INFO 19516 --- [nPool-worker-22] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=3, 到期时间expirationTime=1760666703500 2025-10-17 10:04:53.608 INFO 19516 --- [r_group_level_2] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际恢复成功，topic: delay_topic_level_2 2025-10-17 10:04:53.610 INFO 19516 --- [pool-5-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-event-topic 2025-10-17 10:04:53.615 INFO 19516 --- [nPool-worker-22] c.t.n.d.b.DLQGenericValidationApp : 处理Event事件: {"retryCount":3,"filterKey":"key1","timeStamp":0,"message":"Hello Event!"} 2025-10-17 10:04:53.616 INFO 19516 --- [nPool-worker-22] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=3, 到期时间expirationTime=1760666703616 2025-10-17 10:04:57.431 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_3, currentTime=1760666697431 2025-10-17 10:04:57.679 INFO 19516 --- [r_group_level_3] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际恢复成功，topic: delay_topic_level_3 2025-10-17 10:04:57.680 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-eventv2-topic 2025-10-17 10:04:57.681 INFO 19516 --- [nPool-worker-22] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-event-topic 2025-10-17 10:04:57.684 INFO 19516 --- [nPool-worker-25] c.t.n.d.b.DLQGenericValidationApp : 处理EventV2事件: Hello EventV2! 2025-10-17 10:04:57.684 INFO 19516 --- [nPool-worker-25] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=1, 到期时间expirationTime=1760666698684 2025-10-17 10:04:57.684 INFO 19516 --- [onPool-worker-4] c.t.n.d.b.DLQGenericValidationApp : 处理Event事件: {"retryCount":1,"filterKey":"key1","timeStamp":0,"message":"Hello Event!"} 2025-10-17 10:04:57.684 INFO 19516 --- [onPool-worker-4] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=1, 到期时间expirationTime=1760666698684 2025-10-17 10:04:57.685 INFO 19516 --- [r_group_level_3] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际暂停成功，topic: delay_topic_level_3 2025-10-17 10:04:57.686 INFO 19516 --- [r_group_level_1] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际暂停成功，topic: delay_topic_level_1 2025-10-17 10:04:58.694 INFO 19516 --- [pool-2-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_1, currentTime=1760666698694 2025-10-17 10:04:58.709 INFO 19516 --- [r_group_level_1] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际恢复成功，topic: delay_topic_level_1 2025-10-17 10:04:58.710 INFO 19516 --- [pool-2-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-eventv2-topic 2025-10-17 10:04:58.712 INFO 19516 --- [nPool-worker-22] c.t.n.d.b.DLQGenericValidationApp : 处理EventV2事件: Hello EventV2! 2025-10-17 10:04:58.712 INFO 19516 --- [nPool-worker-22] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=2, 到期时间expirationTime=1760666703712 2025-10-17 10:04:58.713 INFO 19516 --- [onPool-worker-4] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-event-topic 2025-10-17 10:04:58.728 INFO 19516 --- [nPool-worker-22] c.t.n.d.b.DLQGenericValidationApp : 处理Event事件: {"retryCount":2,"filterKey":"key1","timeStamp":0,"message":"Hello Event!"} 2025-10-17 10:04:58.728 INFO 19516 --- [nPool-worker-22] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=2, 到期时间expirationTime=1760666703728 2025-10-17 10:04:58.729 INFO 19516 --- [r_group_level_2] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际暂停成功，topic: delay_topic_level_2 2025-10-17 10:05:03.508 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_3, currentTime=1760666703508 2025-10-17 10:05:03.508 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-eventv2-topic 2025-10-17 10:05:03.529 INFO 19516 --- [ad | producer-1] .s.e.p.k.d.DLQGenericEventHandlerWrapper : 死信发送成功 | 主题: test-eventv2-topic_dlq_topic, 分区: 0, 重试次数: 4, 异常信息: null 2025-10-17 10:05:03.530 INFO 19516 --- [onPool-worker-4] c.t.n.d.b.DLQGenericValidationApp : 收到EventV2死信消息class com.tplink.smb.eventcenter.api.EventV2 2025-10-17 10:05:03.631 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_3, currentTime=1760666703631 2025-10-17 10:05:03.723 INFO 19516 --- [pool-5-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_2, currentTime=1760666703723 2025-10-17 10:05:03.817 INFO 19516 --- [r_group_level_3] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际恢复成功，topic: delay_topic_level_3 2025-10-17 10:05:03.817 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-event-topic 2025-10-17 10:05:03.836 INFO 19516 --- [ad | producer-1] .s.e.p.k.d.DLQGenericEventHandlerWrapper : 死信发送成功 | 主题: test-event-topic_dlq_topic, 分区: 0, 重试次数: 4, 异常信息: null 2025-10-17 10:05:03.836 INFO 19516 --- [onPool-worker-4] c.t.n.d.b.DLQGenericValidationApp : 收到Event死信消息Hello Event! 2025-10-17 10:05:03.864 INFO 19516 --- [r_group_level_2] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际恢复成功，topic: delay_topic_level_2 2025-10-17 10:05:03.864 INFO 19516 --- [pool-5-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-eventv2-topic 2025-10-17 10:05:03.865 INFO 19516 --- [onPool-worker-4] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-event-topic 2025-10-17 10:05:03.866 INFO 19516 --- [onPool-worker-4] c.t.n.d.b.DLQGenericValidationApp : 处理EventV2事件: Hello EventV2! 2025-10-17 10:05:03.868 INFO 19516 --- [onPool-worker-4] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=3, 到期时间expirationTime=1760666713868 2025-10-17 10:05:03.868 INFO 19516 --- [nPool-worker-22] c.t.n.d.b.DLQGenericValidationApp : 处理Event事件: {"retryCount":3,"filterKey":"key1","timeStamp":0,"message":"Hello Event!"} 2025-10-17 10:05:03.868 INFO 19516 --- [nPool-worker-22] c.t.s.e.port.kafka.KafkaEventCenter : 延迟等级level=3, 到期时间expirationTime=1760666713868 2025-10-17 10:05:03.882 INFO 19516 --- [r_group_level_3] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际暂停成功，topic: delay_topic_level_3 2025-10-17 10:05:13.877 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_3, currentTime=1760666713877 2025-10-17 10:05:13.878 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-event-topic 2025-10-17 10:05:13.878 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.delay.DelayHandler : 延迟任务执行：targetTopic=delay_topic_level_3, currentTime=1760666713878 2025-10-17 10:05:13.911 INFO 19516 --- [ad | producer-1] .s.e.p.k.d.DLQGenericEventHandlerWrapper : 死信发送成功 | 主题: test-event-topic_dlq_topic, 分区: 0, 重试次数: 4, 异常信息: null 2025-10-17 10:05:13.911 INFO 19516 --- [nPool-worker-22] c.t.n.d.b.DLQGenericValidationApp : 收到Event死信消息Hello Event! 2025-10-17 10:05:14.126 INFO 19516 --- [r_group_level_3] c.t.s.e.p.k.c.GenericKafkaConsumerTask : 实际恢复成功，topic: delay_topic_level_3 2025-10-17 10:05:14.126 INFO 19516 --- [pool-8-thread-1] c.t.s.e.port.kafka.KafkaEventCenter : 延迟事件已转发，targetTopic: test-eventv2-topic 2025-10-17 10:05:14.158 INFO 19516 --- [ad | producer-1] .s.e.p.k.d.DLQGenericEventHandlerWrapper : 死信发送成功 | 主题: test-eventv2-topic_dlq_topic, 分区: 0, 重试次数: 4, 异常信息: null 2025-10-17 10:05:14.158 INFO 19516 --- [nPool-worker-22] c.t.n.d.b.DLQGenericValidationApp : 收到EventV2死信消息class com.tplink.smb.eventcenter.api.EventV2@Slf4j @RequiredArgsConstructor public class DLQEventHandlerWrapper implements EventHandler { // 原事件处理器 private final EventHandler delegate; // DLQ配置 private final DLQConfig dlqConfig; // 事件中心（用于发送死信） private final String mainTopic; // 当前主Topic private final KafkaEventCenter eventCenter; private String lastException; /** * 处理事件，包含重试与死信发送逻辑。 *

核心逻辑： * 1. 若DLQ未启用，直接调用原始处理器； * 2. 若启用，按配置重试处理事件； * 3. 重试耗尽后发送事件到DLQ。 * * @param event 待处理的事件 */ @Override public void handleEvent(Event event) { if (!dlqConfig.isEnabled()) { delegate.handleEvent(event); return; } int retryCount = event.getRetryCount(); // 达到阈值直接进入死信 if (retryCount > dlqConfig.getMaxRetries()) { sendToDLQ(event, retryCount, lastException, mainTopic + "_dlq_topic"); return; } try { delegate.handleEvent(event); } catch (Exception e) { int newRetryCount = retryCount + 1; event.setRetryCount(newRetryCount); if (newRetryCount > dlqConfig.getMaxRetries()) { sendToDLQ(event, newRetryCount, e.getMessage(), mainTopic + "_dlq_topic"); } else { long delay = dlqConfig.getRetryStrategy().getNextDelay(newRetryCount); eventCenter.sendDelay(mainTopic, event, delay, new FlexibleMatchStrategy(DelayQueueStrategy.ApproximationMode.UP)); log.info("事件延时 {}ms 后重试", delay); } } } /** * 将事件发送至死信队列 */ private void sendToDLQ(Event event,int retryCount,String lastException,String dlqTopic) { // 构造自定义EventFuture回调，捕获发送结果 EventFuture dlqEventFuture = new EventFuture() { @Override public void onSuccess(EventCenterSendResult result) { log.info("发送至死信队列！原始事件: {},时间戳：{}，主题: {}, 分区: {}, 已重试次数: {},最后一次异常信息：{}", event.toString(),result.getTimestamp(),result.getTopic(), result.getPartition(), retryCount - 1 , lastException); } @Override public void onFailure(Throwable throwable) { // 关键：记录重试次数和发送异常 log.error("死信消息发送失败！原始事件: {}, 已重试次数: {}, 异常信息: {}", event.toString(), retryCount - 1 , throwable.getMessage(), throwable); } }; try { // 调用带回调的send方法（匹配参数：topic, event, eventFuture） eventCenter.send(dlqTopic, event, dlqEventFuture); } catch (Exception e) { // 处理send方法本身的异常（如参数错误） log.error("调用死信发送方法时发生异常", e); } } }@Slf4j @SpringBootApplication(exclude = {DataSourceAutoConfiguration.class, DataSourceTransactionManagerAutoConfiguration.class}) @ComponentScan(basePackages = { "com.tplink.nbu.demo.basicspringboot", "com.tplink.smb.eventcenter.port.kafka.deadletter", "com.tplink.smb.eventcenter.api.config" }) public class DLQGenericValidationApp implements CommandLineRunner { @Autowired private KafkaEventCenter eventCenter; @Autowired private DLQConfig deadLetterConfig; // 定义三个主Topic及对应的死信Topic private static final String EVENT_TOPIC = "test-event-topic"; private static final String EVENTV2_TOPIC = "test-eventv2-topic"; private static final String CUSTOM_TOPIC = "test-custom-topic"; private static final String EVENT_DLQ_TOPIC = EVENT_TOPIC + "_dlq_topic"; private static final String EVENTV2_DLQ_TOPIC = EVENTV2_TOPIC + "_dlq_topic"; private static final String CUSTOM_DLQ_TOPIC = CUSTOM_TOPIC + "_dlq_topic"; public static void main(String[] args) { SpringApplication.run(DLQGenericValidationApp.class, args); } @Override public void run(String... args) throws Exception { eventCenter.registerDelayConsumer(); // 步骤1：为三类事件分别注册带DLQ的泛型消费者 registerEventConsumer(); registerEventV2Consumer(); registerCustomEventConsumer(); // 步骤2：为三个死信Topic分别注册监听 registerEventDLQListener(); registerEventV2DLQListener(); registerCustomDLQListener(); // 步骤3：发送三类测试事件到各自主Topic sendTestEvents(); } /** * 注册Event类型的消费者（主Topic: EVENT_TOPIC） */ private void registerEventConsumer() { GenericEventHandler<Event> eventHandler = event -> { log.info("处理Event事件: {}", event); throw new RuntimeException("模拟Event处理失败"); // 主动抛异常触发重试 }; eventCenter.registerUnicastGenerically( EVENT_TOPIC, // Event主Topic "event-consumer-group", eventHandler, ForkJoinPool.commonPool(), PartitionAssignorMode.COOPERATIVE_STICKY, Event.class, deadLetterConfig ); } /** * 注册EventV2类型的消费者（主Topic: EVENTV2_TOPIC） */ private void registerEventV2Consumer() { GenericEventHandler<EventV2> eventV2Handler = event -> { log.info("处理EventV2事件: {}", event.getMessage()); throw new RuntimeException("模拟EventV2处理失败"); // 主动抛异常触发重试 }; eventCenter.registerUnicast( EVENTV2_TOPIC, // EventV2主Topic "eventv2-consumer-group", // EventV2消费者组 eventV2Handler, ForkJoinPool.commonPool(), PartitionAssignorMode.COOPERATIVE_STICKY, SerializeEnum.KRYO, deadLetterConfig ); } /** * 注册CustomEvent类型的消费者（主Topic: CUSTOM_TOPIC） */ private void registerCustomEventConsumer() { GenericEventHandler<CustomEvent> customHandler = event -> { log.info("处理CustomEvent事件"); throw new RuntimeException("模拟CustomEvent处理失败"); // 主动抛异常触发重试 }; eventCenter.registerUnicastGenerically( CUSTOM_TOPIC, // Custom主Topic "custom-consumer-group", // Custom消费者组 customHandler, ForkJoinPool.commonPool(), PartitionAssignorMode.COOPERATIVE_STICKY, CustomEvent.class, deadLetterConfig ); } /** * 注册Event死信队列监听（验证Event死信） */ private void registerEventDLQListener() { eventCenter.registerUnicast( EVENT_DLQ_TOPIC, // Event死信Topic "event-dlq-group", // Event死信消费者组 event-> log.info("收到Event死信消息{}",event.getMessage()), // 明确Event类型 ForkJoinPool.commonPool()// 与Event发送时的序列化方式一致（假设Event用JSON） ); } /** * 注册EventV2死信队列监听（验证EventV2死信） */ private void registerEventV2DLQListener() { eventCenter.registerUnicast( EVENTV2_DLQ_TOPIC, // EventV2死信Topic "eventv2-dlq-group", // EventV2死信消费者组 event -> log.info("收到EventV2死信消息{}",event.getClass()), ForkJoinPool.commonPool(), SerializeEnum.KRYO ); } /** * 注册CustomEvent死信队列监听（验证CustomEvent死信） */ private void registerCustomDLQListener() { eventCenter.registerUnicastGenerically( CUSTOM_DLQ_TOPIC, // Custom死信Topic "custom-dlq-group", // Custom死信消费者组 event -> log.info("收到Custom死信消息{}",event.content), ForkJoinPool.commonPool(), CustomEvent.class ); } /** * 发送三类测试事件到各自主Topic */ private void sendTestEvents() { // 1. 发送Event到EVENT_TOPIC（JSON序列化） Event event = new Event("key1", "Hello Event!"); eventCenter.sendGenerically(EVENT_TOPIC, event, buildTestFuture("Event")); // 2. 发送EventV2到EVENTV2_TOPIC（Kryo序列化） EventV2<String> eventV2 = new EventV2<>("key2", "Hello EventV2!"); eventCenter.send(EVENTV2_TOPIC, eventV2, buildTestFuture("EventV2"),SerializeEnum.KRYO); // 3. 发送CustomEvent到CUSTOM_TOPIC（JSON序列化） CustomEvent customEvent = new CustomEvent("key3", "Hello CustomEvent!"); eventCenter.sendGenerically(CUSTOM_TOPIC, customEvent, buildTestFuture("CustomEvent")); } /** * 构建测试用的GenericEventFuture */ private GenericEventFuture buildTestFuture(String eventType) { return new GenericEventFuture () { @Override public void onSuccess(GenericEventCenterSendResult result) { log.info("[{}] 发送成功 | 主题: {}, 分区: {}", eventType, result.getTopic(), result.getPartition()); } @Override public void onFailure(Throwable throwable) { log.error("[{}] 发送失败 | 原因: {}", eventType, throwable.getMessage()); } }; }请分析，为什么泛型发送测试时，异常显示为null

核心逻辑： * 1. 若DLQ未启用，直接调用原始处理器； * 2. 若启用，按配置重试处理事件； * 3. 重试耗尽后发送事件到DLQ。 * * @param event 待处理的事件 */ @Override public void handleEvent(Event event) { if (!dlqConfig.isEnabled()) { delegate.handleEvent(event); return; } int retryCount = event.getRetryCount(); // 达到阈值直接进入死信 if (retryCount > dlqConfig.getMaxRetries()) { sendToDLQ(event, retryCount, lastException, mainTopic + "_dlq_topic"); return; } try { delegate.handleEvent(event); } catch (Exception e) { int newRetryCount = retryCount + 1; event.setRetryCount(newRetryCount); if (newRetryCount > dlqConfig.getMaxRetries()) { sendToDLQ(event, newRetryCount, e.getMessage(), mainTopic + "_dlq_topic"); } else { long delay = dlqConfig.getRetryStrategy().getNextDelay(newRetryCount); eventCenter.sendDelay(mainTopic, event, delay, new FlexibleMatchStrategy(DelayQueueStrategy.ApproximationMode.UP)); log.info("事件延时 {}ms 后重试", delay); } } } /** * 将事件发送至死信队列 */ private void sendToDLQ(Event event,int retryCount,String lastException,String dlqTopic) { // 构造自定义EventFuture回调，捕获发送结果 EventFuture dlqEventFuture = new EventFuture() { @Override public void onSuccess(EventCenterSendResult result) { log.info("发送至死信队列！原始事件: {},时间戳：{}，主题: {}, 分区: {}, 已重试次数: {},最后一次异常信息：{}", event.toString(),result.getTimestamp(),result.getTopic(), result.getPartition(), retryCount - 1 , lastException); } @Override public void onFailure(Throwable throwable) { // 关键：记录重试次数和发送异常 log.error("死信消息发送失败！原始事件: {}, 已重试次数: {}, 异常信息: {}", event.toString(), retryCount - 1 , throwable.getMessage(), throwable); } }; try { // 调用带回调的send方法（匹配参数：topic, event, eventFuture） eventCenter.send(dlqTopic, event, dlqEventFuture); } catch (Exception e) { // 处理send方法本身的异常（如参数错误） log.error("调用死信发送方法时发生异常", e); } } }@Slf4j @SpringBootApplication(exclude = {DataSourceAutoConfiguration.class, DataSourceTransactionManagerAutoConfiguration.class}) @ComponentScan(basePackages = { "com.tplink.nbu.demo.basicspringboot", "com.tplink.smb.eventcenter.port.kafka.deadletter", "com.tplink.smb.eventcenter.api.config" }) public class DLQGenericValidationApp implements CommandLineRunner { @Autowired private KafkaEventCenter eventCenter; @Autowired private DLQConfig deadLetterConfig; // 定义三个主Topic及对应的死信Topic private static final String EVENT_TOPIC = "test-event-topic"; private static final String EVENTV2_TOPIC = "test-eventv2-topic"; private static final String CUSTOM_TOPIC = "test-custom-topic"; private static final String EVENT_DLQ_TOPIC = EVENT_TOPIC + "_dlq_topic"; private static final String EVENTV2_DLQ_TOPIC = EVENTV2_TOPIC + "_dlq_topic"; private static final String CUSTOM_DLQ_TOPIC = CUSTOM_TOPIC + "_dlq_topic"; public static void main(String[] args) { SpringApplication.run(DLQGenericValidationApp.class, args); } @Override public void run(String... args) throws Exception { eventCenter.registerDelayConsumer(); // 步骤1：为三类事件分别注册带DLQ的泛型消费者 registerEventConsumer(); registerEventV2Consumer(); registerCustomEventConsumer(); // 步骤2：为三个死信Topic分别注册监听 registerEventDLQListener(); registerEventV2DLQListener(); registerCustomDLQListener(); // 步骤3：发送三类测试事件到各自主Topic sendTestEvents(); } /** * 注册Event类型的消费者（主Topic: EVENT_TOPIC） */ private void registerEventConsumer() { GenericEventHandler<Event> eventHandler = event -> { log.info("处理Event事件: {}", event); throw new RuntimeException("模拟Event处理失败"); // 主动抛异常触发重试 }; eventCenter.registerUnicastGenerically( EVENT_TOPIC, // Event主Topic "event-consumer-group", eventHandler, ForkJoinPool.commonPool(), PartitionAssignorMode.COOPERATIVE_STICKY, Event.class, deadLetterConfig ); } /** * 注册EventV2类型的消费者（主Topic: EVENTV2_TOPIC） */ private void registerEventV2Consumer() { GenericEventHandler<EventV2> eventV2Handler = event -> { log.info("处理EventV2事件: {}", event.getMessage()); throw new RuntimeException("模拟EventV2处理失败"); // 主动抛异常触发重试 }; eventCenter.registerUnicast( EVENTV2_TOPIC, // EventV2主Topic "eventv2-consumer-group", // EventV2消费者组 eventV2Handler, ForkJoinPool.commonPool(), PartitionAssignorMode.COOPERATIVE_STICKY, SerializeEnum.KRYO, deadLetterConfig ); } /** * 注册CustomEvent类型的消费者（主Topic: CUSTOM_TOPIC） */ private void registerCustomEventConsumer() { GenericEventHandler<CustomEvent> customHandler = event -> { log.info("处理CustomEvent事件"); throw new RuntimeException("模拟CustomEvent处理失败"); // 主动抛异常触发重试 }; eventCenter.registerUnicastGenerically( CUSTOM_TOPIC, // Custom主Topic "custom-consumer-group", // Custom消费者组 customHandler, ForkJoinPool.commonPool(), PartitionAssignorMode.COOPERATIVE_STICKY, CustomEvent.class, deadLetterConfig ); } /** * 注册Event死信队列监听（验证Event死信） */ private void registerEventDLQListener() { eventCenter.registerUnicast( EVENT_DLQ_TOPIC, // Event死信Topic "event-dlq-group", // Event死信消费者组 event-> log.info("收到Event死信消息{}",event.getMessage()), // 明确Event类型 ForkJoinPool.commonPool()// 与Event发送时的序列化方式一致（假设Event用JSON） ); } /** * 注册EventV2死信队列监听（验证EventV2死信） */ private void registerEventV2DLQListener() { eventCenter.registerUnicast( EVENTV2_DLQ_TOPIC, // EventV2死信Topic "eventv2-dlq-group", // EventV2死信消费者组 event -> log.info("收到EventV2死信消息{}",event.getClass()), ForkJoinPool.commonPool(), SerializeEnum.KRYO ); } /** * 注册CustomEvent死信队列监听（验证CustomEvent死信） */ private void registerCustomDLQListener() { eventCenter.registerUnicastGenerically( CUSTOM_DLQ_TOPIC, // Custom死信Topic "custom-dlq-group", // Custom死信消费者组 event -> log.info("收到Custom死信消息{}",event.content), ForkJoinPool.commonPool(), CustomEvent.class ); } /** * 发送三类测试事件到各自主Topic */ private void sendTestEvents() { // 1. 发送Event到EVENT_TOPIC（JSON序列化） Event event = new Event("key1", "Hello Event!"); eventCenter.sendGenerically(EVENT_TOPIC, event, buildTestFuture("Event")); // 2. 发送EventV2到EVENTV2_TOPIC（Kryo序列化） EventV2<String> eventV2 = new EventV2<>("key2", "Hello EventV2!"); eventCenter.send(EVENTV2_TOPIC, eventV2, buildTestFuture("EventV2"),SerializeEnum.KRYO); // 3. 发送CustomEvent到CUSTOM_TOPIC（JSON序列化） CustomEvent customEvent = new CustomEvent("key3", "Hello CustomEvent!"); eventCenter.sendGenerically(CUSTOM_TOPIC, customEvent, buildTestFuture("CustomEvent")); } /** * 构建测试用的GenericEventFuture */ private GenericEventFuture buildTestFuture(String eventType) { return new GenericEventFuture () { @Override public void onSuccess(GenericEventCenterSendResult result) { log.info("[{}] 发送成功 | 主题: {}, 分区: {}", eventType, result.getTopic(), result.getPartition()); } @Override public void onFailure(Throwable throwable) { log.error("[{}] 发送失败 | 原因: {}", eventType, throwable.getMessage()); } }; }请分析，为什么泛型发送测试时，异常显示为null

本文标签： 事件 编程 系统