RedisServiceImpl.java

package com.viontech.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.viontech.service.adapter.IRedisService;
import com.viontech.util.Base64PicUtil;
import com.viontech.util.VideoUtil;
import com.viontech.vo.VideoVo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.FileSystemResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.http.*;
import org.springframework.mock.web.MockMultipartFile;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.multipart.MultipartFile;

import javax.annotation.Resource;
import java.io.File;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Service
public class RedisServiceImpl implements IRedisService {
    private static final Logger log = LoggerFactory.getLogger(RedisServiceImpl.class);

    @Resource
    RedisTemplate redisTemplate;

    @Autowired
    ThreadPoolTaskExecutor dataForwardServiceExecutor;
    @Autowired
    ThreadPoolTaskExecutor videoForwardServiceExecutor;

    @Value("${data_forward_url}")
    String data_forward_url;

    @Value("${consumer_crash}")
    Integer consumer_crash;

    @Value("${video_forward_url}")
    String video_forward_url;

    @Value("${video_download_url}")
    String video_download_url;
    @Autowired
    RestTemplate restTemplate;
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    HttpHeaders headers;
    HttpHeaders videoheaders;
    @Value("${pic.path}")
    private String picPath;
    @Value("${video.path}")
    private String videoPath;
    private CopyOnWriteArrayList<String> ssuccessCacheFiles;

    private CopyOnWriteArrayList<String> errorCacheFiles;


    public RedisServiceImpl() {
        this.headers = new HttpHeaders();
        videoheaders = new HttpHeaders();
        this.headers.setContentType(MediaType.parseMediaType("application/json;charset=UTF-8"));
        videoheaders.setContentType(MediaType.parseMediaType("multipart/form-data;charset=UTF-8"));
        this.ssuccessCacheFiles = new CopyOnWriteArrayList<>();
        this.errorCacheFiles = new CopyOnWriteArrayList<>();
    }

