Kotlin + Netty 在 Android 上實現 Socket 的服務端

一. 背景

最近的一個項目：須要使用 Android App 做爲 Socket 的服務端，而且一個端口可以同時監聽 TCP/Web Socket 協議。

天然而然，項目決定採用 Netty 框架。Netty 服務端在收到客戶端發來的消息後，可以作出相應的業務處理。在某些場景下，服務端也須要給客戶端 App/網頁發送消息。

二. Netty 的使用

2.1 Netty 服務端

首先，定義好 NettyServer，它使用object聲明表示是一個單例。用於 Netty 服務端的啓動、關閉以及發送消息。

object NettyServer {

    private val TAG = "NettyServer"

    private var channel: Channel?=null
    private lateinit var listener: NettyServerListener<String>
    private lateinit var bossGroup: EventLoopGroup
    private lateinit var workerGroup: EventLoopGroup

    var port = 8888
        set(value)  {
            field = value
        }

    var webSocketPath = "/ws"
        set(value)  {
            field = value
        }

    var isServerStart: Boolean = false
        private set

    fun start() {
        object : Thread() {
            override fun run() {
                super.run()
                bossGroup = NioEventLoopGroup(1)
                workerGroup = NioEventLoopGroup()
                try {
                    val b = ServerBootstrap()
                    b.group(bossGroup, workerGroup)
                            .channel(NioServerSocketChannel::class.java)
                            .localAddress(InetSocketAddress(port))
                            .childOption(ChannelOption.SO_KEEPALIVE, true)
                            .childOption(ChannelOption.SO_REUSEADDR, true)
                            .childOption(ChannelOption.TCP_NODELAY, true)
                            .childHandler(NettyServerInitializer(listener,webSocketPath))

                    // Bind and start to accept incoming connections.
                    val f = b.bind().sync()
                    Log.i(TAG, NettyServer::class.java.name + " started and listen on " + f.channel().localAddress())

                    isServerStart = true
                    listener.onStartServer()
                    f.channel().closeFuture().sync()
                } catch (e: Exception) {
                    Log.e(TAG, e.localizedMessage)
                    e.printStackTrace()
                } finally {
                    isServerStart = false
                    listener.onStopServer()

                    disconnect()
                }
            }
        }.start()

    }

    fun disconnect() {
        workerGroup.shutdownGracefully()
        bossGroup.shutdownGracefully()
    }

    fun setListener(listener: NettyServerListener<String>) {
        this.listener = listener
    }

    // 異步發送TCP消息
    fun sendMsgToClient(data: String, listener: ChannelFutureListener) = channel?.run {

        val flag = this.isActive

        if (flag) {

            this.writeAndFlush(data + System.getProperty("line.separator")).addListener(listener)
        }

        flag
    } ?: false

    // 同步發送TCP消息
    fun sendMsgToClient(data: String) = channel?.run {

        if (this.isActive) {

            return this.writeAndFlush(data + System.getProperty("line.separator")).awaitUninterruptibly().isSuccess
        }

        false
    } ?: false

    // 異步發送WebSocket消息
    fun sendMsgToWS(data: String,listener: ChannelFutureListener) = channel?.run {

        val flag = this.isActive

        if (flag) {

            this.writeAndFlush(TextWebSocketFrame(data)).addListener(listener)
        }

        flag
    } ?: false

    // 同步發送TCP消息
    fun sendMsgToWS(data: String) = channel?.run {

        if (this.isActive) {

            return this.writeAndFlush(TextWebSocketFrame(data)).awaitUninterruptibly().isSuccess
        }

        false
    } ?: false

    /** * 切換通道 * 設置服務端，與哪一個客戶端通訊 * @param channel */
    fun selectorChannel(channel: Channel?) {
        this.channel = channel
    }
}
複製代碼

NettyServerInitializer 是服務端跟客戶端鏈接以後使用的 childHandler：

