package com.ard.work.service.task; import com.ard.common.utils.StringUtils; import com.ard.common.utils.file.MimeTypeUtils; import com.ard.oss.service.impl.OssServiceContext; import com.ard.work.domain.entity.ArdMediaPic; import com.ard.work.domain.entity.ArdMediaVideo; import com.ard.work.mapper.ArdMediaPicMapper; import com.ard.work.mapper.ArdMediaVideoMapper; import com.ard.work.util.osSelect; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Component; import javax.annotation.Resource; import javax.net.ssl.HttpsURLConnection; import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManager; import javax.net.ssl.X509TrustManager; import java.io.*; import java.net.HttpURLConnection; import java.net.URL; import java.net.URLConnection; import java.nio.file.Files; import java.nio.file.Path; import java.security.cert.X509Certificate; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; /** * 定时同步媒体任务 * * @author ard */ @Component @Slf4j public class MediaTask { @Resource private ArdMediaPicMapper ardMediaPicMapper; @Resource private ArdMediaVideoMapper ardMediaVideoMapper; @Resource private OssServiceContext ossServiceContext; @Resource(name = "mediaVideoExecutor") private ThreadPoolTaskExecutor mediaVideoExecutor; @Resource(name = "mediaPicExecutor") private ThreadPoolTaskExecutor mediaPicExecutor; private final AtomicBoolean runningVideo = new AtomicBoolean(false); private final AtomicBoolean runningPic = new AtomicBoolean(false); @Scheduled(cron = "0/10 * * * * ?") public void syncVideoMediaUrl() { log.debug("开始同步视频媒体"); if (!runningVideo.compareAndSet(false, true)) { log.debug("上一次同步还没完成,跳过本次调度"); return; } try { while (true) { List videos = ardMediaVideoMapper.selectPendingVideos(10); if (videos.isEmpty()) { break; // 没有待处理视频,退出循环 } for (ArdMediaVideo video : videos) { int updated = ardMediaVideoMapper.markProcessing(video.getId()); if (updated != 1) continue; mediaVideoExecutor.execute(() -> processOneVideo(video)); } Thread.sleep(200); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { runningVideo.set(false); } } @Scheduled(cron = "0/10 * * * * ?") public void syncPicMediaUrl() { log.warn("开始同步图片媒体"); if (!runningPic.compareAndSet(false, true)) { log.warn("上一次同步还没完成,跳过本次调度"); return; } try { while (true) { List pics = ardMediaPicMapper.selectPendingPics(10); // 查询待处理的图片 if (pics.isEmpty()) { break; // 没有待处理图片,退出循环 } for (ArdMediaPic pic : pics) { // 🔒 原子抢占:0 → 1 int updated = ardMediaPicMapper.markProcessing(pic.getId()); if (updated != 1) { continue; // 已被别的任务抢占 } mediaPicExecutor.execute(() -> processOnePic(pic)); } // 短暂 sleep,避免无限循环CPU空转 Thread.sleep(200); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { runningPic.set(false); } } private void processOneVideo(ArdMediaVideo video) { int maxRetries = 3; // 最大重试次数 int retryDelay = 3000; // 每次重试等待 3 秒 int attempt = 0; while (attempt < maxRetries) { attempt++; try (InputStream is = openUrlInputStream(video.getSourceUrl())) { if (is != null) { String mediaUrl = ossServiceContext.upload( "record", video.getTaskId() + "/" + video.getId() + ".mp4", is, MimeTypeUtils.VIDEO_MP4 ); video.setMediaUrl(mediaUrl); // 生成视频缩略图 processVideoThumbnail(video); video.setSyncStatus(2); // 成功 break; // 上传成功,退出循环 } else { log.warn("视频尚未生成,尝试重试 {}/{}", attempt, maxRetries); } } catch (Exception e) { log.error("视频同步异常 videoId={}, 尝试 {}/{}", video.getId(), attempt, maxRetries, e); } if (attempt < maxRetries) { try { Thread.sleep(retryDelay); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } else { video.setSyncStatus(3); // 达到最大重试仍失败 } } video.setUpdateTime(new Date()); try { ardMediaVideoMapper.updateArdMediaVideo(video); } catch (Exception e) { log.error("更新视频状态失败 videoId={}", video.getId(), e); } } private void processOnePic(ArdMediaPic pic) { int maxRetries = 3; // 最大重试次数 int retryDelay = 2000; // 每次重试等待 2 秒 int attempt = 0; while (attempt < maxRetries) { attempt++; try (InputStream inputStream = openUrlInputStream(pic.getSourceUrl())) { if (inputStream != null) { String url = ossServiceContext.upload( "pic", pic.getTaskId() + "/" + pic.getId() + ".jpg", inputStream, MimeTypeUtils.IMAGE_JPG ); pic.setMediaUrl(url); pic.setSyncStatus(2); // 成功 break; // 上传成功,退出循环 } else { log.warn("图片尚未生成,尝试重试 {}/{}", attempt, maxRetries); } } catch (Exception e) { log.error("图片同步异常 taskId={}, photoId={}, 尝试 {}/{}", pic.getTaskId(), pic.getId(), attempt, maxRetries, e); } if (attempt < maxRetries) { try { Thread.sleep(retryDelay); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } else { pic.setSyncStatus(3); // 达到最大重试仍失败 } } pic.setUpdateTime(new Date()); try { ardMediaPicMapper.updatePic(pic); } catch (Exception e) { log.error("更新图片状态失败 photoId={}", pic.getId(), e); } } /** * 处理视频缩略图 */ private void processVideoThumbnail(ArdMediaVideo video) { // 生成缩略图 try (ByteArrayInputStream in = generateThumbnail(video.getSourceUrl())) { String thumbUrl = ossServiceContext.upload( "record", video.getTaskId() + "/" + video.getId() + ".jpg", in, MimeTypeUtils.IMAGE_JPEG ); video.setThumbnailUrl(thumbUrl); } catch (Exception e) { log.warn("视频缩略图生成或上传失败 taskId={}, videoId={}", video.getTaskId(), video.getId(), e); } // 获取文件大小 try { long videoSize = getFileSizeFromUrl(video.getSourceUrl()); video.setFileSize(videoSize); } catch (Exception e) { log.warn("获取视频大小失败 url={}", video.getSourceUrl(), e); } } private long getFileSizeFromUrl(String fileUrl) throws IOException { URL url = new URL(fileUrl); HttpURLConnection conn = (HttpURLConnection) url.openConnection(); conn.setRequestMethod("HEAD"); conn.setConnectTimeout(5000); conn.setReadTimeout(5000); conn.connect(); long size = conn.getContentLengthLong(); // 返回字节数 conn.disconnect(); return size; } private ByteArrayInputStream generateThumbnail(String videoFile) { String ffmpeg = osSelect.isWindows() ? "ffmpeg" : "/usr/local/ffmpeg/bin/ffmpeg"; Path tempImg = null; try { tempImg = Files.createTempFile("thumb_", ".jpg"); List cmd = new ArrayList<>(); cmd.add(ffmpeg); cmd.add("-y"); cmd.add("-loglevel"); cmd.add("error"); cmd.add("-i"); cmd.add(videoFile); cmd.add("-ss"); cmd.add(String.valueOf(1)); cmd.add("-vframes"); cmd.add("1"); cmd.add("-vf"); cmd.add("scale=" + 200 + ":-1"); cmd.add(tempImg.toAbsolutePath().toString()); log.info("ffmpeg cmd: {}", String.join(" ", cmd)); Process process = new ProcessBuilder(cmd).start(); drain(process.getErrorStream()); boolean finished = process.waitFor(15, TimeUnit.SECONDS); // 超时保护 if (!finished) { process.destroyForcibly(); throw new RuntimeException("ffmpeg timeout"); } if (process.exitValue() != 0) { throw new RuntimeException("ffmpeg exit=" + process.exitValue()); } byte[] data = Files.readAllBytes(tempImg); return new ByteArrayInputStream(data); } catch (Exception e) { throw new RuntimeException("生成缩略图失败", e); } finally { if (tempImg != null) { try { Files.deleteIfExists(tempImg); } catch (IOException ignore) { } } } } private static void drain(InputStream is) { new Thread(() -> { try (BufferedReader br = new BufferedReader(new InputStreamReader(is))) { while (br.readLine() != null) { } } catch (IOException ignore) { } }, "ffmpeg-stderr").start(); } /** * 将 URL 转成 InputStream * * @param fileUrl 文件 URL * @return InputStream,如果失败返回 null */ private InputStream openUrlInputStream(String fileUrl) { try { URL url = new URL(fileUrl); // 支持 HTTPS 信任所有证书(如果需要访问自签名证书) if ("https".equalsIgnoreCase(url.getProtocol())) { TrustManager[] trustAllCerts = new TrustManager[]{ new X509TrustManager() { @Override public void checkClientTrusted(X509Certificate[] chain, String authType) { } @Override public void checkServerTrusted(X509Certificate[] chain, String authType) { } @Override public X509Certificate[] getAcceptedIssuers() { return null; } } }; SSLContext sc = SSLContext.getInstance("SSL"); sc.init(null, trustAllCerts, new java.security.SecureRandom()); HttpsURLConnection.setDefaultSSLSocketFactory(sc.getSocketFactory()); HttpsURLConnection.setDefaultHostnameVerifier((hostname, session) -> true); } URLConnection connection = url.openConnection(); connection.setConnectTimeout(5000); connection.setReadTimeout(10000); return connection.getInputStream(); } catch (Exception e) { e.printStackTrace(); return null; } } }