单端口udp和tcp开启端口复用

gb28181/source.go

@@ -26,7 +26,7 @@ const (
 )
 
 var (
-	TransportManger stream.TransportManager
+	TransportManger transport.Manager
 	SharedUDPServer *UDPServer
 	SharedTCPServer *TCPServer
 )
@@ -312,40 +312,21 @@ func NewGBSource(id string, ssrc uint32, tcp bool, active bool) (GBSource, uint1
 		port = stream.AppConfig.GB28181.Port[0]
 	} else if !active {
 		if tcp {
-			err := TransportManger.AllocTransport(true, func(port_ uint16) error {
-
-				addr, _ := net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", stream.AppConfig.GB28181.Addr, port_))
-				server, err := NewTCPServer(addr, NewSingleFilter(source))
-				if err != nil {
-
-					return err
-				}
-
-				source.(*PassiveSource).transport = server.tcp
-				port = port_
-				return nil
-			})
-
+			tcpServer, err := NewTCPServer(NewSingleFilter(source))
 			if err != nil {
 				return nil, 0, err
 			}
+
+			port = uint16(tcpServer.tcp.ListenPort())
+			source.(*PassiveSource).transport = tcpServer.tcp
 		} else {
-			err := TransportManger.AllocTransport(false, func(port_ uint16) error {
-
-				addr, _ := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", stream.AppConfig.GB28181.Addr, port_))
-				server, err := NewUDPServer(addr, NewSingleFilter(source))
-				if err != nil {
-					return err
-				}
-
-				source.(*UDPSource).transport = server.udp
-				port = port_
-				return nil
-			})
-
+			server, err := NewUDPServer(NewSingleFilter(source))
 			if err != nil {
 				return nil, 0, err
 			}
+
+			port = uint16(server.udp.ListenPort())
+			source.(*UDPSource).transport = server.udp
 		}
 	}
 

gb28181/source_active.go

