开源项目：用环信MQTT实现"世界频道"只需5分钟【附源码】

　　说到“世界频道”想必大家都不陌生，常见的如王者荣耀的世界广播摇人组队以及最近兴起的Discord社区交友等等。究其目的就是在应用内让海量用户可以实时互动。有些开发者为了实现这种场景会选择聊天室方案来实现，但是这种方式存在一定的局限性，比如聊天室人数上限、海量消息处理等各种情况。

　　当然如果有钱有颜，可以直接选择云厂商产品(比如环信的聊天室方案和超级社区)，如果有才有time，也可以选择平替版MQTT实现方案。今天小猿将介绍用环信MQTT消息云实现应用内的世界频道，满满干货，不要错过～～

　　使用MQTT实现世界频道-Demo效果演示

　　协议优势：

　　在介绍具体方案之前，我们先唠一唠为啥选择MQTT协议。

　　轻量级：MQTT本身是物联网的连接协议，专为受限设备和低带宽场景使用。所以其代码占用空间较小，同样适用于注重SDK大小的移动应用领域(比如：游戏领域)。

　　易集成：MQTT作为标准开放的消息协议，经过多年演进，已支持30多种开发语言，10余种SDK，无论何种开发环境，都可以快速找到开源SDK。

　　高并发：MQTT是轻量级的消息传输协议，2字节心跳报文，最小化传输和连接成本，云厂商broker产品都可支持千万级并发接入，适用于高并发连接场景。

　　低成本：MQTT是基于客户端-服务器的订阅/发布模型，通过服务器中间件实现消息分发，减少消息复制成本，快速实现一对多在线推送。

　　灵活性：MQTT协议支持多种消息特性，包括：topic主题层级、消息分级(QoS0，1，2)、遗嘱消息、保留消息等，可以灵活实现多种业务场景。

　　衍生功能：随着MQTT云服务的发展，部分服务器厂商已支持消息存储、获取在线设备列表、查看历史消息等衍生功能，降低开发工作量与消息存储成本。

　　实现方案：

　　言归正传，上干货。本次技术实现方案包含：移动客户端(Android)、后端服务(Java)以及MQTT服务器。这里提一下，MQTT服务器使用环信MQTT消息云，使用三方云服务比较省心，既节省开发时间，产品性能也不需要担心，现在注册可以直接使用环信MQTT消息云超高额度的免费版：每月100并发连接、300万消息，完全满足功能开发使用。

　　客户端实现：

　　客户端实现主要包含以下两部分：

　　底层MQTT业务集成：包含引入SDK、MQTT方法封装、业务交互(消息收发)。

　　APP上层交互：在APP首页提供世界频道入口，实现心情弹幕飘窗(接收)和发送。

　　接下来上底层MQTT业务集成代码。

　　引入SDK：

　　这一步环信官方文档比较明确，就是根据自己的平台引入相应的mqtt客户端sdk，这里简单贴一下AndroidStudio的引入配置

　　1// 在根目录 build.gradle repositories 下加入配置

　　2maven { url "https://repo.eclipse.org/content/repositories/paho-snapshots/" }

　　3...

　　4// 然后加入 MQTT 依赖

　　5// MQTT sdk https://docs-im.easemob.com/mqtt/qsandroidsdk

　　6implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'

　　7implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'

　　方法封装

　　这里贴一下对mqtt相关方法的简单封装，代码在vmmqtt模块儿的MQTTHelper类下：

　　1 /**

　　2 * Create by lzan13 on 2022/3/22

　　3 * 描述：MQTT 帮助类

　　4 */

