基于EMQX设计的IM系统

站长

2024年03月27日 18:10 · 阅读数 6

一、IM系统介绍：

作为一个互联网时代的公司无论你的产品是什么，你都需要一个网站还是app亦或是一个pc软件作为自己的网络平台，而在网络平台当中，即时收集用户反馈、与用户进行互动、方便用户之间互动提高产品活跃度只要存在这些需求，那么你的产品都需要IM（即时通信）的能力，尤其是在客服系统中IM功能更是必不可少的。

一个最简单的IM系统，发送方通过调用接口把消息内容存储到数据库中，然后接收方调用接口到数据库去查询最新的消息，虽然是简单了，但是性能非常低，消息接收方并不知道别人给他发了消息，因此他就需要做一个定时任务定期去数据库查询，定时任务的时间间隔如果过短则会严重占用客户端和服务端的资源，性能非常低下；如果时间间隔过长，则消息的延时太长了，达不到即时通讯的效果。而要设计一个稳定不丢消息且性能高效的IM系统绝非一件容易的事情，正是因为如此，所以很多公司都是采用接入第三方成熟稳定的IM系统。

而接入第三方IM系统，除了价格昂贵之外，第三方IM系统很难兼顾接入的各个系统的个性化的功能，而且数据不可避免的存储在别人那里，远没有数据存储在自己的服务器上安全。因此自研一套IM框架就很有必要。

IM系统分为基于短连接和基于长连接的，这两种各有优缺点。

1、长连接的IM系统在连接建立后就不释放，一直保持连接，直到用户关闭连接或者退出客户端才关闭连接，因此效率更高，但是需要维护长连接，比如说通过心跳来维护，客户端每隔一定时间给服务端发送一个心跳包来告诉服务器客户端在正常运行且网络也是正常的，如果连续几次服务端收不到客户端的心跳包，则认为客户端网络出现异常从而关闭与这个客户端的连接。在客户端网络出现异常到服务器关闭连接的这段时间，如果有消息发出就会导致对方收不到而丢消息，为保证不丢消息就需要设计一个复杂的机制来保证消息的可靠到达。再一个如果服务端有多个服务实例，而某个客户端只与其中一台服务实例建立了连接，因此路由机制必须保证每次都能路由到这台服务实例上，且消息的接收方如果不是连接的这台服务实例的时候，必须将消息转发给消息接收方所连接的服务实例处理，不然消息是没法发送出去的，不像短连接那样任何一台服务实例都可以处理。

2、短连接的IM系统是通过HTTP/HTTPS接口实现的，由于HTTP协议是每次都建立连接、发出请求、返回应答、关闭连接，因此在效率上肯定没有长连接的高，但是他不用心跳来维护连接，且客户端发出HTTP请求是同步进行的，也就是说发出请求后如果不能收到服务端的正确应答则认为请求失败，必须要再次请求，直到收到服务端的正确应答，因此协议本身就保证了消息的可靠达到。在多服务实例的情况下，由于每台服务实例不用保存与客户端的连接信息，因此客户端的请求路由到任何一台服务实例上都可以处理。但是短连接下只能客户端向服务端发送消息，而服务端不能主动向客户端推送消息，虽然也可以通过客户端定期向服务端拉取消息，但是一来性能低下，二来会出现消息的延时性，做不到消息的即时性。

在本篇中讨论短连接下的IM系统设计方案，为解决服务端不能向客户端推送消息的问题，引入了中间件EMQX。

二、EMQX介绍：

1、EMQX简介

MQTT协议是一种基于发布/订阅模式的“轻量级”通讯协议，该协议构建于TCP/IP协议上。

实现MQTT协议需要客户端和服务器端通讯完成，在通讯过程中，MQTT协议中有三种身份：发布者（Publisher）、代理（Broker）（服务器）、订阅者（Subscriber）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

MQTT传输的消息分为：主题（Topic）和负载（payload）两部分：

（1）Topic，可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（payload）；

（2）payload，可以理解为消息的内容，是指订阅者具体要使用的内容。

MQTT协议中设计了个Qos服务等级

（1）、Qos0 : 消息最多传递1次，如果当时客户端不可用，则会丢失该消息。

