把python代码转为Java， 涉及功能主要是音频转文本
This commit is contained in:
@@ -80,6 +80,7 @@ public class AudioStatisticsScheduler {
|
||||
return;
|
||||
}
|
||||
if (System.getProperty("os.name", "").toLowerCase(Locale.ROOT).contains("win")) {
|
||||
log.info("syncYihangyiTranscriptTextFiles 跳过：Windows 环境（该任务仅在 Linux 服务器访问 yihangyi 目录）");
|
||||
return;
|
||||
}
|
||||
Path scanDir = Paths.get(yihangyiTxtScanDir);
|
||||
@@ -190,6 +191,7 @@ public class AudioStatisticsScheduler {
|
||||
return;
|
||||
}
|
||||
if (System.getProperty("os.name", "").toLowerCase(Locale.ROOT).contains("win")) {
|
||||
log.info("uploadLocalSegmentFilesToMinio 跳过：Windows 环境（该任务仅在 Linux 服务器访问本地分段路径）");
|
||||
return;
|
||||
}
|
||||
try {
|
||||
|
||||
153
src/main/java/com/rj/scheduler/YihangyiVllmAsrScheduler.java
Normal file
153
src/main/java/com/rj/scheduler/YihangyiVllmAsrScheduler.java
Normal file
@@ -0,0 +1,153 @@
|
||||
package com.rj.scheduler;
|
||||
|
||||
import com.rj.config.AppConfig;
|
||||
import com.rj.config.YihangyiVllmAsrExecutorMode;
|
||||
import com.rj.config.YihangyiVllmAsrProperties;
|
||||
import com.rj.service.YihangyiVllmAsrService;
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import jakarta.annotation.PreDestroy;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* yihangyi 音频 ASR 轮询。<strong>不使用 Spring {@code @Scheduled}</strong>，避免与全局定时线程池争抢；
|
||||
* 在独立单线程上按固定间隔执行（与 Python 常驻循环等价），批内仍可选 POOL 并行转写。
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
@ConditionalOnProperty(name = "app.audio.upload.yihangyi.asr.enabled", havingValue = "true")
|
||||
public class YihangyiVllmAsrScheduler {
|
||||
|
||||
private final YihangyiVllmAsrProperties properties;
|
||||
private final YihangyiVllmAsrService yihangyiVllmAsrService;
|
||||
private final AppConfig appConfig;
|
||||
|
||||
private ExecutorService workerPool;
|
||||
private ScheduledExecutorService pollExecutor;
|
||||
|
||||
@PostConstruct
|
||||
void start() {
|
||||
if (properties.getExecutorMode() == YihangyiVllmAsrExecutorMode.POOL) {
|
||||
int n = Math.max(1, properties.getPoolSize());
|
||||
workerPool = Executors.newFixedThreadPool(n, r -> {
|
||||
Thread t = new Thread(r, "yihangyi-asr-worker");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
});
|
||||
log.info("yihangyi ASR 批内使用线程池，线程数={}", n);
|
||||
} else {
|
||||
log.info("yihangyi ASR 批内单线程顺序处理");
|
||||
}
|
||||
long intervalMs = Math.max(1000L *60*3, properties.getPollIntervalMs());
|
||||
pollExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
|
||||
Thread t = new Thread(r, "yihangyi-asr-poll");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
});
|
||||
pollExecutor.scheduleWithFixedDelay(this::pollSafe, 0, intervalMs, TimeUnit.MILLISECONDS);
|
||||
log.info("yihangyi ASR 独立轮询线程已启动，fixedDelay={}ms（不占 Spring @Scheduled 线程池）", intervalMs);
|
||||
}
|
||||
|
||||
@PreDestroy
|
||||
void shutdown() {
|
||||
if (pollExecutor != null) {
|
||||
pollExecutor.shutdown();
|
||||
try {
|
||||
if (!pollExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
|
||||
pollExecutor.shutdownNow();
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
pollExecutor.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
if (workerPool != null) {
|
||||
workerPool.shutdown();
|
||||
try {
|
||||
if (!workerPool.awaitTermination(60, TimeUnit.SECONDS)) {
|
||||
workerPool.shutdownNow();
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
workerPool.shutdownNow();
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void pollSafe() {
|
||||
try {
|
||||
log.info("yihangyi ASR 音频-----》文本， 轮询开始");
|
||||
pollAndTranscribe();
|
||||
} catch (Throwable t) {
|
||||
log.error("yihangyi ASR 轮询未捕获异常", t);
|
||||
}
|
||||
}
|
||||
|
||||
void pollAndTranscribe() {
|
||||
if (!appConfig.getScheduler().isStart()) {
|
||||
log.info("yihangyi ASR 调度跳过：app.scheduler.start=false");
|
||||
return;
|
||||
}
|
||||
if (properties.isSkipOnWindows()
|
||||
&& System.getProperty("os.name", "").toLowerCase(Locale.ROOT).contains("win")) {
|
||||
log.info("yihangyi ASR 调度跳过：当前为 Windows 且 app.audio.upload.yihangyi.asr.skip-on-windows=true");
|
||||
return;
|
||||
}
|
||||
Path watchDir = Path.of(properties.getWatchDir());
|
||||
int maxBatch = Math.max(1, properties.getMaxBatch());
|
||||
long t0 = System.nanoTime();
|
||||
log.info(
|
||||
"yihangyi ASR 调度开始：watchDir={} executorMode={} maxBatch={} pollIntervalMs={}",
|
||||
watchDir.toAbsolutePath(),
|
||||
properties.getExecutorMode(),
|
||||
maxBatch,
|
||||
properties.getPollIntervalMs());
|
||||
|
||||
if (!java.nio.file.Files.isDirectory(watchDir)) {
|
||||
log.warn("yihangyi ASR 调度结束（异常）：监视目录不存在或不是文件夹: {}", watchDir.toAbsolutePath());
|
||||
return;
|
||||
}
|
||||
int pendingTotal = yihangyiVllmAsrService.countPendingAudio(watchDir);
|
||||
log.info("yihangyi ASR 待转写音频文件总数: {}", pendingTotal);
|
||||
|
||||
List<Path> pending = yihangyiVllmAsrService.listNewestPendingAudio(watchDir, maxBatch);
|
||||
if (pending.isEmpty()) {
|
||||
log.info(
|
||||
"yihangyi ASR 调度结束：本批无待处理文件，待转写总数={}，耗时={}ms",
|
||||
pendingTotal,
|
||||
(System.nanoTime() - t0) / 1_000_000);
|
||||
return;
|
||||
}
|
||||
|
||||
log.info("yihangyi ASR 本批将处理 {} 个文件: {}", pending.size(),
|
||||
pending.stream().map(p -> p.getFileName().toString()).toList());
|
||||
|
||||
if (properties.getExecutorMode() == YihangyiVllmAsrExecutorMode.POOL && workerPool != null) {
|
||||
List<CompletableFuture<Void>> futures = pending.stream()
|
||||
.map(p -> CompletableFuture.runAsync(() -> yihangyiVllmAsrService.processOneFile(p), workerPool))
|
||||
.toList();
|
||||
CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])).join();
|
||||
} else {
|
||||
for (Path p : pending) {
|
||||
yihangyiVllmAsrService.processOneFile(p);
|
||||
}
|
||||
}
|
||||
log.info(
|
||||
"yihangyi ASR 调度结束：本批已处理 {} 个文件，待转写总数（处理前统计）={}，耗时={}ms",
|
||||
pending.size(),
|
||||
pendingTotal,
|
||||
(System.nanoTime() - t0) / 1_000_000);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user