　　5 object MQTTHelper {

　　6

　　7    private var mqttClient: MqttAndroidClient? = null

　　8

　　9    // 缓存主题集合

　　10    private val topicList = mutableListOf()

　　11

　　12    /**

　　13     * 链接MQTT

　　14     * @param id 用户 Id

　　15     * @param token 用户链接 MQTT 的 Token

　　16     * @param topic 需要订阅的主题，不为空就会在连接成功后进行订阅

　　17     */

　　18    fun connect(id: String, token: String, topic: String = "") {

　　19        // 处理订阅主题

　　20        if (topic.isNotEmpty()) topicList.add(topic)

　　21

　　22        // 拼接链接地址

　　23        val url = "tcp://${MQTTConstants.mqttHost()}:${MQTTConstants.mqttPort()}"

　　24        // 拼接 clientId

　　25        val clientId = "${id}@${MQTTConstants.mqttAppId()}"

　　26        mqttClient = MqttAndroidClient(VMTools.context, url, clientId)

　　27

　　28        //连接参数

　　29        val options = MqttConnectOptions()

　　30        options.isAutomaticReconnect = true //设置自动重连

　　31        options.isCleanSession = true // 缓存

　　32        options.connectionTimeout = CConstants.timeMinute.toInt() // 设置超时时间，单位：秒

　　33        options.keepAliveInterval = CConstants.timeMinute.toInt() // 心跳包发送间隔，单位：秒

　　34        options.userName = id // 用户名

　　35        options.password = token.toCharArray() // 密码

　　36        options.mqttVersion = MqttConnectOptions.MQTT_VERSION_3_1_1;

　　37        // 设置MQTT监听

　　38        mqttClient?.setCallback(object : MqttCallback {

　　39            override fun connectionLost(t: Throwable) {

　　40                // 通知链接断开

　　41                VMLog.d("MQTT 链接断开 $t")

　　42            }

　　43

　　44            @Throws(Exception::class)

　　45            override fun messageArrived(topic: String, message: MqttMessage) {

　　46                // 通知收到消息

　　47                VMLog.d("MQTT 收到消息:$message")

　　48                // 如果未订阅则直接丢弃

　　49                if (!topicList.contains(topic)) return

　　50                notifyEvent(topic, String(message.payload))

　　51            }

　　52

　　53            override fun deliveryComplete(token: IMqttDeliveryToken) {}

　　54        })

　　55        //进行连接

　　56        mqttClient?.connect(options, null, object : IMqttActionListener {

　　57            override fun onSuccess(token: IMqttToken) {

　　58                VMLog.d("MQTT 链接成功")

　　59                // 链接成功，循环订阅缓存的主题

　　60                topicList.forEach { subscribe(it) }

　　61            }

　　62

　　63            override fun onFailure(token: IMqttToken, t: Throwable) {

　　64                VMLog.d("MQTT 链接失败 $t")

　　65            }

　　66        })

　　67    }

　　68

　　69    /**

　　70     * 订阅主题

　　71     * @param topic 主题

　　72     */

　　73    fun subscribe(topic: String) {

　　74        if (!topicList.contains(topic)) {

　　75            topicList.add(topic)

　　76        }

　　77        try {

　　78            //连接成功后订阅主题

　　79            mqttClient?.subscribe(topic, 0, null, object : IMqttActionListener {

　　80                override fun onSuccess(token: IMqttToken) {

　　81                    VMLog.d("MQTT 订阅成功 $topic")

　　82                }

　　83

　　84                override fun onFailure(token: IMqttToken, t: Throwable) {

　　85                    VMLog.d("MQTT 订阅失败 $topic $t")

　　86                }

　　87            })

　　88        } catch (e: MqttException) {

　　89            e.printStackTrace()

　　90        }

　　91    }

　　92

　　93    /**

　　94     * 取消订阅

　　95     * @param topic 主题

　　96     */

　　97    fun unsubscribe(topic: String) {

　　98        if (topicList.contains(topic)) {

　　99            topicList.remove(topic)

　　100        }

　　101        try {

　　102            mqttClient?.unsubscribe(topic)

　　103        } catch (e: MqttException) {

　　104            e.printStackTrace()

　　105        }

　　106    }

　　107

　　108    /**

　　109     * 发送 MQTT 消息

　　110     * @param topic 主题

　　111     * @param content 内容

　　112     */

　　113    fun sendMsg(topic: String, content: String) {

　　114        val msg = MqttMessage()

　　115        msg.payload = content.encodeToByteArray() // 设置消息内容

　　116        msg.qos = 0 //设置消息发送质量，可为0,1,2.

　　117        // 设置消息的topic，并发送。

　　118        mqttClient?.publish(topic, msg, null, object : IMqttActionListener {

　　119            override fun onSuccess(asyncActionToken: IMqttToken) {

　　120                VMLog.d("MQTT 消息发送成功")

　　121            }

　　122

　　123            override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {

　　124                VMLog.d("MQTT 消息发送失败 ${exception.message}")

　　125            }

　　126        })

　　127    }

　　128

　　129    /**

　　130     * 通知 MQTT 事件

　　131     */

　　132    private fun notifyEvent(topic: String, data: String) {

　　133        LDEventBus.post(topic, data)

　　134    }

　　135 }

　　业务交互

　　和业务相关的就是在启动APP后，使用后端服务器返回的鉴权token信息及连接封装接口登录环信通MQTT服务器，登录成功后订阅主题并监听消息。