（2）、Qos1 : 消息至少传递1次

（3）、Qos2 : 消息仅传送一次

EMQX的Topic的层级分隔符是“/”，且支持通配符，#表示多层通配符，可以表示大于等于0的层次，+表示只匹配主题的一层。

2、EMQX的认证：

身份认证是大多数应用的重要组成部分，MQTT 协议支持多种认证方式，能有效阻止非法客户端的连接。这里以HTTP认证方式来举例。

HTTP 认证使用外部自建 HTTP 应用认证数据源，根据 HTTP API 返回的数据判定认证结果，能够实现复杂的认证鉴权逻辑。

认证原理：

EMQ X 在设备连接事件中使用当前客户端相关信息作为参数，向用户自定义的认证服务发起请求查询权限，通过返回的 HTTP 响应状态码 (HTTP statusCode) 来处理认证请求。

认证失败：API 返回 4xx 状态码
认证成功：API 返回 200 状态码
忽略认证：API 返回 200 状态码且消息体 ignore

认证步骤：

首先需要开启插件emqx_auth_http；

接着修改配置文件：

vi /etc/emqx/plugins/emqx_auth_http.conf

#其余配置尚未修改，这里就不展示了
auth.http.auth_req.url = http://192.168.40.128:8991/mqtt/auth
auth.http.auth_req.method = post
auth.http.auth_req.headers.content_type = application/x-www-form-urlencoded
auth.http.auth_req.params = clientid=%c,username=%u,password=%P

你可以在认证请求中使用以下占位符，请求时 EMQ X 将自动填充为客户端信息：

%u：用户名
%c：Client ID
%a：客户端 IP 地址
%r：客户端接入协议
%P：明文密码
%p：客户端端口
%C：TLS 证书公用名（证书的域名或子域名），仅当 TLS 连接时有效
%d：TLS 证书 subject，仅当 TLS 连接时有效

最后实现认证代码：

@RestController
@RequestMapping("/mqtt")
public class AuthController {

    Logger logger = LoggerFactory.getLogger(AuthController.class);

    Map<String, String> userMap = new HashMap<>();

    @PostConstruct
    public void init(){
        userMap.put("user1","user1");
        userMap.put("user2","user2");
    }


    @PostMapping("/auth")
    public ResponseEntity<HttpStatus> auth(String username, String clientid, String password){
        logger.info("开始进入auth方法");
        logger.info("username:{}; clientid:{}; password:{}", username, clientid, password);
        String passwd = userMap.get(username);
        if(!StringUtils.hasLength(passwd)){
            return new ResponseEntity<>(HttpStatus.UNAUTHORIZED);
        }
        return !password.equals(passwd) ? new ResponseEntity<>(HttpStatus.UNAUTHORIZED) : new ResponseEntity<>(HttpStatus.OK);
    }
}

3、WebHook：

WebHook 是由 emqx_web_hook插件提供的将 EMQ X 中的钩子事件通知到某个 Web 服务的功能。它通过在钩子上的挂载回调函数，获取到 EMQ X 中的各种事件，并转发至 emqx_web_hook 中配置的 Web 服务器。

WebHook 对于事件的处理是单向的，它仅支持将 EMQ X 中的事件推送给 Web 服务，并不关心 Web 服务的返回。借助 Webhook 可以完成设备在线、上下线记录，订阅与消息存储、消息送达确认等诸多业务。

webHook的帮助文档页面： https://www.emqx.io/docs/zh/v4.…

EMQX支持的事件通知有：

名称	说明	执行时机
client.connect	处理连接报文	服务端收到客户端的连接报文时
client.connack	下发连接应答	服务端准备下发连接应答报文时
client.connected	成功接入	客户端认证完成并成功接入系统后
client.disconnected	连接断开	客户端连接层在准备关闭时
client.subscribe	订阅主题	收到订阅报文后，执行 `client.check_acl` 鉴权前
client.unsubscribe	取消订阅	收到取消订阅报文后
session.subscribed	会话订阅主题	完成订阅操作后
session.unsubscribed	会话取消订阅	完成取消订阅操作后
message.publish	消息发布	服务端在发布（路由）消息前
message.delivered	消息投递	消息准备投递到客户端前
message.acked	消息回执	服务端在收到客户端发回的消息 ACK 后
message.dropped	消息丢弃	发布出的消息被丢弃后

