VideoUtil.java

package com.viontech.utils;

import com.viontech.ftp.FTPClientHelper;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * .
 *
 * @author 谢明辉
 * @date 2020/9/25
 */

public class VideoUtil {
    private final static ConcurrentHashMap<String, byte[]> VIDEO_MAP = new ConcurrentHashMap<>();
    private final static ConcurrentHashMap<String, Long> VIDEO_EXPIRE_MAP = new ConcurrentHashMap<>();
    private final static ThreadPoolExecutor POOL = new ThreadPoolExecutor(20, 30, 1, TimeUnit.HOURS, new LinkedBlockingQueue<>(), new ThreadPoolExecutor.DiscardPolicy());

    public synchronized static void addVideo(String refId, byte[] fileBytes) {
        VIDEO_MAP.put(refId, fileBytes);
        VIDEO_EXPIRE_MAP.put(refId, System.currentTimeMillis());
    }

    public synchronized static byte[] getVideo(String refId) {
        return VIDEO_MAP.remove(refId);
    }

    public synchronized static void clean() {
        for (Map.Entry<String, Long> entry : VIDEO_EXPIRE_MAP.entrySet()) {
            if ((System.currentTimeMillis() - entry.getValue()) > TimeUnit.HOURS.toMillis(3)) {
                getVideo(entry.getKey());
            }
        }
    }

    public static void submitTask(String refId, String filePath, FTPClientHelper ftpClientHelper) {
        POOL.submit(new VideoPushRunnable(refId, filePath, ftpClientHelper));
    }

    @Slf4j
    private static class VideoPushRunnable implements Runnable {
        private final String refId;
        private final String filePath;
        private final FTPClientHelper ftpClientHelper;

        public VideoPushRunnable(String refId, String filePath, FTPClientHelper ftpClientHelper) {
            this.refId = refId;
            this.filePath = filePath;
            this.ftpClientHelper = ftpClientHelper;
        }

        @Override
        public void run() {
            try {
                long begin = System.currentTimeMillis();
                while (System.currentTimeMillis() - begin < TimeUnit.MINUTES.toMillis(5)) {
                    byte[] video = VideoUtil.getVideo(refId);
                    if (video != null) {
                    ftpClientHelper.storeFile(filePath, video);
                        log.info("视频推送成功,refId:[{}],path:[{}]", refId, filePath);
                        return;
                    }
                    Long aLong = VideoUtil.VIDEO_EXPIRE_MAP.get(refId);
                    if (aLong != null) {
                        log.info("视频已发送,refId:[{}],path:[{}]", refId, filePath);
                        return;
                    }
                    TimeUnit.SECONDS.sleep(10);
                    log.info("视频未找到:[{}]", refId);
                }
            } catch (Exception e) {
                log.error("推送视频出错,refId:[" + refId + "],path:[" + filePath + "]", e);
            }
        }
    }
}