From e0fc7c87b5fce6c43259b9fd574d1ba39884f73c Mon Sep 17 00:00:00 2001 From: lapuda <577732344@qq.com> Date: Tue, 21 Jul 2026 17:43:04 +0800 Subject: [PATCH] feat: refresh es after external import commit --- .../ExternalJobImportAsyncConfig.java | 31 ++++++++++++++++++ .../ExternalJobImportEsRefreshListener.java | 22 +++++++++++++ .../ExternalJobImportEsRefreshService.java | 32 +++++++++++++++++++ .../ExternalJobImportServiceImpl.java | 8 ++++- .../ExternalJobImportSucceededEvent.java | 18 +++++++++++ ...xternalJobImportEsRefreshListenerTest.java | 19 +++++++++++ ...ExternalJobImportEsRefreshServiceTest.java | 20 ++++++++++++ 7 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportAsyncConfig.java create mode 100644 ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshListener.java create mode 100644 ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshService.java create mode 100644 ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportSucceededEvent.java create mode 100644 ruoyi-bussiness/src/test/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshListenerTest.java create mode 100644 ruoyi-bussiness/src/test/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshServiceTest.java diff --git a/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportAsyncConfig.java b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportAsyncConfig.java new file mode 100644 index 0000000..731fe56 --- /dev/null +++ b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportAsyncConfig.java @@ -0,0 +1,31 @@ +package com.ruoyi.cms.externalimport; + +import java.util.concurrent.ThreadPoolExecutor; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +/** + * 外部岗位导入后置任务的独立执行器。 + * + * 全量重建会删除并创建 job_document 索引，必须串行执行，不能占用通用业务线程池。 + */ +@Configuration +@EnableAsync +public class ExternalJobImportAsyncConfig { + + @Bean(name = "externalJobImportEsRefreshExecutor") + public ThreadPoolTaskExecutor externalJobImportEsRefreshExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(1); + executor.setMaxPoolSize(1); + executor.setQueueCapacity(10); + executor.setKeepAliveSeconds(60); + executor.setThreadNamePrefix("external-job-es-refresh-"); + // 队列满时在提交后的监听线程执行，确保成功批次最终一定会请求索引刷新。 + executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); + return executor; + } +} diff --git a/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshListener.java b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshListener.java new file mode 100644 index 0000000..0f29e34 --- /dev/null +++ b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshListener.java @@ -0,0 +1,22 @@ +package com.ruoyi.cms.externalimport; + +import org.springframework.stereotype.Component; +import org.springframework.transaction.event.TransactionPhase; +import org.springframework.transaction.event.TransactionalEventListener; + +/** + * 在外部岗位导入事务提交后安排 ES 全量刷新。 + */ +@Component +public class ExternalJobImportEsRefreshListener { + private final ExternalJobImportEsRefreshService refreshService; + + public ExternalJobImportEsRefreshListener(ExternalJobImportEsRefreshService refreshService) { + this.refreshService = refreshService; + } + + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + public void onImportSucceeded(ExternalJobImportSucceededEvent event) { + refreshService.refreshAfterCommitAsync(event.getBatchId()); + } +} diff --git a/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshService.java b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshService.java new file mode 100644 index 0000000..303bc0b --- /dev/null +++ b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshService.java @@ -0,0 +1,32 @@ +package com.ruoyi.cms.externalimport; + +import com.ruoyi.cms.service.IESJobSearchService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.scheduling.annotation.Async; +import org.springframework.stereotype.Service; + +/** + * 外部导入提交后的 ES 全量重建执行器。 + */ +@Service +public class ExternalJobImportEsRefreshService { + private static final Logger LOGGER = LoggerFactory.getLogger(ExternalJobImportEsRefreshService.class); + + private final IESJobSearchService esJobSearchService; + + public ExternalJobImportEsRefreshService(IESJobSearchService esJobSearchService) { + this.esJobSearchService = esJobSearchService; + } + + @Async("externalJobImportEsRefreshExecutor") + public void refreshAfterCommitAsync(String batchId) { + LOGGER.info("外部岗位导入批次 {} 已提交，开始异步全量重建 ES 岗位索引", batchId); + try { + esJobSearchService.resetTextCache(); + LOGGER.info("外部岗位导入批次 {} 的 ES 岗位索引异步重建完成", batchId); + } catch (RuntimeException ex) { + LOGGER.error("外部岗位导入批次 {} 的 ES 岗位索引异步重建失败", batchId, ex); + } + } +} diff --git a/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportServiceImpl.java b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportServiceImpl.java index e377563..c22a2f5 100644 --- a/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportServiceImpl.java +++ b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportServiceImpl.java @@ -2,6 +2,7 @@ package com.ruoyi.cms.externalimport; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.support.GeneratedKeyHolder; import org.springframework.jdbc.support.KeyHolder; @@ -40,16 +41,19 @@ public class ExternalJobImportServiceImpl implements IExternalJobImportService { private final ExternalUploadClient uploadClient; private final JdbcTemplate jdbcTemplate; private final ObjectMapper objectMapper; + private final ApplicationEventPublisher eventPublisher; private final ExternalJobExcelParser parser = new ExternalJobExcelParser(); public ExternalJobImportServiceImpl(ExternalJobImportProperties properties, ExternalUploadClient uploadClient, JdbcTemplate jdbcTemplate, - ObjectMapper objectMapper) { + ObjectMapper objectMapper, + ApplicationEventPublisher eventPublisher) { this.properties = properties; this.uploadClient = uploadClient; this.jdbcTemplate = jdbcTemplate; this.objectMapper = objectMapper; + this.eventPublisher = eventPublisher; } @Override @@ -158,6 +162,8 @@ public class ExternalJobImportServiceImpl implements IExternalJobImportService { result.setStatus("SUCCESS"); result.setMessage("外部岗位全量快照发布成功"); updateBatch(batchId, "SUCCESS", result, null); + // 仅在事务成功提交后异步全量刷新 ES，避免索引读到未提交或最终回滚的岗位数据。 + eventPublisher.publishEvent(new ExternalJobImportSucceededEvent(batchId)); return result; } diff --git a/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportSucceededEvent.java b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportSucceededEvent.java new file mode 100644 index 0000000..8a5fb18 --- /dev/null +++ b/ruoyi-bussiness/src/main/java/com/ruoyi/cms/externalimport/ExternalJobImportSucceededEvent.java @@ -0,0 +1,18 @@ +package com.ruoyi.cms.externalimport; + +/** + * 外部岗位全量快照已成功落库的领域事件。 + * + * 事件由事务监听器在提交后处理，因此不会为回滚、阻断或重复跳过的批次刷新 ES。 + */ +public class ExternalJobImportSucceededEvent { + private final String batchId; + + public ExternalJobImportSucceededEvent(String batchId) { + this.batchId = batchId; + } + + public String getBatchId() { + return batchId; + } +} diff --git a/ruoyi-bussiness/src/test/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshListenerTest.java b/ruoyi-bussiness/src/test/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshListenerTest.java new file mode 100644 index 0000000..17f8b45 --- /dev/null +++ b/ruoyi-bussiness/src/test/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshListenerTest.java @@ -0,0 +1,19 @@ +package com.ruoyi.cms.externalimport; + +import org.junit.jupiter.api.Test; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +class ExternalJobImportEsRefreshListenerTest { + + @Test + void delegatesCommittedBatchToAsyncRefreshService() { + ExternalJobImportEsRefreshService refreshService = mock(ExternalJobImportEsRefreshService.class); + ExternalJobImportEsRefreshListener listener = new ExternalJobImportEsRefreshListener(refreshService); + + listener.onImportSucceeded(new ExternalJobImportSucceededEvent("batch-20260721")); + + verify(refreshService).refreshAfterCommitAsync("batch-20260721"); + } +} diff --git a/ruoyi-bussiness/src/test/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshServiceTest.java b/ruoyi-bussiness/src/test/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshServiceTest.java new file mode 100644 index 0000000..0546b0e --- /dev/null +++ b/ruoyi-bussiness/src/test/java/com/ruoyi/cms/externalimport/ExternalJobImportEsRefreshServiceTest.java @@ -0,0 +1,20 @@ +package com.ruoyi.cms.externalimport; + +import com.ruoyi.cms.service.IESJobSearchService; +import org.junit.jupiter.api.Test; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +class ExternalJobImportEsRefreshServiceTest { + + @Test + void rebuildsJobIndexForCommittedBatch() { + IESJobSearchService searchService = mock(IESJobSearchService.class); + ExternalJobImportEsRefreshService refreshService = new ExternalJobImportEsRefreshService(searchService); + + refreshService.refreshAfterCommitAsync("batch-20260721"); + + verify(searchService).resetTextCache(); + } +}