fuxingfan před 4 roky
rodič
revize
e159ce973d

+ 0
- 6
pom.xml Zobrazit soubor

@@ -145,12 +145,6 @@
145 145
             <artifactId>spring-boot-starter-aop</artifactId>
146 146
         </dependency>
147 147
 
148
-		<dependency>
149
-			<groupId>org.springframework.boot</groupId>
150
-			<artifactId>spring-boot-starter-data-redis</artifactId>
151
-			<version>2.1.7.RELEASE</version>
152
-		</dependency>
153
-
154 148
 		<dependency>
155 149
 			<groupId>fadada</groupId>
156 150
 			<artifactId>fadada_api_sdk</artifactId>

+ 0
- 229
src/main/java/com/yunzhi/marketing/common/VideoThreadDownLoad.java Zobrazit soubor

@@ -1,229 +0,0 @@
1
-package com.yunzhi.marketing.common;
2
-
3
-import org.slf4j.Logger;
4
-import org.slf4j.LoggerFactory;
5
-
6
-import java.io.File;
7
-import java.io.IOException;
8
-import java.io.InputStream;
9
-import java.io.RandomAccessFile;
10
-import java.net.HttpURLConnection;
11
-import java.net.URL;
12
-import java.util.concurrent.CountDownLatch;
13
-import java.util.concurrent.LinkedBlockingDeque;
14
-import java.util.concurrent.ThreadPoolExecutor;
15
-import java.util.concurrent.TimeUnit;
16
-import java.util.concurrent.locks.ReentrantLock;
17
-
18
-public class VideoThreadDownLoad {
19
-    public static final Logger LOG = LoggerFactory.getLogger(VideoThreadDownLoad.class);
20
-    /**
21
-     * 线程下载成功标志
22
-     */
23
-    private static int flag = 0;
24
-
25
-    /**
26
-     * 服务器请求路径
27
-     */
28
-    private String serverPath;
29
-    /**
30
-     * 本地路径
31
-     */
32
-    private String localPath;
33
-
34
-    /**
35
-     * 线程计数同步辅助
36
-     */
37
-    private CountDownLatch latch;
38
-
39
-    private final int THREAD_POOL_CORE_SIZE= 10;
40
-    private final int THREAD_POOL_MAX_CORE_SIZE=30;
41
-    private final int THREAD_POOL_KEEP_ALIVE_TIME=60;
42
-
43
-    // 定长线程池
44
-    private ThreadPoolExecutor threadPool = new ThreadPoolExecutor(THREAD_POOL_CORE_SIZE,THREAD_POOL_MAX_CORE_SIZE,THREAD_POOL_KEEP_ALIVE_TIME, TimeUnit.SECONDS,new LinkedBlockingDeque<>());
45
-
46
-    public VideoThreadDownLoad(String serverPath, String localPath) {
47
-        this.serverPath = serverPath;
48
-        this.localPath = localPath;
49
-    }
50
-
51
-    public synchronized static void downLoad(String serverPath, String localPath) {
52
-        LOG.info("{}文件开始下载,下载位置为{}",serverPath,localPath);
53
-        ReentrantLock lock = new ReentrantLock();
54
-        lock.lock();
55
-        VideoThreadDownLoad videoThreadDownLoad = new VideoThreadDownLoad(serverPath, localPath);
56
-        long startTime = System.currentTimeMillis();
57
-        boolean flag = false;
58
-        try {
59
-            flag = videoThreadDownLoad.executeDownLoad();
60
-            long endTime = System.currentTimeMillis();
61
-            if (flag) {
62
-                LOG.info(serverPath + "文件下载成功,共耗时" + (endTime - startTime) / 1000 + "s");
63
-            }
64
-        } catch (Exception ex) {
65
-            LOG.error(ex.getMessage());
66
-        } finally {
67
-            VideoThreadDownLoad.flag = 0; // 重置 下载状态
68
-            if (!flag) {
69
-                LOG.info("{}文件下载失败,本地文件已经删除。", serverPath);
70
-                File file = new File(localPath);
71
-                file.delete();
72
-            }
73
-            lock.unlock();
74
-        }
75
-    }
76
-
77
-    /**
78
-     * 下载操作
79
-     * @return
80
-     */
81
-    public boolean executeDownLoad() {
82
-        try {
83
-            URL url = new URL(serverPath);
84
-            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
85
-            conn.setConnectTimeout(5000);//设置超时时间
86
-            conn.setRequestMethod("GET");//设置请求方式
87
-            conn.setRequestProperty("Connection", "Keep-Alive");
88
-            int code = conn.getResponseCode();
89
-            if (code != 200 && code != 206) {
90
-                LOG.info("无效网络地址:{}", serverPath);
91
-                return false;
92
-            }
93
-            //服务器返回的数据的长度,实际上就是文件的长度,单位是字节
94
-            long length = getRemoteFileSize(serverPath);
95
-
96
-            LOG.info("文件总长度:" + length + "字节(B)");
97
-            RandomAccessFile raf = new RandomAccessFile(localPath, "rwd");
98
-            //指定创建的文件的长度
99
-            raf.setLength(length);
100
-            raf.close();
101
-            //分割文件
102
-            int partCount = getPartCount(length);
103
-            int partSize = (int)(length / partCount);
104
-            latch = new CountDownLatch(partCount);
105
-            for (int threadId = 1; threadId <= partCount; threadId++) {
106
-                // 每一个线程下载的开始位置
107
-                long startIndex = (threadId - 1) * partSize;
108
-                // 每一个线程下载的开始位置
109
-                long endIndex = startIndex + partSize - 1;
110
-                if (threadId == partCount) {
111
-                    //最后一个线程下载的长度稍微长一点
112
-                    endIndex = length;
113
-                }
114
-//                LOG.info("线程" + threadId + "下载:" + startIndex + "字节~" + endIndex + "字节");
115
-                threadPool.execute(new DownLoadThread(threadId, startIndex, endIndex, latch));
116
-            }
117
-            latch.await();
118
-            if(flag == 0){
119
-                return true;
120
-            }
121
-        } catch (Exception e) {
122
-            e.printStackTrace();
123
-            LOG.error("文件下载失败,文件地址:{},失败原因:{}", serverPath, e.getMessage());
124
-        }
125
-        return false;
126
-    }
127
-
128
-    /**
129
-     * 计算分割多少文件比较合适
130
-     * @param length
131
-     * @return
132
-     */
133
-    private int getPartCount(long length) {
134
-        int size = (int) (length / 10485760);
135
-        return size + 1;
136
-    }
137
-
138
-
139
-    /**
140
-     * 内部方法,获取远程文件大小
141
-     * @param remoteFileUrl
142
-     * @return
143
-     * @throws IOException
144
-     */
145
-    private long getRemoteFileSize(String remoteFileUrl) throws IOException {
146
-        long fileSize = 0;
147
-        HttpURLConnection httpConnection = (HttpURLConnection) new URL(remoteFileUrl).openConnection();
148
-        httpConnection.setRequestMethod("HEAD");
149
-        int responseCode = 0;
150
-        try {
151
-            responseCode = httpConnection.getResponseCode();
152
-        } catch (IOException e) {
153
-            e.printStackTrace();
154
-        }
155
-        if (responseCode >= 400) {
156
-            LOG.error("Web服务器响应错误!");
157
-            return 0;
158
-        }
159
-        String sHeader;
160
-        for (int i = 1;; i++) {
161
-            sHeader = httpConnection.getHeaderFieldKey(i);
162
-            if (sHeader != null && sHeader.equals("Content-Length")) {
163
-                fileSize = Long.parseLong(httpConnection.getHeaderField(sHeader));
164
-                break;
165
-            }
166
-        }
167
-        return fileSize;
168
-    }
169
-
170
-    /**
171
-     * 内部类用于实现下载
172
-     */
173
-    public class DownLoadThread implements Runnable {
174
-
175
-        /**
176
-         * 线程ID
177
-         */
178
-        private int threadId;
179
-        /**
180
-         * 下载起始位置
181
-         */
182
-        private long startIndex;
183
-        /**
184
-         * 下载结束位置
185
-         */
186
-        private long endIndex;
187
-
188
-        private CountDownLatch latch;
189
-
190
-        public DownLoadThread(int threadId, long startIndex, long endIndex, CountDownLatch latch) {
191
-            this.threadId = threadId;
192
-            this.startIndex = startIndex;
193
-            this.endIndex = endIndex;
194
-            this.latch = latch;
195
-        }
196
-
197
-        @Override
198
-        public void run() {
199
-            try {
200
-                URL url = new URL(serverPath);
201
-                HttpURLConnection conn = (HttpURLConnection) url.openConnection();
202
-                conn.setRequestProperty("Connection", "Keep-Alive");
203
-                conn.setRequestMethod("GET");
204
-                //请求服务器下载部分的文件的指定位置
205
-                conn.setRequestProperty("Range", "bytes=" + startIndex + "-" + endIndex);
206
-                conn.setConnectTimeout(5000);
207
-                int code = conn.getResponseCode();
208
-                InputStream is = conn.getInputStream();//返回资源
209
-                RandomAccessFile raf = new RandomAccessFile(localPath, "rwd");
210
-                raf.seek(startIndex);//定位文件
211
-                int len = 0;
212
-                byte[] buffer = new byte[1024];
213
-                while ((len = is.read(buffer)) != -1) {
214
-                    raf.write(buffer, 0, len);
215
-                }
216
-                is.close();
217
-                raf.close();
218
-            } catch (Exception e) {
219
-                //线程下载出错
220
-                VideoThreadDownLoad.flag = 1;
221
-                LOG.error(e.getMessage());
222
-            } finally {
223
-                //计数值减一
224
-                latch.countDown();
225
-            }
226
-
227
-        }
228
-    }
229
-}