接入EMQX事件的步骤：

首先第一步还是开启插件emqx_web_hook；

接着修改配置文件：

vi etc/plugins/emqx_web_hook.conf

//事件需要转发的目的服务器地址
api.url=http://127.0.0.1:8080
//对消息类事件中的 Payload 字段进行编码，注释或其他则表示不编码（取值有base64, base62两种）
encode_payload=base64

## 格式示例
web.hook.rule.<Event>.<Number> = <Rule>

## 示例值
web.hook.rule.message.publish.1 = {"action": "on_message_publish", "topic": "a/b/c"}
web.hook.rule.message.publish.2 = {"action": "on_message_publish", "topic": "foo/#"}

最后实现代码接收处理EMQX事件：

@PostMapping("/mqtt/action/web_hook")
public ResponseEntity webHook(@RequestBody String body) {
    //......省略部分代码.........
    JsonNode bodyNode = objectMapper.readValue(body, JsonNode.class);
    JsonNode actionNode = bodyNode.get("action");
    String actionName = actionNode.textValue();
    MQDeclareTemplate.MQDeclare mqDeclare = emqProperties.getClientDeclare();
    routingKey = actionName.split("_")[1];
	
    ```
    if (actionName.startsWith("client")) {
        Message msg = this.convertMsg(body, Message.class);
        this.hasClientRouter(message);
    }
	
    return new ResponseEntity(HttpStatus.OK);
}

private void hasClientRouter(Message message) {
    String routingKey = message.getMessageProperties().getReceivedRoutingKey();
    String queue = message.getMessageProperties().getConsumerQueue();
    try {
        JsonNode msgNode = objectMapper.readValue(message.getBody(), JsonNode.class);       
        String action = msgNode.get("action").asText();
        EmqClientParseUtils.ClientPair clientPair = EmqClientParseUtils.parse(msgNode.get("clientid").asText());        
        EmqClientParseUtils.UserNamePair userNamePair = EmqClientParseUtils.parseUserName(msgNode.get("username").asText());        
        if ("client_connect".equalsIgnoreCase(action)) {
            emqxClientEventHook.connect(clientPair, userNamePair, msgNode);
        } else if ("client_connected".equalsIgnoreCase(action)) {
            emqxClientEventHook.connected(clientPair, userNamePair, msgNode);
        } else if ("client_disconnected".equalsIgnoreCase(action)) {
            emqxClientEventHook.disconnected(clientPair, userNamePair, msgNode);
        }
    } catch (Exception e) {
        logger.warn(e.getMessage(), e);
    }
}

@Override
public void connect(EmqClientParseUtils.ClientPair clientPair, EmqClientParseUtils.UserNamePair userNamePair, JsonNode message) {
    process(IOnlineAndOffLineConsumer.CONNECT,clientPair.getClientId());
}

@Override
public void connected(EmqClientParseUtils.ClientPair clientPair, EmqClientParseUtils.UserNamePair userNamePair, JsonNode message) {
    process(IOnlineAndOffLineConsumer.CONNECTED,clientPair.getClientId());
}

@Override
public void disconnected(EmqClientParseUtils.ClientPair clientPair, EmqClientParseUtils.UserNamePair userNamePair, JsonNode message) {
    process(IOnlineAndOffLineConsumer.DISCONNECTED,clientPair.getClientId());
}

private void process(String type,String userId){    
    List<IOnlineAndOffLineConsumer> consumers=map.get(type);
    if(CollectionUtils.isEmpty(consumers)){
        return;
    }
    consumers.forEach(consumer->consumer.accept(Long.valueOf(userId)));
}

首先在clientContainerIM中指定emqx webhook事件消息的接收者是mqConsumer::onClient，接着onClient调用hasClientRouter，hasClientRouter中解析client_connect, client_connected, client_disconnected几个事件，对这几个事件的处理，则是循环调用IOnlineAndOffLineConsumer接口，因此业务系统要处理客户端上线连接和下线的事件处理，只需要实现IOnlineAndOffLineConsumer接口，在接口中实现自己的业务代码即可。

三、IM整体架构：

基于EMQX设计的IM系统

可以看到系统整体架构是一个标准的微服务架构，在业务层有各种微服务，IM服务也是作为其中一个，调用IM服务的接口也和其他服务的接口一样，都需要先登录认证获取token，然后前端的请求经过网关验证，验证通过则请求转发到业务层上。

前端在启动时就需要连接到EMQX上，这里EMQX也需要授权验证，授权验证用到了上面介绍的EMQX的emqx_auth_http插件，上面有详细介绍。同时客户端还需要订阅Topic，Topic名称为：p2p/{userId}，其中userId为当前用户自己的userId，这样服务端在publish时只需要发到这个Topic就行了。

四、发送消息的流程：

1、p2p消息发送流程：

基于EMQX设计的IM系统

1)、发送消息的Rest接口，首先调用前置处理接口和生成消息服务端的唯一ID ServerMID以及SessionId。有前置处理接口就必然有后置处理接口，这里定义成接口就是留给业务系统去实现的，以便实现在发送消息前对消息内容进行修改和发送完成之后进行后置处理。消息的唯一ID分两种：客户端唯一ID（ClientMID）和服务端唯一ID（ServerMID），ClientMID是消息发送方生成的，在发送方的客户端是唯一的，ServerMID是服务端生成的，可以通过这个ID进行消息去重处理。SessionId是表示这一个会话的唯一ID，比如对于p2p会话，SessionId可以表示成“userId1-userId2”，也就是将发送方的userId和接收方的userId排序，小的排在前面，大的排在后面，这样不管是userId1给userId2发起会话，还是userId2给userId2发起会话，他们的SessionId是一样的。ServerMID和SessionId都可以定义成接口交给业务方去实现。

2)、将消息内容保存到消息表中；

3)、根据前面生成的SessionId检查会话表中是否存在，如果不存在则生成新的会话记录保存到会话表中；

4)、调用消息发送的后置处理接口；

5)、为提高接口的响应速度，迅速处理大量消息的快速发送，这里采用了MQ进行消息异步处理和削峰处理。将消息内容封装好交给MQ并返回；

6)、消费上一步的MQ消息，解析出消息内容，生成Topic：p2p/{targetuserid}，将消息通过emqx接口mqtt/publish推送给客户端，由于客户端在启动时就已经订阅了Topic p2p/{selfUserId}，因此客户端是能收到该条消息的。如果客户端没有上线是收不到消息的，也就是不会发送回执ack信号的，因此这条消息就存储在数据库中并标记为未读消息，这样下次客户端启动时会调用接口拉取所有的未读消息。

7)、由于上一步的消息推送有可能出现消息丢失的情况，因此这里需要做消息可靠性达到的处理：封装消息，设置好消息的delayTime和retryCount并交给MQ；

8)、上一步交给MQ的消息在达到delayTime后会自动进入死信队列，然后被消费到；

9)、如果该消息在以及回执过的消息列表中，则返回successCall（successCall可以定义好接口给业务方实现，同时也意味着客户端已经成功收到消息，不用再做重试处理）；

10)、判断消息重试的次数是否大于最大重试次数（我这里设置的最大重试次数是3），如果大于则说明接收客户端可能掉线或者网络出现故障，直接返回timeoutCall并从MQ队列移除；

11)、调用retryCall，retryCall就是通过emqx接口mqtt/publish再次推送消息给客户端；

12)、retryCount+1，再次封装消息交给MQ的死信队列，重复上面的步骤处理。

2、消息回执处理：

1）、客户端接收到消息后，可以手工发送ack回执消息，也可以通过接口方法参数自动回执ack标记；

2）、EMQX Broker接收到回执消息后，会通过webHook插件发送message.acked事件给我们的IM服务；我们的IM服务中需要定义一个Rest接口接收事件通知，由于事件太多，为加快接口响应速度，以便让接口能快速的处理更多事件通知，这里可以将收到的事件通知交给MQ并迅速返回；

3）、消息刚才的MQ消息，从消息内容中解析出userId，将该userId保存到redis的ack信息中，这样在消息发送的重试机制中就能读取到该消息的ack标记。