feaets

pom.xml

@@ -26,6 +26,11 @@
 			<artifactId>fastjson</artifactId>
 			<version>1.2.4</version>
 		</dependency>
+		<dependency>
+			<groupId>io.netty</groupId>
+			<artifactId>netty-all</artifactId>
+			<version>4.1.55.Final</version>
+		</dependency>
 		<!--mybaits-->
 		<dependency>
 			<groupId>org.mybatis.spring.boot</groupId>

src/main/java/com/yunzhi/inte/InteApplication.java

@@ -1,15 +1,18 @@
 package com.yunzhi.inte;
 
-import org.mybatis.spring.annotation.MapperScan;
+import com.yunzhi.inte.rtsp.MonitorServer;
 import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.ConfigurableApplicationContext;
 
 @SpringBootApplication
 public class InteApplication {
+	public static ConfigurableApplicationContext ac;
 
 	public static void main(String[] args) {
-		SpringApplication.run(InteApplication.class, args);
-	}
+		ac = SpringApplication.run(InteApplication.class, args);
+		System.out.println("------------");
 
+//		ac.getBean(MonitorServer.class).start();
+	}
 }

src/main/java/com/yunzhi/inte/controller/MonitorController.java

@@ -30,11 +30,13 @@ public class MonitorController {
      */
     @RequestMapping(value="/monitor",method= RequestMethod.GET)
     public ResponseBean MonitorPage(@RequestParam(value ="pageNum",defaultValue = "1") Integer pageNum,
-                                   @RequestParam(value ="pageSize",defaultValue = "10") Integer pageSize) throws Exception{
+                                   @RequestParam(value ="pageSize",defaultValue = "10") Integer pageSize,
+                                    @RequestParam(value ="status", required = false) String status) throws Exception{
 
         IPage<Monitor> pg = new Page<>(pageNum, pageSize);
-//        QueryWrapper<Info> queryWrapper = new QueryWrapper<>();
-//        queryWrapper.orderByDesc("qz");
+        QueryWrapper<Monitor> queryWrapper = new QueryWrapper<>();
+        queryWrapper.eq(null != status && !"".equals(status), "status", status);
+        queryWrapper.orderByDesc("id");
 
         IPage<Monitor> result = monitorService.page(pg);
         return ResponseBean.success(result);

src/main/java/com/yunzhi/inte/rtsp/ChannelGroupHolder.java

@@ -0,0 +1,72 @@
+package com.yunzhi.inte.rtsp;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelId;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * 保存群组
+ */
+@Slf4j
+public class ChannelGroupHolder {
+
+    private static Map<String, ChannelGroup> channelGroups = new HashMap<>();
+    private static Map<String, String> channelTagMapping = new HashMap<>();
+
+    public static Map<String, ChannelGroup> getChannelGroups() {
+        return channelGroups;
+    }
+
+    /**
+     * 添加channel到群组
+     */
+    public static synchronized void add(String tag, Channel channel) {
+        ChannelGroup group = channelGroups.get(tag);
+        if (group == null) {
+            group = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
+            log.info("创建新群组：{}", tag);
+            channelGroups.put(tag, group);
+        }
+        group.add(channel);
+        ChannelId channelId = channel.id();
+        channelTagMapping.put(channelId.toString(), tag);
+        log.info("群组{}加入新channel：{}", tag, channelId);
+    }
+
+    /**
+     * 根据IP获取群组
+     */
+    public static ChannelGroup get(String tag) {
+        return channelGroups.get(tag);
+    }
+
+    /**
+     * 移除指定channel，如果群组变为空，返回对应IP
+     */
+    public static String remove(Channel channel) {
+        String channelId = channel.id().toString();
+        String tag = channelTagMapping.get(channelId);
+        channelTagMapping.remove(channelId);
+        if (tag != null && !channelTagMapping.containsValue(tag)) {
+            log.info("移除channel：{}，所在群组：{}", channelId, tag);
+            return tag;
+        }
+        return null;
+    }
+
+    /**
+     * 关闭所有群组
+     */
+    public static void close() {
+        for (ChannelGroup group : channelGroups.values()) {
+            group.close();
+        }
+    }
+
+}

src/main/java/com/yunzhi/inte/rtsp/FFmpeg.java

@@ -0,0 +1,26 @@
+package com.yunzhi.inte.rtsp;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+
+@Slf4j
+@Component
+public class FFmpeg {
+    @Value("${my.minitor.port}")
+    Integer port;
+
+    @Value("${my.minitor.ffmpeg}")
+    String ffmpegBin;
+
+    public Process run(String source) throws IOException {
+        String command = ffmpegBin + " -f rtsp -rtsp_transport tcp -i " + source
+                + " -strict -2 -c:v libx264 -vsync 2 -c:a aac "
+                + "-f mpegts -codec:v mpeg1video -s 640x480 -b:v 1000k -bf 0 http://127.0.0.1:" + port
+                + "/push/" + source;
+        log.info(command);
+        return Runtime.getRuntime().exec(command);
+    }
+}

src/main/java/com/yunzhi/inte/rtsp/HttpRequestHandler.java

@@ -0,0 +1,107 @@
+package com.yunzhi.inte.rtsp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.ChannelGroupFuture;
+import io.netty.handler.codec.http.*;
+import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+
+/**
+ * 处理HTTP请求
+ */
+@Slf4j
+@SuppressWarnings("deprecation")
+public class HttpRequestHandler extends SimpleChannelInboundHandler<HttpObject> {
+    private final String wsUri;
+    private String monitorIp;
+
+    public HttpRequestHandler(String wsUri) {
+        this.wsUri = wsUri;
+    }
+
+    @Override
+    public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
+        if (msg instanceof HttpRequest) {
+            HttpRequest request = (HttpRequest) msg;
+            String uri = request.getUri();
+            log.info("收到HTTP请求：{}", uri);
+
+            // 处理WebSocket请求
+            if (uri.contains(wsUri)) {
+                String ip = uri.split(wsUri + "/")[1];
+                ChannelGroupHolder.add(ip, ctx.channel());
+                TransferRtspHelper.open(ip);
+                ctx.fireChannelRead(new DefaultFullHttpRequest(request.protocolVersion(), request.method(), wsUri,
+                        Unpooled.buffer(), request.headers(), request.headers()));
+            } else if (uri.contains("/push/")) {
+                monitorIp = uri.split("/push/")[1];
+                log.info("开始接收转码推流：{}", monitorIp);
+            }
+            // 处理HTTP请求
+            else {
+                FullHttpResponse response = null;
+                response = generateResponse(request, null);
+                ctx.write(response);
+                ChannelFuture future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+                if (!HttpHeaders.isKeepAlive(request)) {
+                    future.addListener(ChannelFutureListener.CLOSE);
+                }
+            }
+        } else if (msg instanceof HttpContent && !(msg instanceof LastHttpContent)) {
+            HttpContent chunk = (HttpContent) msg;
+            ChannelGroup group = ChannelGroupHolder.get(monitorIp);
+            ChannelGroupFuture groupFuture = group.writeAndFlush(new BinaryWebSocketFrame(chunk.retain().content()));
+            groupFuture.sync();
+        }
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        cause.printStackTrace();
+        ctx.close();
+    }
+
+    /**
+     * 生成HTTP响应
+     */
+    private FullHttpResponse generateResponse(HttpRequest request, RandomAccessFile file) throws IOException {
+        ByteBuf buf = generateByteBuf(file);
+        FullHttpResponse response = new DefaultFullHttpResponse(request.getProtocolVersion(), HttpResponseStatus.OK,
+                buf);
+        if (HttpHeaders.isKeepAlive(request)) {
+            response.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
+        }
+        response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, buf.writerIndex());
+        return response;
+    }
+
+    /**
+     * 生成HTTP响应内容
+     */
+    private ByteBuf generateByteBuf(RandomAccessFile file) throws IOException {
+        ByteBuf buf = Unpooled.buffer(0);
+        if (file != null) {
+            int length = (int) file.length();
+            byte[] fileByte = new byte[length];
+            file.read(fileByte);
+            file.close();
+
+            buf = Unpooled.buffer(length);
+            for (int i = 0; i < length; i++) {
+                buf.writeByte(fileByte[i]);
+            }
+        }
+        return buf;
+    }
+
+}