+ 0
- 37
src/main/java/com/yunzhi/marketing/config/RedisConfig.java Zobrazit soubor

@@ -1,37 +0,0 @@
1
-package com.yunzhi.marketing.config;
2
-
3
-import com.yunzhi.marketing.delay.service.RedisDelayQueue;
4
-import com.yunzhi.marketing.listener.VideoRedisListener;
5
-import org.springframework.beans.factory.annotation.Value;
6
-import org.springframework.context.annotation.Bean;
7
-import org.springframework.context.annotation.Configuration;
8
-import org.springframework.data.redis.connection.RedisConnectionFactory;
9
-import org.springframework.data.redis.core.RedisTemplate;
10
-import org.springframework.data.redis.listener.ChannelTopic;
11
-import org.springframework.data.redis.listener.RedisMessageListenerContainer;
12
-import springfox.documentation.swagger2.mappers.ModelMapper;
13
-
14
-import java.io.IOException;
15
-
16
-
17
-/**
18
- * @author zhaodong
19
- * @date 2019/8/14
20
- */
21
-@Configuration
22
-public class RedisConfig {
23
-
24
-
25
-    @Bean
26
-    public RedisDelayQueue orderDelayQueue(RedisTemplate redisTemplate) throws IOException {
27
-        return new RedisDelayQueue(redisTemplate);
28
-    }
29
-
30
-    @Bean
31
-    public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
32
-        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
33
-        container.setConnectionFactory(connectionFactory);
34
-        container.addMessageListener(new VideoRedisListener(""), new ChannelTopic(""));
35
-        return container;
36
-    }
37
-}