class NettyServerInitializer(private val mListener: NettyServerListener<String>,private val webSocketPath:String) : ChannelInitializer<SocketChannel>() {

    @Throws(Exception::class)
    public override fun initChannel(ch: SocketChannel) {

        val pipeline = ch.pipeline()

        pipeline.addLast("active",ChannelActiveHandler(mListener))
        pipeline.addLast("socketChoose", SocketChooseHandler(webSocketPath))

        pipeline.addLast("string_encoder",StringEncoder(CharsetUtil.UTF_8))
        pipeline.addLast("linebased",LineBasedFrameDecoder(1024))
        pipeline.addLast("string_decoder",StringDecoder(CharsetUtil.UTF_8))
        pipeline.addLast("commonhandler", CustomerServerHandler(mListener))
    }
}
複製代碼

NettyServerInitializer 包含了多個 Handler：鏈接使用的ChannelActiveHandler，協議選擇使用的 SocketChooseHandler，TCP 消息使用的 StringEncoder、LineBasedFrameDecoder、StringDecoder，以及最終處理消息的 CustomerServerHandler。

ChannelActiveHandler:

@ChannelHandler.Sharable
class ChannelActiveHandler(var mListener: NettyServerListener<String>) : ChannelInboundHandlerAdapter() {

    @Throws(Exception::class)
    override fun channelActive(ctx: ChannelHandlerContext) {

        val insocket = ctx.channel().remoteAddress() as InetSocketAddress
        val clientIP = insocket.address.hostAddress
        val clientPort = insocket.port

        Log.i("ChannelActiveHandler","新的鏈接：$clientIP : $clientPort")
        mListener.onChannelConnect(ctx.channel())
    }

}
複製代碼

SocketChooseHandler 經過讀取消息來區分是 WebSocket 仍是 Socket。若是是 WebSocket 的話，去掉 Socket 使用的相關 Handler。

class SocketChooseHandler(val webSocketPath:String) : ByteToMessageDecoder() {

    @Throws(Exception::class)
    override fun decode(ctx: ChannelHandlerContext, `in`: ByteBuf, out: List<Any>) {
        val protocol = getBufStart(`in`)
        if (protocol.startsWith(WEBSOCKET_PREFIX)) {
            PipelineAdd.websocketAdd(ctx,webSocketPath)

            ctx.pipeline().remove("string_encoder")
            ctx.pipeline().remove("linebased")
            ctx.pipeline().remove("string_decoder")
        }
        `in`.resetReaderIndex()
        ctx.pipeline().remove(this.javaClass)
    }

    private fun getBufStart(`in`: ByteBuf): String {
        var length = `in`.readableBytes()
        if (length > MAX_LENGTH) {
            length = MAX_LENGTH
        }

        // 標記讀位置
        `in`.markReaderIndex()
        val content = ByteArray(length)
        `in`.readBytes(content)
        return String(content)
    }

    companion object {
        /** 默認暗號長度爲23 */
        private val MAX_LENGTH = 23
        /** WebSocket握手的協議前綴 */
        private val WEBSOCKET_PREFIX = "GET /"
    }
}
複製代碼

StringEncoder、LineBasedFrameDecoder、StringDecoder 都是 Netty 內置的編、解碼器。其中，LineBasedFrameDecoder 用於解決 TCP粘包/拆包的問題。

CustomerServerHandler:

@ChannelHandler.Sharable
class CustomerServerHandler(private val mListener: NettyServerListener<String>) : SimpleChannelInboundHandler<Any>() {

