redis

pom.xml

@@ -145,6 +145,11 @@
             <artifactId>spring-boot-starter-aop</artifactId>
         </dependency>
 
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-redis</artifactId>
+			<version>2.1.7.RELEASE</version>
+		</dependency>
 
 		<dependency>
 			<groupId>fadada</groupId>

src/main/java/com/yunzhi/marketing/common/VideoThreadDownLoad.java

@@ -0,0 +1,229 @@
+package com.yunzhi.marketing.common;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.RandomAccessFile;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+public class VideoThreadDownLoad {
+    public static final Logger LOG = LoggerFactory.getLogger(VideoThreadDownLoad.class);
+    /**
+     * 线程下载成功标志
+     */
+    private static int flag = 0;
+
+    /**
+     * 服务器请求路径
+     */
+    private String serverPath;
+    /**
+     * 本地路径
+     */
+    private String localPath;
+
+    /**
+     * 线程计数同步辅助
+     */
+    private CountDownLatch latch;
+
+    private final int THREAD_POOL_CORE_SIZE= 10;
+    private final int THREAD_POOL_MAX_CORE_SIZE=30;
+    private final int THREAD_POOL_KEEP_ALIVE_TIME=60;
+
+    // 定长线程池
+    private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(THREAD_POOL_CORE_SIZE,THREAD_POOL_MAX_CORE_SIZE,THREAD_POOL_KEEP_ALIVE_TIME, TimeUnit.SECONDS,new LinkedBlockingDeque<>());
+
+    public VideoThreadDownLoad(String serverPath, String localPath) {
+        this.serverPath = serverPath;
+        this.localPath = localPath;
+    }
+
+    public synchronized static void downLoad(String serverPath, String localPath) {
+        LOG.info("{}文件开始下载，下载位置为{}",serverPath,localPath);
+        ReentrantLock lock = new ReentrantLock();
+        lock.lock();
+        VideoThreadDownLoad videoThreadDownLoad = new VideoThreadDownLoad(serverPath, localPath);
+        long startTime = System.currentTimeMillis();
+        boolean flag = false;
+        try {
+            flag = videoThreadDownLoad.executeDownLoad();
+            long endTime = System.currentTimeMillis();
+            if (flag) {
+                LOG.info(serverPath + "文件下载成功,共耗时" + (endTime - startTime) / 1000 + "s");
+            }
+        } catch (Exception ex) {
+            LOG.error(ex.getMessage());
+        } finally {
+            VideoThreadDownLoad.flag = 0; // 重置 下载状态
+            if (!flag) {
+                LOG.info("{}文件下载失败,本地文件已经删除。", serverPath);
+                File file = new File(localPath);
+                file.delete();
+            }
+            lock.unlock();
+        }
+    }
+
+    /**
+     * 下载操作
+     * @return
+     */
+    public boolean executeDownLoad() {
+        try {
+            URL url = new URL(serverPath);
+            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
+            conn.setConnectTimeout(5000);//设置超时时间
+            conn.setRequestMethod("GET");//设置请求方式
+            conn.setRequestProperty("Connection", "Keep-Alive");
+            int code = conn.getResponseCode();
+            if (code != 200 && code != 206) {
+                LOG.info("无效网络地址：{}", serverPath);
+                return false;
+            }
+            //服务器返回的数据的长度，实际上就是文件的长度,单位是字节
+            long length = getRemoteFileSize(serverPath);
+
+            LOG.info("文件总长度:" + length + "字节(B)");
+            RandomAccessFile raf = new RandomAccessFile(localPath, "rwd");
+            //指定创建的文件的长度
+            raf.setLength(length);
+            raf.close();
+            //分割文件
+            int partCount = getPartCount(length);
+            int partSize = (int)(length / partCount);
+            latch = new CountDownLatch(partCount);
+            for (int threadId = 1; threadId <= partCount; threadId++) {
+                // 每一个线程下载的开始位置
+                long startIndex = (threadId - 1) * partSize;
+                // 每一个线程下载的开始位置
+                long endIndex = startIndex + partSize - 1;
+                if (threadId == partCount) {
+                    //最后一个线程下载的长度稍微长一点
+                    endIndex = length;
+                }
+//                LOG.info("线程" + threadId + "下载:" + startIndex + "字节~" + endIndex + "字节");
+                threadPool.execute(new DownLoadThread(threadId, startIndex, endIndex, latch));
+            }
+            latch.await();
+            if(flag == 0){
+                return true;
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            LOG.error("文件下载失败，文件地址：{},失败原因：{}", serverPath, e.getMessage());
+        }
+        return false;
+    }
+
+    /**
+     * 计算分割多少文件比较合适
+     * @param length
+     * @return
+     */
+    private int getPartCount(long length) {
+        int size = (int) (length / 10485760);
+        return size + 1;
+    }
+
+
+    /**
+     * 内部方法，获取远程文件大小
+     * @param remoteFileUrl
+     * @return
+     * @throws IOException
+     */
+    private long getRemoteFileSize(String remoteFileUrl) throws IOException {
+        long fileSize = 0;
+        HttpURLConnection httpConnection = (HttpURLConnection) new URL(remoteFileUrl).openConnection();
+        httpConnection.setRequestMethod("HEAD");
+        int responseCode = 0;
+        try {
+            responseCode = httpConnection.getResponseCode();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        if (responseCode >= 400) {
+            LOG.error("Web服务器响应错误!");
+            return 0;
+        }
+        String sHeader;
+        for (int i = 1;; i++) {
+            sHeader = httpConnection.getHeaderFieldKey(i);
+            if (sHeader != null && sHeader.equals("Content-Length")) {
+                fileSize = Long.parseLong(httpConnection.getHeaderField(sHeader));
+                break;
+            }
+        }
+        return fileSize;
+    }
+
+    /**
+     * 内部类用于实现下载
+     */
+    public class DownLoadThread implements Runnable {
+
+        /**
+         * 线程ID
+         */
+        private int threadId;
+        /**
+         * 下载起始位置
+         */
+        private long startIndex;
+        /**
+         * 下载结束位置
+         */
+        private long endIndex;
+
+        private CountDownLatch latch;
+
+        public DownLoadThread(int threadId, long startIndex, long endIndex, CountDownLatch latch) {
+            this.threadId = threadId;
+            this.startIndex = startIndex;
+            this.endIndex = endIndex;
+            this.latch = latch;
+        }
+
+        @Override
+        public void run() {
+            try {
+                URL url = new URL(serverPath);
+                HttpURLConnection conn = (HttpURLConnection) url.openConnection();
+                conn.setRequestProperty("Connection", "Keep-Alive");
+                conn.setRequestMethod("GET");
+                //请求服务器下载部分的文件的指定位置
+                conn.setRequestProperty("Range", "bytes=" + startIndex + "-" + endIndex);
+                conn.setConnectTimeout(5000);
+                int code = conn.getResponseCode();
+                InputStream is = conn.getInputStream();//返回资源
+                RandomAccessFile raf = new RandomAccessFile(localPath, "rwd");
+                raf.seek(startIndex);//定位文件
+                int len = 0;
+                byte[] buffer = new byte[1024];
+                while ((len = is.read(buffer)) != -1) {
+                    raf.write(buffer, 0, len);
+                }
+                is.close();
+                raf.close();
+            } catch (Exception e) {
+                //线程下载出错
+                VideoThreadDownLoad.flag = 1;
+                LOG.error(e.getMessage());
+            } finally {
+                //计数值减一
+                latch.countDown();
+            }
+
+        }
+    }
+}

src/main/java/com/yunzhi/marketing/config/RedisConfig.java

@@ -0,0 +1,37 @@
+package com.yunzhi.marketing.config;
+
+import com.yunzhi.marketing.delay.service.RedisDelayQueue;
+import com.yunzhi.marketing.listener.VideoRedisListener;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import springfox.documentation.swagger2.mappers.ModelMapper;
+
+import java.io.IOException;
+
+
+/**
+ * @author zhaodong
+ * @date 2019/8/14
+ */
+@Configuration
+public class RedisConfig {
+
+
+    @Bean
+    public RedisDelayQueue orderDelayQueue(RedisTemplate redisTemplate) throws IOException {
+        return new RedisDelayQueue(redisTemplate);
+    }
+
+    @Bean
+    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
+        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(connectionFactory);
+        container.addMessageListener(new VideoRedisListener(""), new ChannelTopic(""));
+        return container;
+    }
+}

src/main/java/com/yunzhi/marketing/delay/DelayMessage.java

@@ -0,0 +1,28 @@
+package com.yunzhi.marketing.delay;
+
+public class DelayMessage<T> {
+    private long score;
+    private  T object;
+
+
+    public DelayMessage(long score, T object) {
+        this.score = score;
+        this.object = object;
+    }
+
+    public long getScore() {
+        return score;
+    }
+
+    public void setScore(long score) {
+        this.score = score;
+    }
+
+    public T getObject() {
+        return object;
+    }
+
+    public void setObject(T object) {
+        this.object = object;
+    }
+}

src/main/java/com/yunzhi/marketing/delay/service/RedisDelayQueue.java

@@ -0,0 +1,43 @@
+package com.yunzhi.marketing.delay.service;
+
+import com.tdh.brokenmeeting.delay.DelayMessage;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.script.RedisScript;
+
+import java.io.IOException;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Date;
+import java.util.List;
+import java.util.Set;
+
+public class RedisDelayQueue {
+    private final DateTimeFormatter dtf=DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+    private RedisScript<List> delayLuaScript;
+    private RedisTemplate<String,Object> redisTemplate;
+
+
+    public RedisDelayQueue(RedisTemplate redisTemplate) throws IOException {
+        this.redisTemplate = redisTemplate;
+//        String delayLuaScriptStr= IOUtils.toString(getClass().getClassLoader().getResourceAsStream("lua/delayLua.lua"), "utf-8");
+//        delayLuaScript=new DefaultRedisScript<>(delayLuaScriptStr,List.class);
+    }
+
+    public <T> void add(DelayMessage<T> delayMessage, String redisDelayQueue){
+        System.out.println("发送时间:"+ LocalDateTime.now().format(dtf)+",消息:" +delayMessage.getObject().toString());
+        redisTemplate.opsForZSet().add(redisDelayQueue,delayMessage.getObject(),delayMessage.getScore());
+    }
+
+    public Set get(String redisDelayQueue){
+        Set set = redisTemplate.opsForZSet().rangeByScore(redisDelayQueue, 0,new Double(new Date().getTime()));
+        if(!set.isEmpty()){
+            redisTemplate.opsForZSet().remove(redisDelayQueue,set.toArray());
+        }
+        return set;
+    }
+
+    public void remove(String key,String redisDelayQueue){
+        redisTemplate.opsForZSet().remove(key,redisDelayQueue);
+    }
+
+}

src/main/java/com/yunzhi/marketing/listener/VideoRedisListener.java

@@ -0,0 +1,50 @@
+package com.yunzhi.marketing.listener;
+
+import com.alibaba.fastjson.JSONObject;
+import com.yunzhi.marketing.common.VideoThreadDownLoad;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+
+import java.io.File;
+
+public class VideoRedisListener implements MessageListener {
+    public static final Logger LOG = LoggerFactory.getLogger(VideoRedisListener.class);
+
+    private String videoLocalPath;
+
+    public VideoRedisListener(String localPath) {
+        videoLocalPath = localPath;
+    }
+
+    @Override
+    public void onMessage(Message message, byte[] pattern) {
+        LOG.info("订阅方收到录制文件信息:{}",message.toString());
+        JSONObject jsonObject = JSONObject.parseObject(message.toString());
+        String serverPath = jsonObject.getString("video_url");
+        //构造文件下载到本地的路径
+        String localPath = getLocalPath(jsonObject);
+        //多线程下载文件
+        VideoThreadDownLoad.downLoad(serverPath,localPath);
+    }
+
+    /**
+     * 构建本地存储路径
+     * @param jsonObject
+     * @return
+     */
+    private String getLocalPath(JSONObject jsonObject) {
+        String fileName = jsonObject.getString("stream_id") ;
+        String dirName = fileName.split("_")[0];
+        String localDir = videoLocalPath + File.separator +  dirName;
+        File dir = new File(localDir);
+        if (!dir.isDirectory()) {
+            dir.mkdir();
+        }
+        String videoUrl = jsonObject.getString("video_url");
+        String suffix = videoUrl.substring(videoUrl.lastIndexOf("."));
+        String localPath = localDir + File.separator + fileName + "_" + jsonObject.getInteger("start_time") + "_" + jsonObject.getInteger("end_time") + suffix;
+        return localPath;
+    }
+}