+ 0
- 28
src/main/java/com/yunzhi/marketing/delay/DelayMessage.java Zobrazit soubor

@@ -1,28 +0,0 @@
1
-package com.yunzhi.marketing.delay;
2
-
3
-public class DelayMessage<T> {
4
-    private long score;
5
-    private  T object;
6
-
7
-
8
-    public DelayMessage(long score, T object) {
9
-        this.score = score;
10
-        this.object = object;
11
-    }
12
-
13
-    public long getScore() {
14
-        return score;
15
-    }
16
-
17
-    public void setScore(long score) {
18
-        this.score = score;
19
-    }
20
-
21
-    public T getObject() {
22
-        return object;
23
-    }
24
-
25
-    public void setObject(T object) {
26
-        this.object = object;
27
-    }
28
-}

+ 0
- 43
src/main/java/com/yunzhi/marketing/delay/service/RedisDelayQueue.java Zobrazit soubor

@@ -1,43 +0,0 @@
1
-package com.yunzhi.marketing.delay.service;
2
-
3
-import com.tdh.brokenmeeting.delay.DelayMessage;
4
-import org.springframework.data.redis.core.RedisTemplate;
5
-import org.springframework.data.redis.core.script.RedisScript;
6
-
7
-import java.io.IOException;
8
-import java.time.LocalDateTime;
9
-import java.time.format.DateTimeFormatter;
10
-import java.util.Date;
11
-import java.util.List;
12
-import java.util.Set;
13
-
14
-public class RedisDelayQueue {
15
-    private final DateTimeFormatter dtf=DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
16
-    private RedisScript<List> delayLuaScript;
17
-    private RedisTemplate<String,Object> redisTemplate;
18
-
19
-
20
-    public RedisDelayQueue(RedisTemplate redisTemplate) throws IOException {
21
-        this.redisTemplate = redisTemplate;
22
-//        String delayLuaScriptStr= IOUtils.toString(getClass().getClassLoader().getResourceAsStream("lua/delayLua.lua"), "utf-8");
23
-//        delayLuaScript=new DefaultRedisScript<>(delayLuaScriptStr,List.class);
24
-    }
25
-
26
-    public <T> void add(DelayMessage<T> delayMessage, String redisDelayQueue){
27
-        System.out.println("发送时间:"+ LocalDateTime.now().format(dtf)+",消息:" +delayMessage.getObject().toString());
28
-        redisTemplate.opsForZSet().add(redisDelayQueue,delayMessage.getObject(),delayMessage.getScore());
29
-    }
30
-
31
-    public Set get(String redisDelayQueue){
32
-        Set set = redisTemplate.opsForZSet().rangeByScore(redisDelayQueue, 0,new Double(new Date().getTime()));
33
-        if(!set.isEmpty()){
34
-            redisTemplate.opsForZSet().remove(redisDelayQueue,set.toArray());
35
-        }
36
-        return set;
37
-    }
38
-
39
-    public void remove(String key,String redisDelayQueue){
40
-        redisTemplate.opsForZSet().remove(key,redisDelayQueue);
41
-    }
42
-
43
-}