src/main/java/com/yunzhi/inte/rtsp/MonitorServer.java

@@ -0,0 +1,54 @@
+package com.yunzhi.inte.rtsp;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+/**
+ * 引导服务器
+ */
+@Slf4j
+@Component
+public class MonitorServer extends Thread {
+    @Value("${my.minitor.port}")
+    Integer port;
+
+    @Autowired
+    FFmpeg fFmpeg;
+
+    public void run() {
+        TransferRtspHelper.TransferTask.fFmpeg = fFmpeg;
+        EventLoopGroup group = new NioEventLoopGroup();
+        EventLoopGroup workerGroup = new NioEventLoopGroup();
+
+        ServerBootstrap bootstrap = new ServerBootstrap();
+        bootstrap.group(group, workerGroup).channel(NioServerSocketChannel.class)
+                .childHandler(new MonitorServerInitializer());
+
+        try {
+            Channel channel = bootstrap.bind(port).sync().channel();
+
+            TransferRtspHelper.schedule();
+            log.info("服务器开始运行......[:{}]", port);
+
+            channel.closeFuture().sync();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            group.shutdownGracefully();
+            workerGroup.shutdownGracefully();
+            ChannelGroupHolder.close();
+            log.info("服务器已关闭");
+        }));
+    }
+}