    @Throws(Exception::class)
    override fun channelReadComplete(ctx: ChannelHandlerContext) {
    }

    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
        cause.printStackTrace()
        ctx.close()
    }

    @Throws(Exception::class)
    override fun channelRead0(ctx: ChannelHandlerContext, msg: Any) {

        val buff = msg as ByteBuf
        val info = buff.toString(CharsetUtil.UTF_8)
        Log.d(TAG,"收到消息內容：$info")
    }

    @Throws(Exception::class)
    override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {

        if (msg is WebSocketFrame) {  // 處理 WebSocket 消息

            val webSocketInfo = (msg as TextWebSocketFrame).text().trim { it <= ' ' }

            Log.d(TAG, "收到WebSocketSocket消息：$webSocketInfo")

            mListener.onMessageResponseServer(webSocketInfo , ctx.channel().id().asShortText())
        } else if (msg is String){   // 處理 Socket 消息

            Log.d(TAG, "收到socket消息：$msg")

            mListener.onMessageResponseServer(msg, ctx.channel().id().asShortText())
        }
    }

    // 斷開鏈接
    @Throws(Exception::class)
    override fun channelInactive(ctx: ChannelHandlerContext) {
        super.channelInactive(ctx)
        Log.d(TAG, "channelInactive")

        val reAddr = ctx.channel().remoteAddress() as InetSocketAddress
        val clientIP = reAddr.address.hostAddress
        val clientPort = reAddr.port

        Log.d(TAG,"鏈接斷開：$clientIP : $clientPort")

        mListener.onChannelDisConnect(ctx.channel())
    }

    companion object {

        private val TAG = "CustomerServerHandler"
    }
}
複製代碼

2.2 Netty 客戶端

客戶端也須要一個啓動、關閉、發送消息的 NettyTcpClient，而且 NettyTcpClient 的建立採用 Builder 模式。

class NettyTcpClient private constructor(val host: String, val tcp_port: Int, val index: Int) {

    private lateinit var group: EventLoopGroup

    private lateinit var listener: NettyClientListener<String>

    private var channel: Channel? = null

    /** * 獲取TCP鏈接狀態 * * @return 獲取TCP鏈接狀態 */
    var connectStatus = false

    /** * 最大重連次數 */
    var maxConnectTimes = Integer.MAX_VALUE
        private set

    private var reconnectNum = maxConnectTimes

    private var isNeedReconnect = true

    var isConnecting = false
        private set

    var reconnectIntervalTime: Long = 5000
        private set

    /** * 心跳間隔時間 */
    var heartBeatInterval: Long = 5
        private set//單位秒

    /** * 是否發送心跳 */
    var isSendheartBeat = false
        private set

    /** * 心跳數據，能夠是String類型，也能夠是byte[]. */
    private var heartBeatData: Any? = null

    fun connect() {
        if (isConnecting) {
            return
        }

        val clientThread = object : Thread("Netty-Client") {
            override fun run() {
                super.run()
                isNeedReconnect = true
                reconnectNum = maxConnectTimes
                connectServer()
            }
        }
        clientThread.start()
    }


    private fun connectServer() {

        synchronized(this@NettyTcpClient) {

            var channelFuture: ChannelFuture?=null

            if (!connectStatus) {
                isConnecting = true
                group = NioEventLoopGroup()
                val bootstrap = Bootstrap().group(group)
                        .option(ChannelOption.TCP_NODELAY, true)//屏蔽Nagle算法試圖
                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                        .channel(NioSocketChannel::class.java as Class<out Channel>?)
                        .handler(object : ChannelInitializer<SocketChannel>() {

                            @Throws(Exception::class)
                            public override fun initChannel(ch: SocketChannel) {

                                if (isSendheartBeat) {
                                    ch.pipeline().addLast("ping", IdleStateHandler(0, heartBeatInterval, 0, TimeUnit.SECONDS)) //5s未發送數據，回調userEventTriggered
                                }

                                ch.pipeline().addLast(StringEncoder(CharsetUtil.UTF_8))
                                ch.pipeline().addLast(StringDecoder(CharsetUtil.UTF_8))
                                ch.pipeline().addLast(LineBasedFrameDecoder(1024))//黏包處理,須要客戶端、服務端配合
                                ch.pipeline().addLast(NettyClientHandler(listener, index, isSendheartBeat, heartBeatData))
                            }
                        })

                try {
                    channelFuture = bootstrap.connect(host, tcp_port).addListener {
                        if (it.isSuccess) {
                            Log.d(TAG, "鏈接成功")
                            reconnectNum = maxConnectTimes
                            connectStatus = true
                            channel = channelFuture?.channel()
                        } else {
                            Log.d(TAG, "鏈接失敗")
                            connectStatus = false
                        }
                        isConnecting = false
                    }.sync()

                    // Wait until the connection is closed.
                    channelFuture.channel().closeFuture().sync()
                    Log.d(TAG, " 斷開鏈接")
                } catch (e: Exception) {
                    e.printStackTrace()
                } finally {
                    connectStatus = false
                    listener.onClientStatusConnectChanged(ConnectState.STATUS_CONNECT_CLOSED, index)

                    if (channelFuture != null) {
                        if (channelFuture.channel() != null && channelFuture.channel().isOpen) {
                            channelFuture.channel().close()
                        }
                    }
                    group.shutdownGracefully()
                    reconnect()
                }
            }
        }
    }


