feat(driver): add rtmp transcoding service and netty tcp socket client/server implementation

2026-11-01 10:41:52 +08:00 · 2019-04-30 23:40:17 +08:00
parent 56f70c9524
commit ea91e898bd
17 changed files with 506 additions and 0 deletions
@@ -0,0 +1,7 @@
+target/
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>com.pnoker.rtmp</groupId>
+    <artifactId>dc3-rtmp</artifactId>
+    <packaging>jar</packaging>
+
+    <description>iot-dc3 RTSP流转RTMP流 服务。</description>
+
+    <parent>
+        <groupId>com.pnoker.service</groupId>
+        <artifactId>dc3-service</artifactId>
+        <version>3.0</version>
+    </parent>
+
+    <dependencies>
+
+    </dependencies>
+
+    <build>
+        <finalName>${project.artifactId}</finalName>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
@@ -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.rtsp;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
+import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.springframework.scheduling.annotation.EnableAsync;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description:
+ */
+@EnableAsync
+@EnableFeignClients
+@EnableEurekaClient
+@SpringBootApplication
+public class RtmpApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(RtmpApplication.class, args);
+    }
+}
@@ -0,0 +1,13 @@
+package com.pnoker.rtsp.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 CommandTask {
+    public static LinkedBlockingQueue taskQueue = new LinkedBlockingQueue(20);
+}
@@ -0,0 +1,32 @@
+package com.pnoker.rtsp.bean;
+
+import com.pnoker.rtsp.handle.OutputHandle;
+import lombok.Data;
+
+import java.io.IOException;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description: 任务描述实体类
+ */
+@Data
+public class Tasker {
+    private String id;
+    private String command;
+    private Process process;
+    private OutputHandle outputHandle;
+
+    public Tasker(String id, String command) {
+        this.id = id;
+        this.command = command;
+    }
+
+    public void start() throws IOException {
+        Runtime runtime = Runtime.getRuntime();
+        this.process = runtime.exec(this.command);
+        this.outputHandle = new OutputHandle(this.process);
+        new Thread(this.outputHandle).start();
+    }
+}
@@ -0,0 +1,53 @@
+package com.pnoker.rtsp.command;
+
+
+import lombok.Data;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description:
+ */
+@Data
+public class CommandBuilder {
+    private String cmd;
+    private StringBuilder stringBuilder;
+
+    public CommandBuilder(String path) {
+        this.stringBuilder = new StringBuilder(path);
+    }
+
+    public CommandBuilder create(String exe) {
+        if (null != stringBuilder) {
+            stringBuilder.append(exe);
+        }
+        return this;
+    }
+
+    public CommandBuilder add(String cmd) {
+        if (null != stringBuilder) {
+            stringBuilder.append(" " + cmd);
+        }
+        return this;
+    }
+
+    public CommandBuilder add(String key, String cmd) {
+        return add(key).add(cmd);
+    }
+
+    public void build() {
+        cmd = stringBuilder.toString();
+    }
+
+    public static void main(String[] args) {
+        CommandBuilder builder = new CommandBuilder("D:/Documents/FFmpeg/bin/");
+        builder.create("ffmpeg")
+                .add("-i", "rtsp://184.72.239.149/vod/mp4://BigBuckBunny_175k.mov")
+                .add("-vcodec", "copy")
+                .add("-acodec", "copy")
+                .add("-f", "flv")
+                .add("-y", "rtmp://106.14.182.20:1935/rtmp/tomcat").build();
+        System.out.println(builder.getCmd());
+    }
+}
@@ -0,0 +1,65 @@
+package com.pnoker.rtsp.handle;
+
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description: Process输出处理线程
+ */
+@Slf4j
+@Data
+public class OutputHandle implements Runnable {
+    /**
+     * 运行状态，false：运行失败
+     */
+    private volatile boolean status = true;
+    private Process process;
+
+    public OutputHandle(Process process) {
+        this.process = process;
+    }
+
+    @Override
+    public void run() {
+        if (status) {
+            read(process.getErrorStream());
+        }
+    }
+
+    public void read(InputStream inputStream) {
+        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
+        String line;
+        try {
+            if (!status) {
+                throw new IOException();
+            }
+            while (null != (line = reader.readLine())) {
+                handle(line);
+            }
+        } catch (IOException e) {
+            log.error(e.getMessage(), e);
+        } finally {
+            try {
+                inputStream.close();
+                process.destroyForcibly();
+            } catch (IOException e) {
+                log.error(e.getMessage(), e);
+            }
+        }
+    }
+
+    public void handle(String message) {
+        message = message.toLowerCase();
+        if (message.contains("fail") || message.contains("miss")) {
+            this.status = false;
+        }
+    }
+}
@@ -0,0 +1,18 @@
+server:
+  port: 8806
+  undertow:
+    worker-threads: 20
+    buffer-size: 512
+    io-threads: 2
+
+feign:
+  hystrix:
+    enabled: true
+
+logging:
+  file: dc3/logs/dc3-rtmp-dev.log
+
+eureka:
+  client:
+    serviceUrl:
+      defaultZone: http://localhost:8800/eureka/
@@ -0,0 +1,18 @@
+server:
+  port: 8806
+  undertow:
+    worker-threads: 20
+    buffer-size: 512
+    io-threads: 2
+
+feign:
+  hystrix:
+    enabled: true
+
+logging:
+  file: dc3/logs/dc3-rtmp-product.log
+
+eureka:
+  client:
+    serviceUrl:
+      defaultZone: http://localhost:8800/eureka/,http://localhost:8801/eureka/
@@ -0,0 +1,5 @@
+spring:
+  application:
+    name: dc3-service-rtsp-rtmp
+  profiles:
+    active: dev
@@ -0,0 +1,9 @@
+
+${AnsiColor.BRIGHT_CYAN}Spring Boot${spring-boot.formatted-version}${AnsiColor.BRIGHT_RED}
+    .___     ________    .__        __            .__          __    _____
+  __| _/____ \_____  \   |__| _____/  |_   ______ |  | _____ _/  |__/ ____\___________  _____
+ / __ |/ ___\  _(__  <   |  |/  _ \   __\  \____ \|  | \__  \\   __\   __\/  _ \_  __ \/     \
+/ /_/ \  \___ /       \  |  (  <_> )  |    |  |_> >  |__/ __ \|  |  |  | (  <_> )  | \/  Y Y  \
+\____ |\___  >______  /  |__|\____/|__|    |   __/|____(____  /__|  |__|  \____/|__|  |__|_|  /
+     \/    \/       \/                     |__|             \/                              \/
+${AnsiColor.BRIGHT_CYAN}Copyright c 2018-2020 The ${AnsiColor.BRIGHT_YELLOW}Pnoker${AnsiColor.BRIGHT_CYAN} Authors
@@ -0,0 +1,79 @@
+package com.pnoker.socket.client;
+
+import com.pnoker.socket.handler.ClientHandler;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.string.StringDecoder;
+import io.netty.handler.codec.string.StringEncoder;
+import io.netty.util.AttributeKey;
+import io.netty.util.CharsetUtil;
+import lombok.extern.slf4j.Slf4j;
+
+import java.net.InetSocketAddress;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description:
+ */
+@Slf4j
+public class NettyClient {
+
+
+    private static Bootstrap b;
+    private static ChannelFuture f;
+    private static final EventLoopGroup workerGroup = new NioEventLoopGroup();
+
+    private static void init() {
+        try {
+            log.info("init...");
+            b = new Bootstrap();
+            b.group(workerGroup);
+            b.channel(NioSocketChannel.class);
+            b.option(ChannelOption.SO_KEEPALIVE, true);
+            b.handler(new ChannelInitializer<SocketChannel>() {
+                @Override
+                protected void initChannel(SocketChannel socketChannel) {
+                    // 解码编码
+                    socketChannel.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
+                    socketChannel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
+
+                    socketChannel.pipeline().addLast(new ClientHandler());
+                }
+            });
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static Object startAndWrite(InetSocketAddress address, Object send) throws InterruptedException {
+        init();
+        f = b.connect(address).sync();
+        // 传数据给服务端
+        f.channel().writeAndFlush(send);
+        f.channel().closeFuture().sync();
+        return f.channel().attr(AttributeKey.valueOf("Attribute_key")).get();
+    }
+
+    public static void main(String[] args) {
+        InetSocketAddress address = new InetSocketAddress("127.0.0.1", 7000);
+        String message = "hello";
+        try {
+            Object result = NettyClient.startAndWrite(address, message);
+            log.info("....result:" + result);
+        } catch (Exception e) {
+            log.error(e.getMessage());
+        } finally {
+            f.channel().close();
+            workerGroup.shutdownGracefully();
+            log.info("Closed client!");
+        }
+    }
+}
@@ -0,0 +1,19 @@
+package com.pnoker.socket.handler;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.util.AttributeKey;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description:
+ */
+public class ClientHandler extends SimpleChannelInboundHandler<Object> {
+    @Override
+    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) {
+        channelHandlerContext.channel().attr(AttributeKey.valueOf("Attribute_key")).set(msg);
+        channelHandlerContext.close();
+    }
+}
@@ -0,0 +1,25 @@
+package com.pnoker.socket.handler;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description:
+ */
+public class ServerHandler extends SimpleChannelInboundHandler<Object> {
+
+    @Override
+    public void channelRead0(ChannelHandlerContext ctx, Object msg) {
+        System.out.println("server receive message :" + msg);
+        ctx.channel().writeAndFlush("yes server already accept your message" + msg);
+        ctx.close();
+    }
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) {
+        System.out.println("channelActive>>>>>>>>");
+    }
+}
@@ -0,0 +1,68 @@
+package com.pnoker.socket.server;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelOption;
+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.stereotype.Component;
+
+import java.net.InetSocketAddress;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description:
+ */
+@Slf4j
+@Component
+public class NettyServer {
+
+    private final EventLoopGroup bossGroup = new NioEventLoopGroup();
+    private final EventLoopGroup workerGroup = new NioEventLoopGroup();
+
+    private Channel channel;
+
+    /**
+     * 启动服务
+     */
+    public ChannelFuture run(InetSocketAddress address) {
+
+        ChannelFuture f = null;
+        try {
+            ServerBootstrap b = new ServerBootstrap();
+            b.group(bossGroup, workerGroup)
+                    .channel(NioServerSocketChannel.class)
+                    .childHandler(new ServerChannelInitializer())
+                    .option(ChannelOption.SO_BACKLOG, 128)
+                    .childOption(ChannelOption.SO_KEEPALIVE, true);
+
+            f = b.bind(address).syncUninterruptibly();
+            channel = f.channel();
+        } catch (Exception e) {
+            log.error("Netty start error:", e);
+        } finally {
+            if (f != null && f.isSuccess()) {
+                log.info("Netty server listening " + address.getHostName() + " on port " + address.getPort() + " and ready for connections...");
+            } else {
+                log.error("Netty server start up Error!");
+            }
+        }
+
+        return f;
+    }
+
+    public void destroy() {
+        log.info("Shutdown Netty Server...");
+        if (channel != null) {
+            channel.close();
+        }
+        workerGroup.shutdownGracefully();
+        bossGroup.shutdownGracefully();
+        log.info("Shutdown Netty Server Success!");
+    }
+}
@@ -0,0 +1,25 @@
+package com.pnoker.socket.server;
+
+import com.pnoker.socket.handler.ServerHandler;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.string.StringDecoder;
+import io.netty.handler.codec.string.StringEncoder;
+import io.netty.util.CharsetUtil;
+
+/**
+ * <p>Copyright(c) 2018. Pnoker All Rights Reserved.
+ * <p>Author     : Pnoker
+ * <p>Email      : pnokers@gmail.com
+ * <p>Description:
+ */
+public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
+    @Override
+    protected void initChannel(SocketChannel socketChannel) {
+        // 解码编码
+        socketChannel.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
+        socketChannel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
+
+        socketChannel.pipeline().addLast(new ServerHandler());
+    }
+}