+ 0
- 50
src/main/java/com/yunzhi/marketing/listener/VideoRedisListener.java Zobrazit soubor

@@ -1,50 +0,0 @@
1
-package com.yunzhi.marketing.listener;
2
-
3
-import com.alibaba.fastjson.JSONObject;
4
-import com.yunzhi.marketing.common.VideoThreadDownLoad;
5
-import org.slf4j.Logger;
6
-import org.slf4j.LoggerFactory;
7
-import org.springframework.data.redis.connection.Message;
8
-import org.springframework.data.redis.connection.MessageListener;
9
-
10
-import java.io.File;
11
-
12
-public class VideoRedisListener implements MessageListener {
13
-    public static final Logger LOG = LoggerFactory.getLogger(VideoRedisListener.class);
14
-
15
-    private String videoLocalPath;
16
-
17
-    public VideoRedisListener(String localPath) {
18
-        videoLocalPath = localPath;
19
-    }
20
-
21
-    @Override
22
-    public void onMessage(Message message, byte[] pattern) {
23
-        LOG.info("订阅方收到录制文件信息:{}",message.toString());
24
-        JSONObject jsonObject = JSONObject.parseObject(message.toString());
25
-        String serverPath = jsonObject.getString("video_url");
26
-        //构造文件下载到本地的路径
27
-        String localPath = getLocalPath(jsonObject);
28
-        //多线程下载文件
29
-        VideoThreadDownLoad.downLoad(serverPath,localPath);
30
-    }
31
-
32
-    /**
33
-     * 构建本地存储路径
34
-     * @param jsonObject
35
-     * @return
36
-     */
37
-    private String getLocalPath(JSONObject jsonObject) {
38
-        String fileName = jsonObject.getString("stream_id") ;
39
-        String dirName = fileName.split("_")[0];
40
-        String localDir = videoLocalPath + File.separator +  dirName;
41
-        File dir = new File(localDir);
42
-        if (!dir.isDirectory()) {
43
-            dir.mkdir();
44
-        }
45
-        String videoUrl = jsonObject.getString("video_url");
46
-        String suffix = videoUrl.substring(videoUrl.lastIndexOf("."));
47
-        String localPath = localDir + File.separator + fileName + "_" + jsonObject.getInteger("start_time") + "_" + jsonObject.getInteger("end_time") + suffix;
48
-        return localPath;
49
-    }
50
-}