    fun disconnect() {
        Log.d(TAG, "disconnect")
        isNeedReconnect = false
        group.shutdownGracefully()
    }

    fun reconnect() {
        Log.d(TAG, "reconnect")
        if (isNeedReconnect && reconnectNum > 0 && !connectStatus) {
            reconnectNum--
            SystemClock.sleep(reconnectIntervalTime)
            if (isNeedReconnect && reconnectNum > 0 && !connectStatus) {
                Log.e(TAG, "從新鏈接")
                connectServer()
            }
        }
    }

    /** * 異步發送 * * @param data 要發送的數據 * @param listener 發送結果回調 * @return 方法執行結果 */
    fun sendMsgToServer(data: String, listener: MessageStateListener) = channel?.run {

        val flag = this != null && connectStatus

        if (flag) {

            this.writeAndFlush(data + System.getProperty("line.separator")).addListener { channelFuture -> listener.isSendSuccss(channelFuture.isSuccess) }
        }

        flag

    } ?: false

    /** * 同步發送 * * @param data 要發送的數據 * @return 方法執行結果 */
    fun sendMsgToServer(data: String) = channel?.run {

        val flag = this != null && connectStatus

        if (flag) {

            val channelFuture = this.writeAndFlush(data + System.getProperty("line.separator")).awaitUninterruptibly()
            return channelFuture.isSuccess
        }

        false

    }?:false

    fun setListener(listener: NettyClientListener<String>) {
        this.listener = listener
    }

    /** * Builder 模式建立NettyTcpClient */
    class Builder {

        /** * 最大重連次數 */
        private var MAX_CONNECT_TIMES = Integer.MAX_VALUE

        /** * 重連間隔 */
        private var reconnectIntervalTime: Long = 5000
        /** * 服務器地址 */
        private var host: String? = null
        /** * 服務器端口 */
        private var tcp_port: Int = 0
        /** * 客戶端標識，(由於可能存在多個鏈接) */
        private var mIndex: Int = 0

        /** * 是否發送心跳 */
        private var isSendheartBeat: Boolean = false
        /** * 心跳時間間隔 */
        private var heartBeatInterval: Long = 5

        /** * 心跳數據，能夠是String類型，也能夠是byte[]. */
        private var heartBeatData: Any? = null


        fun setMaxReconnectTimes(reConnectTimes: Int): Builder {
            this.MAX_CONNECT_TIMES = reConnectTimes
            return this
        }


        fun setReconnectIntervalTime(reconnectIntervalTime: Long): Builder {
            this.reconnectIntervalTime = reconnectIntervalTime
            return this
        }


        fun setHost(host: String): Builder {
            this.host = host
            return this
        }