src/main/java/com/yunzhi/inte/rtsp/MonitorServerInitializer.java

@@ -0,0 +1,24 @@
+package com.yunzhi.inte.rtsp;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.http.HttpContentCompressor;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+
+/**
+ * 初始化ChannelPipeline
+ */
+public class MonitorServerInitializer extends ChannelInitializer<Channel> {
+
+    @Override
+    public void initChannel(Channel ch) {
+        ChannelPipeline pipeline = ch.pipeline();
+        pipeline.addLast(new HttpServerCodec());
+        pipeline.addLast(new HttpContentCompressor());
+        pipeline.addLast(new HttpRequestHandler("/ws"));
+        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
+        pipeline.addLast(new WebSocketHandler());
+    }
+}

src/main/java/com/yunzhi/inte/rtsp/TransferRtspHelper.java

@@ -0,0 +1,147 @@
+package com.yunzhi.inte.rtsp;
+
+import com.yunzhi.inte.InteApplication;
+import io.netty.channel.group.ChannelGroup;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 转码
+ */
+@Slf4j
+public class TransferRtspHelper {
+
+    private static final ScheduledExecutorService SCHEDULE = Executors.newScheduledThreadPool(2);
+    private static final ExecutorService PROCESS_EXECUTOR = Executors.newCachedThreadPool();
+    private static Map<String, Transfer> transferMap = new HashMap<>();
+
+    /**
+     * 开启
+     */
+    public static synchronized void open(String ip) throws IOException {
+        Transfer transfer = transferMap.get(ip);
+        if (transfer == null) {
+            transfer = new Transfer();
+            transfer.setIp(ip);
+            transfer.setStartTime(new Date());
+            transfer.setContinueTime(new Date());
+            transfer.setCount(1);
+            PROCESS_EXECUTOR.submit(new TransferTask(transfer));
+            transferMap.put(ip, transfer);
+        } else {
+            transfer.setContinueTime(new Date());
+            transfer.incCount();
+        }
+        log.info("开始转码：{}", transfer);
+    }
+
+    /**
+     * 关闭
+     */
+    public static synchronized void close(String ip) {
+        Transfer transfer = transferMap.get(ip);
+        if (transfer != null) {
+            transfer.decCount();
+            if (transfer.getCount() <= 0) {
+                Process process = transfer.getProcess();
+                if (process != null && process.isAlive()) {
+                    process.destroy();
+                }
+                transferMap.remove(ip);
+            }
+        }
+        log.info("结束转码：{}", transfer);
+    }
+
+    /**
+     * 若当前有在转码，打印相关信息
+     */
+    public static void report() {
+        Map<String, ChannelGroup> channelGroups = ChannelGroupHolder.getChannelGroups();
+        transferMap.forEach((k, v) -> {
+            ChannelGroup group = channelGroups.get(k);
+            if (group != null && !group.isEmpty()) {
+                log.info("当前转码中：{}", v);
+            }
+        });
+    }
+
+    /**
+     * 如果没有对应channel，停止转码，如果前端页面还开着，会自动建立回连接
+     */
+    public static void clear() {
+        List<String> toClearIp = new ArrayList<>();
+        Map<String, ChannelGroup> channelGroups = ChannelGroupHolder.getChannelGroups();
+        transferMap.forEach((k, v) -> {
+            ChannelGroup group = channelGroups.get(k);
+            if (group == null || group.isEmpty()
+                    || System.currentTimeMillis() - v.getContinueTime().getTime() > 1800_000L) {
+                toClearIp.add(k);
+            }
+        });
+        if (!toClearIp.isEmpty()) {
+            log.info("关闭以下无效转码：{}", toClearIp);
+            toClearIp.forEach(TransferRtspHelper::close);
+        }
+    }
+
+    /**
+     * 开启定时器
+     */
+    public static void schedule() {
+        SCHEDULE.scheduleAtFixedRate(TransferRtspHelper::report, 1L, 5L, TimeUnit.MINUTES);
+        SCHEDULE.scheduleAtFixedRate(TransferRtspHelper::clear, 60L, 5L, TimeUnit.SECONDS);
+    }
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    private static class Transfer {
+        private String ip;
+        private Process process;
+        private int count;
+        private Date startTime;
+        private Date continueTime;
+
+        public void incCount() {
+            count++;
+        }
+
+        public void decCount() {
+            count--;
+        }
+    }
+
+    @Data
+    @NoArgsConstructor
+    @AllArgsConstructor
+    public static class TransferTask implements Runnable {
+
+        private Transfer transfer;
+
+        public static FFmpeg fFmpeg;
+
+        @SneakyThrows
+        @Override
+        public void run() {
+            try {
+                String ip = transfer.getIp();
+                Process process = fFmpeg.run(ip);
+                transfer.setProcess(process);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+}

src/main/java/com/yunzhi/inte/rtsp/WebSocketHandler.java

@@ -0,0 +1,40 @@
+package com.yunzhi.inte.rtsp;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * 处理文本帧
+ */
+@Slf4j
+public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
+
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
+            log.info("WebSocket连接成功：{}", ctx.channel().id());
+        } else {
+            super.userEventTriggered(ctx, evt);
+        }
+    }
+
+    @Override
+    public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        Channel channel = ctx.channel();
+        log.info("websocket连接断开：{}", channel.id());
+        String toDeleteTag = ChannelGroupHolder.remove(channel);
+        if (toDeleteTag != null) {
+            TransferRtspHelper.close(toDeleteTag);
+        }
+        ctx.fireChannelInactive();
+    }
+
+}

src/main/resources/application.yml

@@ -34,6 +34,9 @@ my:
   server-base: http://192.168.89.147:8087/api
   upload-path: E:\work\jgz\upload
   static-prefix: /assets
+  minitor:
+    port: 7777
+    ffmpeg: D:\applications\ffmpeg-5.1-essentials_build\bin\ffmpeg.exe
 
 ###
 mybatis-plus: