feat(driver): add WIA data service, controller, and receive thread in dc3-group

2026-11-03 03:32:08 +08:00 · 2019-06-26 23:47:25 +08:00
parent 4f5ed7a1b9
commit 7fb751e507
20 changed files with 396 additions and 22 deletions
@@ -8,6 +8,7 @@ import com.pnoker.transfer.rtmp.feign.RtmpFeignApi;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.env.Environment;
+import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;

 import java.util.List;
@@ -53,4 +54,15 @@ public class RtmpController extends BaseController implements RtmpFeignApi {
        List<Rtmp> list = rtmpService.list(queryWrapper);
        return list;
    }
+
+    @RequestMapping("/insert")
+    public void insert() {
+        for (int i = 0; i < 1000000; i++) {
+            Rtmp wiaData = new Rtmp(i);
+            rtmpService.insert(wiaData);
+            if (i % 100 == 0) {
+                log.info("完成：{},{}%", i, i / 1000000 * 10);
+            }
+        }
+    }
 }
@@ -33,4 +33,6 @@ public interface RtmpService {
     * @return
     */
    List<Rtmp> list(Wrapper<Rtmp> wrapper);
+
+    int insert(Rtmp rtmp);
 }
@@ -39,4 +39,9 @@ public class RtmpServiceImpl implements RtmpService {
    public List<Rtmp> list(Wrapper<Rtmp> wrapper) {
        return rtmpMapper.selectList(wrapper);
    }
+
+    @Override
+    public int insert(Rtmp rtmp) {
+        return rtmpMapper.insert(rtmp);
+    }
 }
@@ -40,8 +40,7 @@ public class BaseController {
     * @return
     */
    protected HttpServletRequest getRequest() {
-        HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
-        return request;
+        return ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
    }

    /**
@@ -55,7 +54,7 @@ public class BaseController {
        try {
            path = URLDecoder.decode(path, "UTF-8") + resource;
        } catch (UnsupportedEncodingException e) {
-            log.error("BaseController.getResourcePath.error {}", e.getMessage(), e);
+            log.error(e.getMessage(), e);
        }
        return path;
    }
@@ -1,13 +0,0 @@
-package com.pnoker.device.group.bean;
-
-import java.util.concurrent.LinkedBlockingQueue;
-
-/**
- * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
- * <p>Author     : Pnoker
- * <p>Email      : pnokers@gmail.com
- * <p>Description:
- */
-public class Global {
-    public static LinkedBlockingQueue<Thread> wiaReceiveThreadQueue = new LinkedBlockingQueue(20);
-}
@@ -0,0 +1,56 @@
+package com.pnoker.device.group.controller;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.pnoker.common.base.BaseController;
+import com.pnoker.device.group.model.wia.WiaData;
+import com.pnoker.device.group.service.wia.WiaDataService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description:
+ */
+@Slf4j
+@RestController
+public class WiaDataController extends BaseController {
+    @Autowired
+    private WiaDataService wiaDataService;
+
+    @RequestMapping("/list")
+    public List<WiaData> list() {
+        QueryWrapper<WiaData> queryWrapper = new QueryWrapper<>();
+        List<WiaData> list = wiaDataService.list(queryWrapper);
+        return list.subList(0, 100);
+    }
+
+    @Transactional
+    @RequestMapping("/insert/{num}")
+    public String insert(@PathVariable int num) {
+        LinkedBlockingQueue blockingQueue = new LinkedBlockingQueue(5000);
+        if(blockingQueue.size()>1000){
+            
+        }
+        long start = System.currentTimeMillis();
+        for (int i = 0; i < num; i++) {
+            WiaData wiaData = new WiaData();
+            wiaData.setId(i);
+            wiaData.setVariableId(0);
+            wiaData.setValue(1);
+            wiaData.setTime(System.currentTimeMillis());
+            wiaDataService.insert(wiaData);
+            log.info("完成：{}", i);
+        }
+        long end = System.currentTimeMillis();
+        return ((end - start) / 1000) + "";
+    }
+}
@@ -0,0 +1,73 @@
+package com.pnoker.device.group.global;
+
+import com.pnoker.device.group.model.wia.WiaData;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description: 全局 队列集合和线程池，用于存放线程队列、数据队列以及线程池
+ */
+@Slf4j
+public class Queues {
+    /**
+     * 核心线程池大小
+     */
+    public static int corePoolSize = 4;
+
+    /**
+     * 最大线程池大小
+     */
+    public static int maximumPoolSize = 32;
+
+    /**
+     * 线程最大空闲时间
+     */
+    public static long keepAliveTime = 10;
+
+    /**
+     * 时间单位，秒
+     */
+    public static TimeUnit unit = TimeUnit.SECONDS;
+
+    /**
+     * 用于存放 wia 数据，支持2000并发，该队列用于优化大批量数据入库性能
+     */
+    public static LinkedBlockingQueue<WiaData> wiaDataQueue = new LinkedBlockingQueue(2000);
+
+    /**
+     * 用于存放 wia 数据采集线程，为了性能，最多一次性放入32个
+     */
+    public static LinkedBlockingQueue<Runnable> wiaReceiveThreadQueue = new LinkedBlockingQueue(32);
+
+    /**
+     * 数据采集线程池，用于全部数采线程使用
+     */
+    public static ThreadPoolExecutor receivePoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, wiaReceiveThreadQueue, new ReceiveTreadFactory(), new IgnorePolicy());
+
+    static class ReceiveTreadFactory implements ThreadFactory {
+        private final AtomicInteger mThreadNum = new AtomicInteger(1);
+
+        @Override
+        public Thread newThread(Runnable runnable) {
+            Thread thread = new Thread(runnable, "dc3-group-thread-" + mThreadNum.getAndIncrement());
+            log.info("{} has been created", thread.getName());
+            return thread;
+        }
+    }
+
+    static class IgnorePolicy implements RejectedExecutionHandler {
+        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
+            doLog(r, e);
+        }
+
+        private void doLog(Runnable r, ThreadPoolExecutor e) {
+            log.error("{} rejected,completedTaskCount:{}", r.toString(), e.getCompletedTaskCount());
+        }
+    }
+
+}
@@ -0,0 +1,15 @@
+package com.pnoker.device.group.mapper.wia;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.pnoker.device.group.model.wia.WiaData;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description: Wia数据 数据库操作接口
+ */
+@Mapper
+public interface WiaDataMapper extends BaseMapper<WiaData> {
+}
@@ -0,0 +1,17 @@
+package com.pnoker.device.group.model.wia;
+
+import lombok.Data;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description:
+ */
+@Data
+public class WiaData {
+    private long id;
+    private long variableId;
+    private float value;
+    private long time;
+}
@@ -9,7 +9,7 @@ import lombok.Data;
 * <p>Description: Wia 数采设备信息
 */