        fun setTcpPort(tcp_port: Int): Builder {
            this.tcp_port = tcp_port
            return this
        }

        fun setIndex(mIndex: Int): Builder {
            this.mIndex = mIndex
            return this
        }

        fun setHeartBeatInterval(intervalTime: Long): Builder {
            this.heartBeatInterval = intervalTime
            return this
        }

        fun setSendheartBeat(isSendheartBeat: Boolean): Builder {
            this.isSendheartBeat = isSendheartBeat
            return this
        }

        fun setHeartBeatData(heartBeatData: Any): Builder {
            this.heartBeatData = heartBeatData
            return this
        }

        fun build(): NettyTcpClient {
            val nettyTcpClient = NettyTcpClient(host!!, tcp_port, mIndex)
            nettyTcpClient.maxConnectTimes = this.MAX_CONNECT_TIMES
            nettyTcpClient.reconnectIntervalTime = this.reconnectIntervalTime
            nettyTcpClient.heartBeatInterval = this.heartBeatInterval
            nettyTcpClient.isSendheartBeat = this.isSendheartBeat
            nettyTcpClient.heartBeatData = this.heartBeatData
            return nettyTcpClient
        }
    }

    companion object {
        private val TAG = "NettyTcpClient"
        private val CONNECT_TIMEOUT_MILLIS = 5000
    }
}
複製代碼

Android 的客戶端相對而言比較簡單，須要的 Handler 包括：支持心跳的 IdleStateHandler， TCP 消息須要使用的 Handler (跟服務端同樣分別是StringEncoder、StringDecoder、LineBasedFrameDecoder)，以及對收到 TCP 消息進行處理的 NettyClientHandler。

NettyClientHandler：

class NettyClientHandler(private val listener: NettyClientListener<String>, private val index: Int, private val isSendheartBeat: Boolean, private val heartBeatData: Any?) : SimpleChannelInboundHandler<String>() {

    /** * * 設定IdleStateHandler心跳檢測每x秒進行一次讀檢測， * 若是x秒內ChannelRead()方法未被調用則觸發一次userEventTrigger()方法 * * @param ctx ChannelHandlerContext * @param evt IdleStateEvent */
    override fun userEventTriggered(ctx: ChannelHandlerContext, evt: Any) {

        if (evt is IdleStateEvent) {
            if (evt.state() == IdleState.WRITER_IDLE) {   //發送心跳

                if (isSendheartBeat) {
                    if (heartBeatData == null) {

                        ctx.channel().writeAndFlush("Heartbeat" + System.getProperty("line.separator")!!)
                    } else {

                        if (heartBeatData is String) {
                            Log.d(TAG, "userEventTriggered: String")
                            ctx.channel().writeAndFlush(heartBeatData + System.getProperty("line.separator")!!)
                        } else if (heartBeatData is ByteArray) {
                            Log.d(TAG, "userEventTriggered: byte")
                            val buf = Unpooled.copiedBuffer((heartBeatData as ByteArray?)!!)
                            ctx.channel().writeAndFlush(buf)
                        } else {

                            Log.d(TAG, "userEventTriggered: heartBeatData type error")
                        }
                    }
                } else {
                    Log.d(TAG, "不發送心跳")
                }
            }
        }
    }

    /** * * 客戶端上線 * * @param ctx ChannelHandlerContext */
    override fun channelActive(ctx: ChannelHandlerContext) {

        Log.d(TAG, "channelActive")
        listener.onClientStatusConnectChanged(ConnectState.STATUS_CONNECT_SUCCESS, index)
    }

    /** * * 客戶端下線 * * @param ctx ChannelHandlerContext */
    override fun channelInactive(ctx: ChannelHandlerContext) {

        Log.d(TAG, "channelInactive")
    }

    /** * 客戶端收到消息 * * @param channelHandlerContext ChannelHandlerContext * @param msg 消息 */
    override fun channelRead0(channelHandlerContext: ChannelHandlerContext, msg: String) {

        Log.d(TAG, "channelRead0:")
        listener.onMessageResponseClient(msg, index)
    }