    public Object recvData(JSONObject data) {
        try {
            if (StringUtils.isEmpty(this.data_forward_url)) {
                return null;
            }
            List<Map> pics = (List<Map>) data.get("pics");
            if (!CollectionUtils.isEmpty(pics)) pics.stream().forEach(pic -> {
                String picpath = Base64PicUtil.save(String.valueOf(pic.get("pic_base64")), StringUtils.isEmpty(data.getJSONObject("event_data").getString("ID")) ? (System.currentTimeMillis() + "" + (Math.random() * 100.0D) + ".") : (data.getJSONObject("event_data").getString("ID") + "-" + System.currentTimeMillis() + (Math.random() * 100.0D) + "." + ((pic.get("format") == null) ? "jpg" : (String) pic.get("format"))), this.picPath);
                pic.put("src_url", picpath);
                pic.remove("pic_base64");
            });
            data.put("forward_url", this.data_forward_url);
            redisTemplate.opsForList().leftPush("data_forward", data.toJSONString());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public Object uploadVideo(VideoVo videoVo) {
        try {
            if (StringUtils.isEmpty(this.video_forward_url)) return null;
            MultipartFile videofile = videoVo.getFile();
            if (videofile == null || videofile.isEmpty()) return true;
            //String fileName = videoVo.getFile().getOriginalFilename();
            //String tempFilePath = System.getProperty("java.io.tmpdir") + "/" + videoVo.getRefid() + ".mp4";
            //System.out.println("tempFilePath=" + tempFilePath);
            //File tempFile = new File(tempFilePath);
            //videofile.transferTo(tempFile);
            //videoVo.setTempFile(tempFile);
            //boolean upload_result = upload(videoVo);
            //if (!upload_result) {
            String videopath = VideoUtil.save(videofile.getInputStream(), this.videoPath, videoVo.getRefid() + ".mp4");
            videoVo.setFile(null);
            videoVo.setVideo_path(videopath);
            if (videoVo.getTempFile() != null) videoVo.getTempFile().delete();
            videoVo.setTempFile(null);
            redisTemplate.opsForList().leftPush("video_forward", JSONObject.toJSONString(videoVo));
            //this.redisTemplate.opsForValue().set("video_forward_" + System.currentTimeMillis() + (Math.random() * 100.0D), JSONObject.toJSONString(videoVo), 3L, TimeUnit.DAYS);
            //} else {
            //    this.ssuccessCacheFiles.add(videoVo.getVideo_path());
            // }
            if (videoVo.getTempFile() != null) videoVo.getTempFile().delete();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Scheduled(fixedDelay = 2010L)
    @Async
    public void forwardUploadVideo() {
        try {
            Long size = redisTemplate.opsForList().size("video_forward");
            log.info("当前短时录像总数量为:{}", size);
            if (size == null || size <= 0) {
                if (videoForwardServiceExecutor.getActiveCount() == 0) {
                    try {
                        Runtime.getRuntime().exec("rm -rf " + this.videoPath).waitFor();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                return;
            }
            List<String> videodatas = redisTemplate.opsForList().range("video_forward", -50, -1);
            ;
            /*Set<String> keys = this.redisTemplate.keys("*video_forward_*");
            if (CollectionUtils.isEmpty(keys)) {
                this.videoSending = 0;
                try {
                    log.info("rm -rf {}", this.videoPath);
                    Runtime.getRuntime().exec("rm -rf " + this.videoPath).waitFor();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return;
            }*/
            log.debug("视频重发开始");
            MultipartFile multipartFile = null;
            for (Iterator<String> it = videodatas.iterator(); it.hasNext(); ) {
                try {
                    String data = it.next();
                    //System.out.println("videodata:"+data);
                    redisTemplate.opsForList().remove("video_forward", -1, data);
                    videoForwardServiceExecutor.execute(() -> {
                        VideoVo videoVo = JSONObject.parseObject(data, VideoVo.class);
                        MockMultipartFile mockMultipartFile;
                        FileSystemResource resource;
                        LinkedMultiValueMap linkedMultiValueMap = null;
                        HttpEntity<MultiValueMap<String, Object>> formEntity;
                        try {
                            if (!new File(videoVo.getVideo_path()).exists()) {
                                redisTemplate.opsForList().remove("video_forward", -1, data);
                                return;
                            }
                            Thread.sleep(5);
                            mockMultipartFile = new MockMultipartFile(videoVo.getRefid() + ".mp4", VideoUtil.getInputStream(videoVo.getVideo_path()));
                            videoVo.setFile(mockMultipartFile);
                            //boolean b = upload(videoVo);
                            linkedMultiValueMap = new LinkedMultiValueMap();
                            resource = new FileSystemResource(videoVo.getVideo_path());
                            linkedMultiValueMap.add("refid", videoVo.getRefid());
                            linkedMultiValueMap.add("format", videoVo.getFormat());
                            linkedMultiValueMap.add("location_id", videoVo.getLocation_id());
                            linkedMultiValueMap.add("file", resource);
                            formEntity = new HttpEntity(linkedMultiValueMap, videoheaders);
                            ResponseEntity<String> responseEntity = null;
                            try {
                                responseEntity = restTemplate.postForEntity(video_forward_url, formEntity, String.class);
                            } catch (RestClientException exception) {
                                log.error(exception.getMessage());
                            }

                            if (responseEntity != null && (200 == responseEntity.getStatusCode().value() && "200".equals(ecode(responseEntity.getBody())))) {
                                log.info("视频重发地址:{};refid:{};http:{};响应信息{}", video_forward_url, videoVo.getRefid(), responseEntity.getStatusCode().value(), responseEntity.getBody());
                                ssuccessCacheFiles.add(videoVo.getVideo_path());
                                redisTemplate.opsForList().remove("video_forward", -1, data);
                            } else {
                                videoVo.setFile(null);
                                if (videoVo.getTempFile() != null) videoVo.getTempFile().delete();
                                videoVo.setTempFile(null);
                                redisTemplate.opsForList().rightPush("video_forward", JSONObject.toJSONString(videoVo));
                            }

                            /*if(b){
                                //redisTemplate.opsForList().remove("video_forward",-1,data);
                                ssuccessCacheFiles.add(videoVo.getVideo_path());
                            }else {
                                videoVo.setFile(null);
                                if (videoVo.getTempFile() != null)
                                    videoVo.getTempFile().delete();
                                videoVo.setTempFile(null);
                                redisTemplate.opsForList().rightPush("video_forward",JSONObject.toJSONString(videoVo));
                            }*/
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            if (linkedMultiValueMap != null) {
                                linkedMultiValueMap.clear();
                            }
                            videoVo.setFile(null);
                            if (videoVo.getTempFile() != null) {
                                videoVo.getTempFile().delete();
                            }
                            videoVo.setTempFile(null);
                        }
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Scheduled(fixedDelay = 2020L)
    @Async
    public void removeFiles() {
        int num = 0;
        if (!CollectionUtils.isEmpty(this.ssuccessCacheFiles)) {
            Iterator<String> files = this.ssuccessCacheFiles.iterator();
            while (files.hasNext() && num <= 200) {
                String path = files.next();
                try {
                    log.debug("rm -rf {}", path);
                    Runtime.getRuntime().exec("rm -rf " + path).waitFor();
                    //files.remove();
                    ssuccessCacheFiles.remove(path);
                    num++;
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }


    @Scheduled(fixedDelay = 2000L)
    @Async
    public void dataRetransmission() {
        try {
            while (true) {
                Long size = redisTemplate.opsForList().size("data_forward");
                log.info("当前分析数据总数量为:{}", size);
                if (size == null || size <= 0) {
                    if (dataForwardServiceExecutor.getActiveCount() == 0) {
                        try {
                            Runtime.getRuntime().exec("rm -rf " + this.picPath).waitFor();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    return;
                }
                log.info("json数据重发开始");
                for (int i = 0; i < 200; i++) {
                    Object item = redisTemplate.opsForList().rightPop("data_forward");
                    if (item == null) {
                        return;
                    }
                    String data = (String) item;
                    this.dataForwardServiceExecutor.submit(() -> {
                        JSONObject jSONObject = JSONObject.parseObject(data);
                        String forward_url = jSONObject.getString("forward_url");
                        String[] forwards = null;
                        if (!StringUtils.isEmpty(forward_url)) {
                            forwards = forward_url.split(";");
                        }
                        List<Map> pics;
                        StringBuilder failedUrl = new StringBuilder();
                        try {
                            pics = (List<Map>) jSONObject.get("pics");
                            RedisServiceImpl.this.putBase64ToPics(pics);
                            if (forwards != null && forwards.length > 0) {
                                String txt = jSONObject.toJSONString();
                                txt = txt.replace("\"{", "{").replace("}\"", "}");
                                for (String url : forwards) {
                                    if (!sendData(txt, url)) {
                                        failedUrl.append(url).append(";");
                                    }
                                }
                            }

                            forward_url = failedUrl.toString();
                            if (StringUtils.isEmpty(forward_url)) {
                                if (!CollectionUtils.isEmpty(pics)) {
                                    pics.forEach(pic -> {
                                        try {
                                            log.info("rm -rf {}", pic.get("src_url"));
                                            ssuccessCacheFiles.add(String.valueOf(pic.get("src_url")));
                                            Runtime.getRuntime().exec("rm -rf " + pic.get("src_url")).waitFor();
                                            pic.remove("pic_base64");
                                        } catch (Exception e) {
                                            e.printStackTrace();
                                        }
                                    });
                                }
                            } else {
                                jSONObject.remove("pics");
                                if (!CollectionUtils.isEmpty(pics)) pics.forEach(pic -> {
                                    try {
                                        pic.remove("pic_base64");
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                });
                                jSONObject.put("pics", pics);
                                jSONObject.put("forward_url", forward_url);
                                redisTemplate.opsForList().leftPush("data_forward", JSON.toJSON(jSONObject).toString());
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    });
                }
                while (dataForwardServiceExecutor.getActiveCount() > 0) {
                    Thread.sleep(1000L);
                }
            }

        } catch (Exception e) {
            log.error("数据重发异常：{}", e.getLocalizedMessage(), e);
        }
    }


//    @Scheduled(fixedDelay = 2000L)
//    public void dataRetransmission() {
//        try {
//            if (1 == this.dataSending)
//                return;
//            this.dataSending = 1;
//            Set<String> keys = this.redisTemplate.keys("*data_forward_*");
//            if (CollectionUtils.isEmpty(keys)) {
//                this.dataSending = 0;
//                if (CollectionUtils.isEmpty(this.errorCacheFiles))
//                    try {
//                        log.info("rm -rf {}", this.picPath);
//                        Runtime.getRuntime().exec("rm -rf " + this.picPath).waitFor();
//                    } catch (Exception e) {
//                        e.printStackTrace();
//                    }
//                return;
//            }
//            //log.info(");
//            Map.Entry<String, String> item = null;
//            int num = 0;
//            for (Iterator<String> it = keys.iterator(); it.hasNext(); ) {
//                final String _key = it.next();
//                this.dataForwardServiceExecutor.execute(new Runnable() {
//                    public void run() {
//                        String data = null;
//                        String key = "";
//                        JSONObject jSONObject = null;
//                        String forward_url = null;
//                        String[] forwards = null;
//                        List<Map> pics = null;
//                        StringBuffer sb = null;
//                        try {
//                            key = _key;
//                            data = (String)RedisServiceImpl.this.redisTemplate.opsForValue().get(key);
//                            if (RedisServiceImpl.this.redisTemplate.hasKey(key).booleanValue())
//                                RedisServiceImpl.this.redisTemplate.delete(key);
//                            if (StringUtils.isEmpty(data)) {
//                                data = null;
//                                key = "";
//                                return;
//                            }
//                            data = data.replace("\"{", "{").replace("}\"", "}");
//                            if (StringUtils.isEmpty(data)) {
//                                data = null;
//                                key = "";
//                                return;
//                            }
//                            jSONObject = JSONObject.parseObject(data);
//                            forward_url = jSONObject.getString("forward_url");
//                            if (!StringUtils.isEmpty(forward_url))
//                                forwards = forward_url.split(";");
//                            pics = (List<Map>)jSONObject.get("pics");
//                            RedisServiceImpl.this.putBase64ToPics(pics);
//                            if (forwards != null && forwards.length > 0) {
//                                sb = new StringBuffer();
//                                String txt = jSONObject.toJSONString();
//                                txt = txt.replace("\"{", "{").replace("}\"", "}");
//                                for (String url : forwards) {
//                                    if (!RedisServiceImpl.this.sendData(txt, url))
//                                        sb.append(url).append(";");
//                                }
//                            }
//                            if (sb == null || StringUtils.isEmpty(sb.toString())) {
//                                if (!CollectionUtils.isEmpty(pics))
//                                    pics.stream().forEach(pic -> {
//                                        try {
//                                            RedisServiceImpl.log.info("rm -rf {}", pic.get("src_url"));
//                                            RedisServiceImpl.this.errorCacheFiles.remove(String.valueOf(pic.get("src_url")));
//                                            RedisServiceImpl.this.ssuccessCacheFiles.add(String.valueOf(pic.get("src_url")));
//                                            Runtime.getRuntime().exec("rm -rf " + pic.get("src_url")).waitFor();
//                                            pic.remove("pic_base64");
//                                        } catch (Exception e) {
//                                            e.printStackTrace();
//                                        }
//                                    });
//                            } else if (!StringUtils.isEmpty(sb.toString())) {
//                                forward_url = sb.toString();
//                                jSONObject.remove("pics");
//                                if (!CollectionUtils.isEmpty(pics))
//                                    pics.stream().forEach(pic -> {
//                                        try {
//                                            pic.remove("pic_base64");
//                                        } catch (Exception e) {
//                                            e.printStackTrace();
//                                        }
//                                    });
//                                jSONObject.put("pics", pics);
//                                jSONObject.put("forward_url", forward_url);
//                                RedisServiceImpl.this.redisTemplate.opsForValue().set("data_forward_" + System.currentTimeMillis() + (Math.random() * 100.0D), JSON.toJSON(jSONObject).toString(), 3L, TimeUnit.DAYS);
//                            }
//                        } catch (Exception e) {
//                            e.printStackTrace();
//                        } finally {
//                            data = null;
//                            key = "";
//                            jSONObject = null;
//                            forward_url = null;
//                            forwards = null;
//                            pics = null;
//                            sb = null;
//                        }
//                    }
//                });
//                it.remove();
//                if (num > 500)
//                    break;
//                num++;
//            }
//        } catch (Exception e) {
//            log.error("{}", e.getLocalizedMessage(), e);
//        } finally {
//            this.dataSending = 0;
//        }
//    }

    private boolean sendData(String msg, String url) {
        try {
            HttpEntity<String> requestEntity = new HttpEntity(msg, (MultiValueMap) this.headers);
            ResponseEntity<String> responseEntity = this.restTemplate.exchange(url, HttpMethod.POST, requestEntity, String.class, new Object[0]);
            log.info("数据发送url{};http{};{}", new Object[]{url, Integer.valueOf(responseEntity.getStatusCode().value()), responseEntity.getBody()});
            if (200 == responseEntity.getStatusCode().value() && "200".equals(ecode((String) responseEntity.getBody())))
                return true;
            return false;
        } catch (Exception e) {
            log.error("重发异常:{};{}", new Object[]{url, e.getLocalizedMessage(), e});
            return false;
        }
    }

    private void putBase64ToPics(List<Map> pics) {
        if (!CollectionUtils.isEmpty(pics)) pics.stream().forEach(pic -> {
            String pic_base64 = null;
            try {
                pic_base64 = Base64PicUtil.downlodPicForBase64(String.valueOf(pic.get("src_url")));
                pic.put("pic_base64", pic_base64);
            } catch (IOException e) {
                e.printStackTrace();
            }
            pic_base64 = null;
        });
    }

    private boolean upload(VideoVo videoVo) {
        //File tempFile = null;
        try {
            /*MultipartFile videofile = videoVo.getFile();
            if (videofile == null || videofile.isEmpty())
                return true;
            String fileName = videoVo.getFile().getOriginalFilename();
            String tempFilePath = System.getProperty("java.io.tmpdir") + "/" + videoVo.getRefid() + ".mp4";
            System.out.println("tempFilePath=" + tempFilePath);
            tempFile = new File(tempFilePath);
            videofile.transferTo(tempFile);
            videoVo.setTempFile(tempFile);*/
            //RestTemplate restTemplate = new RestTemplate();

            LinkedMultiValueMap linkedMultiValueMap = new LinkedMultiValueMap();
            FileSystemResource resource = new FileSystemResource(videoVo.getVideo_path());
            linkedMultiValueMap.add("refid", videoVo.getRefid());
            linkedMultiValueMap.add("format", videoVo.getFormat());
            linkedMultiValueMap.add("location_id", videoVo.getLocation_id());
            linkedMultiValueMap.add("file", resource);
            HttpEntity<MultiValueMap<String, Object>> formEntity = new HttpEntity(linkedMultiValueMap, (MultiValueMap) videoheaders);
            ResponseEntity<String> responseEntity = restTemplate.postForEntity(this.video_forward_url, formEntity, String.class, new Object[0]);
            log.info("视频重发地址:{};refid:{};http:{};响应信息{}", new Object[]{this.video_forward_url, videoVo.getRefid(), Integer.valueOf(responseEntity.getStatusCode().value()), responseEntity.getBody()});
            if (200 == responseEntity.getStatusCode().value() && "200".equals(ecode(responseEntity.getBody()))) {
                //System.out.println("return true");
                return true;
            }
            return false;
        } catch (Exception e) {
            log.error("异常{};{}", new Object[]{this.video_forward_url, e.getLocalizedMessage()}, e);
            return false;
        }
    }

    private String ecode(String result) {
        Pattern pattern = Pattern.compile("\"code\":[ ]*\"[0-9a-zA-Z_]+\"");
        Matcher matcher = pattern.matcher(result);
        if (matcher.find()) {
            String code = matcher.group().replace("\"", "").split(":")[1];
            return code.trim();
        }
        pattern = Pattern.compile("\"code\":[ ]*[0-9a-zA-Z_]+");
        matcher = pattern.matcher(result);
        if (matcher.find()) {
            String code = matcher.group().replace("\"", "").split(":")[1];
            return code.trim();
        }
        pattern = Pattern.compile("\"ecode\":[ ]*\"[0-9a-zA-Z_]+\"");
        matcher = pattern.matcher(result);
        if (matcher.find()) {
            String code = matcher.group().replace("\"", "").split(":")[1];
            return code.trim();
        }

        pattern = Pattern.compile("\"ecode\":[ ]*[0-9a-zA-Z_]+");
        matcher = pattern.matcher(result);
        if (matcher.find()) {
            String code = matcher.group().replace("\"", "").split(":")[1];
            return code.trim();
        }
        return "";
    }
}