Android kotlin使用Netty网络框架实践TCP通讯（客户端、服务端）

杨帆起航_

2148人浏览 · 2024-09-02 17:31:28

杨帆起航_ · 2024-09-02 17:31:28 发布

开发工具：Android studio

语言:kotlin

设计原理：通讯协议：头+类型+长度+数据+尾，自定义编解码器，解析和包装发送数据流，以下贴出部分关键代码

说明：代码中封装了client和server端，可以点击按钮进行通讯，可以直接在项目中使用，尤其是处理了粘包和分包问题。

编译后的效果图:

注：结尾附上完整代码下载链接

1、配置build.gradle文件

 implementation("io.netty:netty-all:5.0.0.Alpha2")

2、主要代码

2.1 server端主要代码

    /**
     * 启动服务端
     */
    fun start() {
        Executors.newSingleThreadScheduledExecutor().submit {
            XLogUtil.d( "********服务启动********")
            bossGroup =NioEventLoopGroup()
            workerGroup = NioEventLoopGroup()
            try {
                val channelInit = ChannelInitServer(serverManager)
                val serverBootstrap = ServerBootstrap()
                serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel::class.java)//线程组设置为非阻塞
                    .childHandler(channelInit)
                    .option(ChannelOption.SO_BACKLOG, 128)//连接缓冲池的大小
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_KEEPALIVE, false)//设置长连接
                channelFuture = serverBootstrap.bind(Constant.SERVICE_POSR)
                channel = channelFuture?.channel()

                channelFuture!!.addListener { future: Future<in Void> ->
                    if (future.isSuccess) {
                        //服务启动成功
                        XLogUtil.d("********服务启动成功********")
                        MessageHandler.sendMessage(
                            MessageType.SERVER_START_SUCCESS,
                            "服务启动成功"
                        )
                    } else {
                        //服务启动失败
                        XLogUtil.e("********服务启动失败********")
                        MessageHandler.sendMessage(
                            MessageType.SERVER_START_FAILED,
                            "服务启动失败"
                        )
                    }
                }
                
            } catch (e: Exception) {
                e.printStackTrace()

                XLogUtil.e( "NettyServer 服务异常:"+e.message)
            } finally {

            }
        }
    }

2.2 client端主要代码

    /**
     * 启动客户端
     */
    fun start() {
        Executors.newSingleThreadScheduledExecutor().submit {
            XLogUtil.d("***********启动客户端***********")
            val group: EventLoopGroup = NioEventLoopGroup()
            try {
                val channelInit = ChannelInitClient(clientManager)
                val bootstrap = Bootstrap()
                bootstrap.group(group)
                    .channel(NioSocketChannel::class.java)
                    .remoteAddress(InetSocketAddress(address, port))
                    .handler(channelInit)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .option(ChannelOption.SO_KEEPALIVE, false)
                val channelFuture = bootstrap.connect().sync()
                channel = channelFuture.channel()
                channelFuture!!.addListener { future: Future<in Void> ->
                    if (future.isSuccess) {
                        //绑定成功
                        XLogUtil.d("***********客户端连接成功***********")
                        MessageHandler.sendMessage(
                            MessageType.CLIENT_CONNECT_SUCCESS,
                            "客户端连接成功"
                        )
                    } else {
                        //绑定失败
                        XLogUtil.d("***********客户端连接失败***********")
                        MessageHandler.sendMessage(
                            MessageType.CLIENT_CONNECT_FAILED,
                            "客户端连接失败"
                        )
                    }
                }

                channel!!.closeFuture().sync()
                XLogUtil.d("***********客户端关闭成功***********")
                MessageHandler.sendMessage(
                    MessageType.CLIENT_CLOSE_SUCCESS,
                    "客户端关闭成功"
                )
            } catch (e: Exception) {
                e.printStackTrace()
                MessageHandler.sendMessage(
                    MessageType.CLIENT_EXCEPTION,
                    "客户端异常：" + e.message
                )
                XLogUtil.e("NettyClient 客户端异常:" + e.message)
            } finally {
                try {
                    group.shutdownGracefully().sync()
                } catch (e: InterruptedException) {
                    e.printStackTrace()
                    MessageHandler.sendMessage(
                        MessageType.CLIENT_EXCEPTION,
                        "客户端异常2：" + e.message
                    )
                    XLogUtil.e("NettyClient 客户端异常2:" + e.message)
                }
            }
        }
    }