　　1// 请求 token 成功后，调用MQTTHelper.connect()链接 MQTT 服务器，这里会同时传递监听的主题

　　2MQTTHelper.connect(mUser.id, token, MQTTConstants.Topic.newMatchInfo)

　　3

　　4/**

　　5 * 发送匹配信息

　　6 */

　　7private fun sendMatchInfo() {

　　8    if (selfMatch.user.nickname.isEmpty()) return

　　9    // 提交自己的匹配信息到服务器

　　10    mViewModel.submitMatch(selfMatch)

　　11    val json = JSONObject()

　　12    json.put("content", selfMatch.content)

　　13    json.put("emotion", selfMatch.emotion)

　　14    json.put("gender", selfMatch.gender)

　　15    json.put("type", selfMatch.type)

　　16    val jsonUser = JSONObject()

　　17    jsonUser.put("avatar", mUser.avatar)

　　18    jsonUser.put("id", mUser.id)

　　19    jsonUser.put("nickname", mUser.nickname)

　　20    jsonUser.put("username", mUser.username)

　　21    json.put("user", jsonUser)

　　22    MQTTHelper.sendMsg(MQTTConstants.Topic.newMatchInfo, json.toString())

　　23}

　　24

　　25// 监听消息这里使用了一个事件总线进行通知，在上边封装 MQTTHelper 发送消息也使用了这个，

　　26// 订阅 MQTT 事件

　　27LDEventBus.observe(this, MQTTConstants.Topic.newMatchInfo, String::class.java) {

　　28        val match = JsonUtils.fromJson(it, Match::class.java)

　　29        // 这里收到匹配信息之后就增加一条弹幕

　　30    addBarrage(match)

　　31}

　　后端服务实现

　　接下来介绍后端服务实现，主要包含以下两部分：

　　配置连接信息：配置环信MQTT消息云连接信息。

　　获取鉴权信息：获取客户端连接需要的鉴权信息。

　　配置连接信息

　　配置部分只需要按照环信后台配置信息进行替换就好，配置在config目录下的config.xxx.json文件内

　　1/**

　　2 * Easemob MQTT 配置 https://console.easemob.com/app/generalizeMsg/overviewService

　　3 */

　　4config.mqtt = {

　　5    host: 'mqtt host', // MQTT 链接地址

　　6  appId: 'appId', // MQTT AppId

　　7  port: [ 1883, 1884, 80, 443 ], // MQTT 端口 1883(mqtt),1884(mqtts),80(ws),443(wss)

　　8  restHost: 'https://api.cn1.mqtt.chat/app/8igtc0', // MQTT 服务 API 地址

　　9  clientId: 'client id', // 替换环信后台 clientId

　　10  clientSecret: 'client secret', // 替换环信后台 clientSecret

　　11};

　　获取鉴权信息

　　这里主要是获取客户端连接所需要的鉴权信息token，为了安全token肯定是要放在服务器端生成的，废话不多说，上代码：

　　1/**

　　2 * Create by lzan13 on 2022/3/22

　　3 * 描述：MQTT 帮助类

　　4 */