    /** * @param ctx ChannelHandlerContext * @param cause 異常 */
    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {

        Log.e(TAG, "exceptionCaught")
        listener.onClientStatusConnectChanged(ConnectState.STATUS_CONNECT_ERROR, index)
        cause.printStackTrace()
        ctx.close()
    }

    companion object {

        private val TAG = "NettyClientHandler"
    }
}
複製代碼

三. Demo 的實現

3.1 Socket 服務端

啓動 NettyServer：

private fun startServer() {

        if (!NettyServer.isServerStart) {
            NettyServer.setListener(this@MainActivity)
            NettyServer.port = port
            NettyServer.webSocketPath = webSocketPath
            NettyServer.start()
        } else {
            NettyServer.disconnect()
        }
    }
複製代碼

NettyServer 異步發送 TCP 消息：

NettyServer.sendMsgToClient(msg, ChannelFutureListener { channelFuture ->

        if (channelFuture.isSuccess) {

            msgSend(msg)
      } 
})
複製代碼

NettyServer 異步發送 WebSocket 消息：

NettyServer.sendMsgToWS(msg, ChannelFutureListener { channelFuture ->

        if (channelFuture.isSuccess) {

              msgSend(msg)
      } 
 })
複製代碼

Demo 能夠經過 startServer 來啓動 Socket 服務端，也能夠在啓動以前點擊 configServer 來修改服務端的端口以及 WebSocket 的 Endpoint。

3.2 Socket 客戶端

NettyTcpClient 經過 Builder 模式建立：

mNettyTcpClient = NettyTcpClient.Builder()
                    .setHost(ip)                    //設置服務端地址
                    .setTcpPort(port)               //設置服務端端口號
                    .setMaxReconnectTimes(5)        //設置最大重連次數
                    .setReconnectIntervalTime(5)    //設置重連間隔時間。單位：秒
                    .setSendheartBeat(false)        //設置發送心跳
                    .setHeartBeatInterval(5)        //設置心跳間隔時間。單位：秒
                    .setHeartBeatData("I'm is HeartBeatData") //設置心跳數據，能夠是String類型，也能夠是byte[]，之後設置的爲準
                    .setIndex(0)                    //設置客戶端標識.(由於可能存在多個tcp鏈接)
                    .build()

            mNettyTcpClient.setListener(this@MainActivity) //設置TCP監聽
複製代碼

啓動、關閉客戶端鏈接：

private fun connect() {
        Log.d(TAG, "connect")
        if (!mNettyTcpClient.connectStatus) {
            mNettyTcpClient.connect()//鏈接服務器
        } else {
            mNettyTcpClient.disconnect()
        }
    }
複製代碼

NettyTcpClient 異步發送 TCP 消息到服務端：

mNettyTcpClient.sendMsgToServer(msg, object : MessageStateListener {
         override fun isSendSuccss(isSuccess: Boolean) {
                if (isSuccess) {
                           
                   msgSend(msg)
               } 
         }
})
複製代碼

Demo 的客戶端 App 也能夠在啓動以前點擊 configClient 來修改要鏈接的服務端 IP 、端口。

WebSocket 的測試能夠經過：www.websocket-test.com/

Netty Server 端跟網頁通訊：

WebSocket在線測試：

四. 總結

藉助 Kotlin 的特性以及 Netty 框架，咱們在 Android 上也實現了一個 Socket 服務端。

本文 demo github 地址：github.com/fengzhizi71…

本文的例子很簡單，只是發送簡單的消息。在實際生產環境中，咱們採用的消息格式多是 json ，由於 json 更加靈活，經過解析 json 獲取消息的內容。

參考資料：

1. github.com/aLittleGree…
2. Netty 實現 一個端口同時接收 socket 和 webSocket 鏈接