2.3 Server端线程

ChannelInitServer.kt

服务端数据收发线程
class ChannelInitServer internal constructor(adapter: MyServerHandler) :
    ChannelInitializer<SocketChannel?>() {
    private val adapter: MyServerHandler

    init {
        this.adapter = adapter
    }

    override fun initChannel(ch: SocketChannel?) {
        try {
            val channelPipeline: ChannelPipeline = ch!!.pipeline()
            //添加心跳机制，例：每3000ms发送一次心跳
            //channelPipeline.addLast(IdleStateHandler(3000, 3000, 3000, TimeUnit.MILLISECONDS))
            //添加数据处理（接收、发送、心跳）
            //FrameCodec 中处理粘包分包问题
            channelPipeline.addLast(FrameCodec())
            channelPipeline.addLast(adapter)
        } catch (e: Exception) {
            e.printStackTrace()
        }
    }
}

2.4 client 端线程

客户端数据收发线程
class ChannelInitClient internal constructor(adapter: MyClientHandler) :
    ChannelInitializer<Channel?>() {
    private val adapter: MyClientHandler

    init {
        this.adapter = adapter
    }


    override fun initChannel(ch: Channel?) {
        try {
            if (ch == null) {
                XLogUtil.e("ChannelInitClient Channel==null,initChannel fail")
            }
            val channelPipeline: ChannelPipeline = ch!!.pipeline()
            //添加心跳机制，例：每3000ms发送一次心跳
            // channelPipeline.addLast(IdleStateHandler(3000, 3000, 3000, TimeUnit.MILLISECONDS))
            //自定义编解码器，处理粘包分包问题
            channelPipeline.addLast(FrameCodec())
          //添加数据处理
            channelPipeline.addLast(adapter)
        } catch (e: Exception) {
            e.printStackTrace()
        }
    }
}

2.5 在Activity文件中调用

package com.android.agent

import android.annotation.SuppressLint
import android.content.Intent
import android.os.Build
import android.os.Bundle
import android.os.Environment
import android.os.Handler
import android.os.Message
import android.provider.Settings
import android.widget.Button
import android.widget.TextView
import androidx.appcompat.app.AppCompatActivity
import com.alibaba.fastjson.JSON
import com.android.agent.netty.NettyClient
import com.android.agent.netty.NettyServer
import com.android.agent.netty.message.MessageSend
import com.android.agent.netty.message.SettingIp
import com.android.agent.utils.Constant
import com.android.agent.xlog.XLogUtil
import com.android.agent.R


class MainActivity : AppCompatActivity() {

    private var isTestServer = false
    private var isTestClient = false
    private var client: NettyClient? = null
    private var server: NettyServer? = null
    private var result = ""
    private var tvResult: TextView? = null