　　5object MQTTHelper {

　　6

　　7    private var mqttClient: MqttAndroidClient? = null

　　8

　　9    // 缓存主题集合

　　10    private val topicList = mutableListOf()

　　11

　　12    /**

　　13     * 链接MQTT

　　14     * @param id 用户 Id

　　15     * @param token 用户链接 MQTT 的 Token

　　16     * @param topic 需要订阅的主题，不为空就会在连接成功后进行订阅

　　17     */

　　18    fun connect(id: String, token: String, topic: String = "") {

　　19        // 处理订阅主题

　　20        if (topic.isNotEmpty()) topicList.add(topic)

　　21

　　22        // 拼接链接地址

　　23        val url = "tcp://${MQTTConstants.mqttHost()}:${MQTTConstants.mqttPort()}"

　　24        // 拼接 clientId

　　25        val clientId = "${id}@${MQTTConstants.mqttAppId()}"

　　26        mqttClient = MqttAndroidClient(VMTools.context, url, clientId)

　　27

　　28        //连接参数

　　29        val options = MqttConnectOptions()

　　30        options.isAutomaticReconnect = true //设置自动重连

　　31        options.isCleanSession = true // 缓存

　　32        options.connectionTimeout = CConstants.timeMinute.toInt() // 设置超时时间，单位：秒

　　33        options.keepAliveInterval = CConstants.timeMinute.toInt() // 心跳包发送间隔，单位：秒

　　34        options.userName = id // 用户名

　　35        options.password = token.toCharArray() // 密码

　　36        options.mqttVersion = MqttConnectOptions.MQTT_VERSION_3_1_1;

　　37        // 设置MQTT监听

　　38        mqttClient?.setCallback(object : MqttCallback {

　　39            override fun connectionLost(t: Throwable) {

　　40                // 通知链接断开

　　41                VMLog.d("MQTT 链接断开 $t")

　　42            }

　　43

　　44            @Throws(Exception::class)

　　45            override fun messageArrived(topic: String, message: MqttMessage) {

　　46                // 通知收到消息

　　47                VMLog.d("MQTT 收到消息:$message")

　　48                // 如果未订阅则直接丢弃

　　49                if (!topicList.contains(topic)) return

　　50                notifyEvent(topic, String(message.payload))

　　51            }

　　52

　　53            override fun deliveryComplete(token: IMqttDeliveryToken) {}

　　54        })

　　55        //进行连接

　　56        mqttClient?.connect(options, null, object : IMqttActionListener {

　　57            override fun onSuccess(token: IMqttToken) {

　　58                VMLog.d("MQTT 链接成功")

　　59                // 链接成功，循环订阅缓存的主题

　　60                topicList.forEach { subscribe(it) }

　　61            }

　　62

　　63            override fun onFailure(token: IMqttToken, t: Throwable) {

　　64                VMLog.d("MQTT 链接失败 $t")

　　65            }

　　66        })

　　67    }

　　68

　　69    /**

　　70     * 订阅主题

　　71     * @param topic 主题

　　72     */

　　73    fun subscribe(topic: String) {

　　74        if (!topicList.contains(topic)) {

　　75            topicList.add(topic)

　　76        }

　　77        try {

　　78            //连接成功后订阅主题

　　79            mqttClient?.subscribe(topic, 0, null, object : IMqttActionListener {

　　80                override fun onSuccess(token: IMqttToken) {

　　81                    VMLog.d("MQTT 订阅成功 $topic")

　　82                }

　　83

　　84                override fun onFailure(token: IMqttToken, t: Throwable) {

　　85                    VMLog.d("MQTT 订阅失败 $topic $t")

　　86                }

　　87            })

　　88        } catch (e: MqttException) {

　　89            e.printStackTrace()

　　90        }

　　91    }

　　92

　　93    /**

　　94     * 取消订阅

　　95     * @param topic 主题

　　96     */

　　97    fun unsubscribe(topic: String) {

　　98        if (topicList.contains(topic)) {

　　99            topicList.remove(topic)

　　100        }

　　101        try {

　　102            mqttClient?.unsubscribe(topic)

　　103        } catch (e: MqttException) {

　　104            e.printStackTrace()

　　105        }

　　106    }

　　107

　　108    /**

　　109     * 发送 MQTT 消息

　　110     * @param topic 主题

　　111     * @param content 内容

　　112     */

　　113    fun sendMsg(topic: String, content: String) {

　　114        val msg = MqttMessage()

　　115        msg.payload = content.encodeToByteArray() // 设置消息内容

　　116        msg.qos = 0 //设置消息发送质量，可为0,1,2.

　　117        // 设置消息的topic，并发送。

　　118        mqttClient?.publish(topic, msg, null, object : IMqttActionListener {

　　119            override fun onSuccess(asyncActionToken: IMqttToken) {

　　120                VMLog.d("MQTT 消息发送成功")

　　121            }

　　122

　　123            override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {

　　124                VMLog.d("MQTT 消息发送失败 ${exception.message}")

　　125            }

　　126        })

　　127    }

　　128

　　129    /**

　　130     * 通知 MQTT 事件

　　131     */

　　132    private fun notifyEvent(topic: String, data: String) {

　　133        LDEventBus.post(topic, data)

　　134    }

　　135}

　　源码地址

　　核心代码就这么多，不超过500行，这里没有直接调用环信历史消息接口获取消息存储记录，后续可以在进行改良，简化实现流程。源码链接附上，配合使用效果更佳。

　　服务端github源码：

　　https://github.com/lzan13/vmtemplateserver

　　客户端github源码：

　　https://gitee.com/lzan13/VMTemplateAndroid

　　写在最后

　　MQTT协议资源占用小，并发连接高，集成简单，特别适用于高频数据交互场景，比如：游戏的世界广场、视频平台弹幕等等等等，欢迎各位小伙伴集思广益，基于MQTT服务实现更多的业务场景，享受技术带来的便利与快乐。