package jnpf.dmsService.search;
|
|
import lombok.RequiredArgsConstructor;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.DisposableBean;
|
import org.springframework.beans.factory.InitializingBean;
|
import org.springframework.stereotype.Component;
|
|
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.Executors;
|
import java.util.concurrent.TimeUnit;
|
|
@Slf4j
|
@Component
|
@RequiredArgsConstructor
|
public class DmsSearchIndexWorker implements InitializingBean, DisposableBean {
|
private final DmsSearchProperties properties;
|
private final DmsSearchIndexQueue queue;
|
private final DmsSearchIndexProcessor processor;
|
|
private volatile boolean running;
|
private ExecutorService executor;
|
|
@Override
|
public void afterPropertiesSet() {
|
if (!properties.isEnabled()) {
|
return;
|
}
|
running = true;
|
executor = Executors.newSingleThreadExecutor(runnable -> {
|
Thread thread = new Thread(runnable, "dms-search-index-worker");
|
thread.setDaemon(true);
|
return thread;
|
});
|
executor.submit(this::run);
|
}
|
|
private void run() {
|
while (running && !Thread.currentThread().isInterrupted()) {
|
DmsSearchIndexTask task = null;
|
try {
|
task = queue.take();
|
if (task != null) {
|
processor.process(task);
|
}
|
} catch (Exception exception) {
|
retry(task, exception);
|
}
|
}
|
}
|
|
private void retry(DmsSearchIndexTask task, Exception exception) {
|
if (task == null) {
|
log.error("读取全文检索 Redis 队列失败", exception);
|
return;
|
}
|
task.setAttempts(task.getAttempts() + 1);
|
if (task.getAttempts() < Math.max(1, properties.getMaxAttempts())) {
|
log.warn("全文检索任务处理失败，重新入队: fileVersionId={}, attempts={}",
|
task.getFileVersionId(), task.getAttempts(), exception);
|
queue.push(task);
|
} else {
|
log.error("全文检索任务处理失败，已达到最大重试次数: fileVersionId={}",
|
task.getFileVersionId(), exception);
|
}
|
}
|
|
@Override
|
public void destroy() throws InterruptedException {
|
running = false;
|
if (executor != null) {
|
executor.shutdownNow();
|
executor.awaitTermination(5, TimeUnit.SECONDS);
|
}
|
}
|
}
|