    @SuppressLint("MissingInflatedId")
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.R) {
            if (!Environment.isExternalStorageManager()) {
                val intent = Intent(Settings.ACTION_MANAGE_ALL_FILES_ACCESS_PERMISSION);
                startActivity(intent);
                return;
            }
        }

        XLogUtil.d(">>>>>>>>>>welcome to  AndroidGif")

        tvResult = findViewById<TextView>(R.id.tv_text)

        findViewById<Button>(R.id.btnTestClient).setOnClickListener {
            XLogUtil.d(">>>>>>>>>>btnTestClient OnClick 启动"+!isTestClient)
            if (!isTestClient) {
                result = "";
                testNettyClient();
            } else {
                stopNettyClient();
            }
            isTestClient = !isTestClient;
        }

        findViewById<Button>(R.id.btnTestServer).setOnClickListener {
            XLogUtil.d(">>>>>>>>>>btnTestServer OnClicks 启动:"+!isTestServer)
            if (!isTestServer) {
                result = "";
                testNettyServer();

            } else {
                stopNettyServer();
            }
            isTestServer = !isTestServer;
        }

        findViewById<Button>(R.id.btnClientSend).setOnClickListener {
            client?.apply {
                XLogUtil.d("btnClientSend data")
                var setIp= SettingIp("192.168.11.185","192.168.11.1","255.255.255.0","8.8.8.8")
                var sendMsg= MessageSend("xxxxxxxxxxxx",3000,JSON.toJSONString(setIp))
                sentData(JSON.toJSONString(sendMsg),0x30) //charset("GBK")
            }
        }
    }


    private fun testNettyClient() {
         client = NettyClient(Constant.SERVICE_IP, Constant.SERVICE_POSR)
//        client.addHeartBeat(object : HeartBeatListener {
//            override fun getHeartBeat(): ByteArray {
//                val data = "心跳"
//                try {
//                    client.sentData("测试数据".toByteArray(charset("GBK")))
//                    return data.toByteArray(charset("GBK"))
//                } catch (e: UnsupportedEncodingException) {
//                    e.printStackTrace()
//                }
//                return "".toByteArray()
//            }
//        })
        client!!.setHandler(handler)
        client!!.start()

    }

    private fun stopNettyClient() {
        client?.apply {
            stop()
        }
    }

    private fun testNettyServer() {
        server = NettyServer.getInstance()

        server?.apply {
//            addHeartBeat(object : HeartBeatListener {
//                override fun getHeartBeat(): ByteArray {
//                    val data = "心跳"
//                    try {
//                        sentData("123".toByteArray(Charsets.UTF_8))//GBK
//                        return data.toByteArray(Charsets.UTF_8)
//                    } catch (e: UnsupportedEncodingException) {
//                        e.printStackTrace()
//                    }
//                    return "".toByteArray()
//                }
//            })
            setHandler(handler)
            start()
        }
    }

    private fun stopNettyServer() {
        server?.apply {
            stop()
        }
    }

    @SuppressLint("HandlerLeak")
    private val handler: Handler = object : Handler() {
        override fun handleMessage(msg: Message) {
            XLogUtil.d("收到信息:::" + msg.obj.toString())
            result += "\r\n"
            result += msg.obj
            tvResult!!.text = "收到信息:$result"
        }
    }

}

对应的布局文件：

<?xml version="1.0" encoding="utf-8"?>
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    android:orientation="vertical"
    tools:context="com.android.agent.MainActivity">

    <Button
        android:id="@+id/btnTestServer"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:text="测试服务端"
        android:layout_gravity="center_horizontal"
        android:layout_marginTop="50dp"
        />

    <Button
        android:id="@+id/btnTestClient"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_gravity="center_horizontal"
        android:layout_marginTop="100dp"
        android:text="测试客户端"
        />

    <Button
        android:id="@+id/btnClientSend"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_gravity="center_horizontal"
        android:layout_marginTop="100dp"
        android:text="客户端发送数据"
        />

    <TextView
        android:id="@+id/tv_text"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_gravity="center_horizontal"
        android:layout_marginTop="100dp"
        android:text="收到信息:"
        />


</LinearLayout>

2.6 数据编码解码器

需要根据协议去定义自己的编解码器，处理粘包丢包问题

完整代码下载地址：https://download.csdn.net/download/banzhuantuqiang/89705769

Netty自身有很多解码器，也可以结合Google的Protobuf(Google Protocol Buffers, 它是谷歌公司开源的一个序列化框架)使用，看项目需要决定是否需要集成。

开源鸿蒙跨平台开发者社区

开源鸿蒙跨平台开发社区汇聚开发者与厂商，共建“一次开发，多端部署”的开源生态，致力于降低跨端开发门槛，推动万物智联创新。

更多推荐

react native 0.72遇到react-native-screens库出现cxx编译问题的解决方案

《React Native轻量化编译C++错误排查全记录》摘要：本文记录了程序员千里马在维护React Native相簿应用时遭遇的棘手编译问题。项目在轻量化编译时频繁出现react-native-screens的C++编译错误，而完整编译却能通过。作者先后排查了路径限制、缓存污染、网络依赖、版本兼容等问题，尝试了换包、重建工程等多种方案。最终发现是node_modules中老旧的react-na

开源鸿蒙跨平台开发者社区

分布式软总线核心代码分析（二）

本文分析了OpenHarmony分布式软总线5.1版本中"发现服务"的核心实现机制。发现服务通过DiscMgrInit初始化三个通信介质（BLE/CoAP/USB），建立发布服务链表（g_publishInfoList）、发现服务链表（g_discoveryInfoList）和16个能力匹配桶（g_capabilityList）。当设备被发现时，系统会根据设备能力位图与订阅者进