@Data
-public class Device {
+public class WiaDevice {
    private long id;
    private long gatewayId;
    private String name;
@@ -9,7 +9,7 @@ import lombok.Data;
 * <p>Description: Wia 网关设备信息
 */
@Data
-public class Gataway {
+public class WiaGataway {
    private long id;
    private String ipAddress;
    private int port;
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2018 Google LLC. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.pnoker.device.group.service.wia;
+
+import com.baomidou.mybatisplus.core.conditions.Wrapper;
+import com.pnoker.device.group.model.wia.WiaData;
+
+import java.util.List;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description: Rtmp 服务接口
+ */
+public interface WiaDataService {
+    /**
+     * 获取 Rtmp任务 数据列表
+     *
+     * @return
+     */
+    List<WiaData> list(Wrapper<WiaData> wrapper);
+
+    int insert(WiaData wiaData);
+}
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2018 Google LLC. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.pnoker.device.group.service.wia.impl;
+
+import com.baomidou.mybatisplus.core.conditions.Wrapper;
+import com.pnoker.device.group.mapper.wia.WiaDataMapper;
+import com.pnoker.device.group.model.wia.WiaData;
+import com.pnoker.device.group.service.wia.WiaDataService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description: Rtmp 接口实现
+ */
+@Service
+public class WiaDataServiceImpl implements WiaDataService {
+    @Autowired
+    private WiaDataMapper wiaDataMapper;
+
+    @Override
+    public List<WiaData> list(Wrapper<WiaData> wrapper) {
+        return wiaDataMapper.selectList(wrapper);
+    }
+
+    @Override
+    public int insert(WiaData wiaData) {
+        return wiaDataMapper.insert(wiaData);
+    }
+}
@@ -0,0 +1,32 @@
+package com.pnoker.device.group.thread;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description: 数据采集线程基类，数采线程请务必继承此类
+ */
+public class ReceiveThread implements Runnable {
+    /**
+     * 线程名称，建议写明白
+     */
+    private String name;
+
+    public ReceiveThread(String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    @Override
+    public void run() {
+
+    }
+
+    @Override
+    public String toString() {
+        return "ReceiveThread [name=" + name + "]";
+    }
+}
@@ -0,0 +1,19 @@
+package com.pnoker.device.group.thread;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description: wia 数据采集线程
+ */
+public class WiaReceiveThread extends ReceiveThread {
+
+    public WiaReceiveThread(String name) {
+        super(name);
+    }
+
+    @Override
+    public void run() {
+
+    }
+}
@@ -6,10 +6,10 @@ server:
    io-threads: 2
 spring:
  application:
-    name: dc3-group
+    name: dc3-device-group
  datasource:
    driver-class-name: org.sqlite.JDBC
-    url: jdbc:sqlite::resource:sqlite/dc3-group.sqlite
+    url: jdbc:sqlite::resource:sqlite/dc3-group.db
    username:
    password:
    sql-script-encoding: utf-8
@@ -6,7 +6,7 @@ server:
    io-threads: 2
 spring:
  application:
-    name: dc3-manager
+    name: dc3-device-manager

 management:
  endpoints:
@@ -0,0 +1,72 @@
+### 剪切&合并视频
+
+```bash
+//截取从头开始的30s
+ffmpeg -ss 00:00:00 -t 00:00:30 -i keyoutput.mp4 -vcodec copy -acodec copy split.mp4
+//截取从30s开始的30s
+ffmpeg -ss 00:00:30 -t 00:00:30 -i keyoutput.mp4 -vcodec copy -acodec copy split1.mp4
+//进行视频的合并
+ffmpeg -f concat -i list.txt -c copy concat.mp4
+```
+
+### 其他常见指令
+
+```bash
+// 去掉视频中的音频
+ffmpeg -i input.mp4 -vcodec copy -an output.mp4
+// -an: 去掉音频；-vcodec:视频选项，一般后面加copy表示拷贝
+
+// 提取视频中的音频
+ffmpeg -i input.mp4 -acodec copy -vn output.mp3
+// -vn: 去掉视频；-acodec: 音频选项， 一般后面加copy表示拷贝
+
+// 音视频合成
+ffmpeg -y –i input.mp4 –i input.mp3 –vcodec copy –acodec copy output.mp4
+// -y 覆盖输出文件
+
+//剪切视频
+ffmpeg -ss 0:1:30 -t 0:0:20 -i input.mp4 -vcodec copy -acodec copy output.mp4
+// -ss 开始时间; -t 持续时间
+
+// 视频截图
+ffmpeg –i test.mp4 –f image2 -t 0.001 -s 320x240 image-%3d.jpg
+// -s 设置分辨率; -f 强迫采用格式fmt;
+
+// 视频分解为图片
+ffmpeg –i test.mp4 –r 1 –f image2 image-%3d.jpg
+// -r 指定截屏频率
+
+// 将图片合成视频
+ffmpeg -f image2 -i image%d.jpg output.mp4
+
+//视频拼接
+ffmpeg -f concat -i filelist.txt -c copy output.mp4
+
+// 将视频转为gif
+ffmpeg -i input.mp4 -ss 0:0:30 -t 10 -s 320x240 -pix_fmt rgb24 output.gif
+// -pix_fmt 指定编码
+
+// 将视频前30帧转为gif
+ffmpeg -i input.mp4 -vframes 30 -f gif output.gif
+
+// 旋转视频
+ffmpeg -i input.mp4 -vf rotate=PI/2 output.mp4
+
+// 缩放视频
+ffmpeg -i input.mp4 -vf scale=iw/2:-1 output.mp4
+// iw 是输入的宽度， iw/2就是一半;-1 为保持宽高比
+
+//视频变速
+ffmpeg -i input.mp4 -filter:v setpts=0.5*PTS output.mp4
+
+//音频变速
+ffmpeg -i input.mp3 -filter:a atempo=2.0 output.mp3
+
+//音视频同时变速，但是音视频为互倒关系
+ffmpeg -i input.mp4 -filter_complex "[0:v]setpts=0.5*PTS[v];[0:a]atempo=2.0[a]" -map "[v]" -map "[a]" output.mp4
+
+
+// 视频添加水印
+ffmpeg -i input.mp4 -i logo.jpg -filter_complex [0:v][1:v]overlay=main_w-overlay_w-10:main_h-overlay_h-10[out] -map [out] -map 0:a -codec:a copy output.mp4
+// main_w-overlay_w-10 视频的宽度-水印的宽度-水印边距；
+```
@@ -33,7 +33,7 @@
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
-        <version>2.1.5.RELEASE</version>
+        <version>2.1.6.RELEASE</version>
    </parent>

    <!-- Multiple Module -->