@@ -14,7 +14,7 @@ type ActiveSource struct {
 
 func NewActiveSource() (*ActiveSource, uint16, error) {
 	var port uint16
-	TransportManger.AllocTransport(true, func(port_ uint16) error {
+	TransportManger.AllocPort(true, func(port_ uint16) error {
 		port = port_
 		return nil
 	})

gb28181/tcp_server.go

@@ -1,9 +1,11 @@
 package gb28181
 
 import (
+	"fmt"
 	"github.com/yangjiechina/avformat/transport"
 	"github.com/yangjiechina/lkm/stream"
 	"net"
+	"runtime"
 )
 
 // TCPServer GB28181TCP被动收流
@@ -60,22 +62,46 @@ func (T *TCPServer) OnPacket(conn net.Conn, data []byte) []byte {
 	return nil
 }
 
-func NewTCPServer(addr net.Addr, filter Filter) (*TCPServer, error) {
+func NewTCPServer(filter Filter) (*TCPServer, error) {
 	server := &TCPServer{
 		filter: filter,
 	}
 
+	var tcp *transport.TCPServer
+	var err error
+	if stream.AppConfig.GB28181.IsMultiPort() {
+		tcp = &transport.TCPServer{}
+		tcp, err = TransportManger.NewTCPServer(stream.AppConfig.GB28181.Addr)
+		if err != nil {
+			return nil, err
+		}
+
+	} else {
+		tcp = &transport.TCPServer{
+			ReuseServer: transport.ReuseServer{
+				EnableReuse:      true,
+				ConcurrentNumber: runtime.NumCPU(),
+			},
+		}
+
+		var gbAddr *net.TCPAddr
+		addr := fmt.Sprintf("%s:%d", stream.AppConfig.GB28181.Addr, stream.AppConfig.GB28181.Port[0])
+		gbAddr, err = net.ResolveTCPAddr("tcp", addr)
+		if err != nil {
+			return nil, err
+		}
+
+		if err = tcp.Bind(gbAddr); err != nil {
+			return server, err
+		}
+	}
+
+	tcp.SetHandler(server)
+	tcp.Accept()
+	server.tcp = tcp
 	server.StreamServer = stream.StreamServer[*TCPSession]{
 		SourceType: stream.SourceType28181,
 		Handler:    server,
 	}
-
-	tcp := &transport.TCPServer{}
-	tcp.SetHandler(server)
-	if err := tcp.Bind(addr); err != nil {
-		return server, err
-	}
-
-	server.tcp = tcp
 	return server, nil
 }

gb28181/udp_server.go

@@ -1,11 +1,13 @@
 package gb28181
 
 import (
+	"fmt"
 	"github.com/pion/rtp"
 	"github.com/yangjiechina/avformat/transport"
 	"github.com/yangjiechina/lkm/log"
 	"github.com/yangjiechina/lkm/stream"
 	"net"
+	"runtime"
 )
 
 // UDPServer GB28181UDP收流
@@ -54,16 +56,40 @@ func (U *UDPServer) OnPacket(conn net.Conn, data []byte) []byte {
 	return nil
 }
 
-func NewUDPServer(addr net.Addr, filter Filter) (*UDPServer, error) {
+func NewUDPServer(filter Filter) (*UDPServer, error) {
 	server := &UDPServer{
 		filter: filter,
 	}
 
-	udp, err := transport.NewUDPServer(addr, server)
-	if err != nil {
-		return nil, err
+	var udp *transport.UDPServer
+	var err error
+	if stream.AppConfig.GB28181.IsMultiPort() {
+		udp, err = TransportManger.NewUDPServer(stream.AppConfig.GB28181.Addr)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		udp = &transport.UDPServer{
+			ReuseServer: transport.ReuseServer{
+				EnableReuse:      true,
+				ConcurrentNumber: runtime.NumCPU(),
+			},
+		}
+
+		var gbAddr *net.UDPAddr
+		addr := fmt.Sprintf("%s:%d", stream.AppConfig.GB28181.Addr, stream.AppConfig.GB28181.Port[0])
+		gbAddr, err = net.ResolveUDPAddr("udp", addr)
+		if err != nil {
+			return nil, err
+		}
+
+		if err = udp.Bind(gbAddr); err != nil {
+			return server, err
+		}
 	}
 
+	udp.SetHandler(server)
+	udp.Receive()
 	server.udp = udp
 	server.StreamServer = stream.StreamServer[*UDPSource]{
 		SourceType: stream.SourceType28181,

jt1078/jt_server.go

@@ -5,6 +5,7 @@ import (
 	"github.com/yangjiechina/avformat/utils"
 	"github.com/yangjiechina/lkm/stream"
 	"net"
+	"runtime"
 )
 
 type Server interface {
@@ -36,12 +37,18 @@ func (s *jtServer) OnPacket(conn net.Conn, data []byte) []byte {
 func (s *jtServer) Start(addr net.Addr) error {
 	utils.Assert(s.tcp == nil)
 
-	server := &transport.TCPServer{}
-	server.SetHandler(s)
+	server := &transport.TCPServer{
+		ReuseServer: transport.ReuseServer{
+			EnableReuse:      true,
+			ConcurrentNumber: runtime.NumCPU(),
+		},
+	}
 	if err := server.Bind(addr); err != nil {
 		return err
 	}
 
+	server.SetHandler(s)
+	server.Accept()
 	s.tcp = server
 	return nil
 }

main.go

@@ -1,7 +1,9 @@
 package main
 
 import (
+	"encoding/json"
 	"fmt"
+	"github.com/yangjiechina/avformat/transport"
 	"github.com/yangjiechina/lkm/flv"
 	"github.com/yangjiechina/lkm/gb28181"
 	"github.com/yangjiechina/lkm/hls"
@@ -107,12 +109,15 @@ func init() {
 	log.InitLogger(zapcore.Level(stream.AppConfig.Log.Level), stream.AppConfig.Log.Name, stream.AppConfig.Log.MaxSize, stream.AppConfig.Log.MaxBackup, stream.AppConfig.Log.MaxAge, stream.AppConfig.Log.Compress)
 
 	if stream.AppConfig.GB28181.IsMultiPort() {
-		gb28181.TransportManger = stream.NewTransportManager(stream.AppConfig.GB28181.Port[0], stream.AppConfig.GB28181.Port[1])
+		gb28181.TransportManger = transport.NewTransportManager(stream.AppConfig.GB28181.Port[0], stream.AppConfig.GB28181.Port[1])
 	}
 
 	if stream.AppConfig.Rtsp.IsMultiPort() {
-		rtsp.TransportManger = stream.NewTransportManager(stream.AppConfig.Rtsp.Port[0], stream.AppConfig.Rtsp.Port[1])
+		rtsp.TransportManger = transport.NewTransportManager(stream.AppConfig.Rtsp.Port[0], stream.AppConfig.Rtsp.Port[1])
 	}
+
+	indent, _ := json.MarshalIndent(stream.AppConfig, "", "\t")
+	log.Sugar.Infof("server config:%s", indent)
 }
 
 func main() {
@@ -156,35 +161,23 @@ func main() {
 	//多端口模式下, 创建GBSource时才创建收流端口
 	if !stream.AppConfig.GB28181.IsMultiPort() {
 		if stream.AppConfig.GB28181.EnableUDP() {
-			addr := fmt.Sprintf("%s:%d", stream.AppConfig.GB28181.Addr, stream.AppConfig.GB28181.Port[0])
-			gbAddr, err := net.ResolveUDPAddr("udp", addr)
-			if err != nil {
-				panic(err)
-			}
-
-			server, err := gb28181.NewUDPServer(gbAddr, gb28181.NewSharedFilter(128))
+			server, err := gb28181.NewUDPServer(gb28181.NewSharedFilter(128))
 			if err != nil {
 				panic(err)
 			}
 
 			gb28181.SharedUDPServer = server
-			log.Sugar.Info("启动GB28181 UDP收流端口成功:" + gbAddr.String())
+			log.Sugar.Info("启动GB28181 UDP收流端口成功:" + fmt.Sprintf("%s:%d", stream.AppConfig.GB28181.Addr, stream.AppConfig.GB28181.Port[0]))
 		}
 
 		if stream.AppConfig.GB28181.EnableTCP() {
-			addr := fmt.Sprintf("%s:%d", stream.AppConfig.GB28181.Addr, stream.AppConfig.GB28181.Port[0])
-			gbAddr, err := net.ResolveTCPAddr("tcp", addr)
-			if err != nil {
-				panic(err)
-			}
-
-			server, err := gb28181.NewTCPServer(gbAddr, gb28181.NewSharedFilter(128))
+			server, err := gb28181.NewTCPServer(gb28181.NewSharedFilter(128))
 			if err != nil {
 				panic(err)
 			}
 
 			gb28181.SharedTCPServer = server
-			log.Sugar.Info("启动GB28181 TCP收流端口成功:" + gbAddr.String())
+			log.Sugar.Info("启动GB28181 TCP收流端口成功:" + fmt.Sprintf("%s:%d", stream.AppConfig.GB28181.Addr, stream.AppConfig.GB28181.Port[0]))
 		}
 	}
 

rtmp/rtmp_server.go

@@ -4,6 +4,7 @@ import (
 	"github.com/yangjiechina/lkm/log"
 	"github.com/yangjiechina/lkm/stream"
 	"net"
+	"runtime"
 
 	"github.com/yangjiechina/avformat/transport"
 	"github.com/yangjiechina/avformat/utils"
@@ -24,12 +25,19 @@ type server struct {
 func (s *server) Start(addr net.Addr) error {
 	utils.Assert(s.tcp == nil)
 
-	tcp := &transport.TCPServer{}
-	tcp.SetHandler(s)
+	tcp := &transport.TCPServer{
+		ReuseServer: transport.ReuseServer{
+			EnableReuse:      true,
+			ConcurrentNumber: runtime.NumCPU(),
+		},
+	}
+
 	if err := tcp.Bind(addr); err != nil {
 		return err
 	}
 
+	tcp.SetHandler(s)
+	tcp.Accept()
 	s.tcp = tcp
 	return nil
 }

rtsp/rtsp_server.go

@@ -5,6 +5,7 @@ import (
 	"github.com/yangjiechina/avformat/utils"
 	"github.com/yangjiechina/lkm/log"
 	"net"
+	"runtime"
 )
 
 type Server interface {
@@ -28,13 +29,19 @@ func (s *server) Start(addr net.Addr) error {
 	utils.Assert(s.tcp == nil)
 
 	//监听TCP端口
-	tcp := &transport.TCPServer{}
-	tcp.SetHandler(s)
-	err := tcp.Bind(addr)
-	if err != nil {
+	tcp := &transport.TCPServer{
+		ReuseServer: transport.ReuseServer{
+			EnableReuse:      true,
+			ConcurrentNumber: runtime.NumCPU(),
+		},
+	}
+
+	if err := tcp.Bind(addr); err != nil {
 		return err
 	}
 
+	tcp.SetHandler(s)
+	tcp.Accept()
 	s.tcp = tcp
 	return nil
 }

rtsp/rtsp_sink.go

@@ -1,7 +1,6 @@
 package rtsp
 
 import (
-	"fmt"
 	"github.com/pion/rtcp"
 	"github.com/yangjiechina/avformat/librtp"
 	"github.com/yangjiechina/avformat/transport"
@@ -13,7 +12,7 @@ import (
 )
 
 var (
-	TransportManger stream.TransportManager
+	TransportManger transport.Manager
 )
 
 // rtsp拉流sink
@@ -59,42 +58,25 @@ func (s *sink) addSender(index int, tcp bool, ssrc uint32) (uint16, uint16, erro
 	if tcp {
 		s.tcp = true
 	} else {
-		err = TransportManger.AllocPairTransport(func(port uint16) error {
-			//rtp port
-			var addr *net.UDPAddr
-			addr, err = net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", "0.0.0.0", port))
+		sender.Rtp, err = TransportManger.NewUDPServer("0.0.0.0")
+		if err != nil {
+			return 0, 0, err
+		}
 
-			if err == nil {
-				//创建rtp udp server
-				sender.Rtp = &transport.UDPServer{}
-				sender.Rtp.SetHandler2(nil, sender.OnRTPPacket, nil)
-				err = sender.Rtp.Bind(addr)
-			}
+		sender.Rtcp, err = TransportManger.NewUDPServer("0.0.0.0")
+		if err != nil {
+			sender.Rtp.Close()
+			sender.Rtp = nil
+			return 0, 0, err
+		}
 
-			rtpPort = port
-			return nil
-		}, func(port uint16) error {
-			//rtcp port
-			var addr *net.UDPAddr
-			addr, err = net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%d", "0.0.0.0", port))
+		sender.Rtp.SetHandler2(nil, sender.OnRTPPacket, nil)
+		sender.Rtcp.SetHandler2(nil, sender.OnRTCPPacket, nil)
+		sender.Rtp.(*transport.UDPServer).Receive()
+		sender.Rtcp.(*transport.UDPServer).Receive()
 
-			if err == nil {
-				//创建rtcp udp server
-				sender.Rtcp = &transport.UDPServer{}
-				sender.Rtcp.SetHandler2(nil, sender.OnRTCPPacket, nil)
-				err = sender.Rtcp.Bind(addr)
-			} else {
-				sender.Rtp.Close()
-				sender.Rtp = nil
-			}
-
-			rtcpPort = port
-			return nil
-		})
-	}
-
-	if err != nil {
-		return 0, 0, err
+		rtpPort = uint16(sender.Rtp.ListenPort())
+		rtcpPort = uint16(sender.Rtcp.ListenPort())
 	}
 
 	s.senders[index] = &sender

stream/transport_manager.go

@@ -1,72 +0,0 @@
-package stream
-
-import (
-	"fmt"
-	"github.com/yangjiechina/avformat/libbufio"
-	"github.com/yangjiechina/avformat/utils"
-	"sync"
-)
-
-type TransportManager interface {
-	AllocTransport(tcp bool, cb func(port uint16) error) error
-
-	AllocPairTransport(cb, c2 func(port uint16) error) error
-}
-
-func NewTransportManager(start, end uint16) TransportManager {
-	utils.Assert(end > start)
-
-	return &transportManager{
-		startPort: start,
-		endPort:   end,
-		nextPort:  start,
-	}
-}
-
-type transportManager struct {
-	startPort uint16
-	endPort   uint16
-	nextPort  uint16
-	lock      sync.Mutex
-}
-
-func (t *transportManager) AllocTransport(tcp bool, cb func(port uint16) error) error {
-	loop := func(start, end uint16, tcp bool) (uint16, error) {
-		for i := start; i < end; i++ {
-			if used := utils.Used(int(i), tcp); !used {
-				return i, cb(i)
-			}
-		}
-
-		return 0, nil
-	}
-
-	t.lock.Lock()
-	defer t.lock.Unlock()
-
-	port, err := loop(t.nextPort, t.endPort, tcp)
-	if port == 0 {
-		port, err = loop(t.startPort, t.nextPort, tcp)
-	}
-
-	if port == 0 {
-		return fmt.Errorf("no available ports in the [%d-%d] range", t.startPort, t.endPort)
-	} else if err != nil {
-		return err
-	}
-
-	t.nextPort = t.nextPort + 1%t.endPort
-	t.nextPort = uint16(libbufio.MaxInt(int(t.nextPort), int(t.startPort)))
-	return nil
-}
-
-func (t *transportManager) AllocPairTransport(cb func(port uint16) error, cb2 func(port uint16) error) error {
-	if err := t.AllocTransport(false, cb); err != nil {
-		return err
-	}
-
-	if err := t.AllocTransport(false, cb2); err != nil {
-		return err
-	}
-	return nil
-}