自己动手实现即时消息功能

原创
2020/12/16 13:26
阅读数 91

文:徐江威

功能设计

即时消息的两个基本功能就是发送消息和接收消息。我们定义如下通信指令来实现这两个功能:

  • Push 推送消息
  • Pull 拉取消息
  • Notify 消息通知

Push 推送消息指令将客户端消息发给指定的对端,也就是说服务器需要在收到客户端 Push 指令时将消息转发给目标客户端。

Pull 拉取消息指令用于客户端在必要时(例如,移动平台的 App 从后台回到前台时)从服务器端获取未被即时推送的消息。

Notify 消息通知是这三个指令里唯一个服务器发给客户端的指令,用来通知客户端“你有新消息送达”。客户端在收到该指令后即可将消息进行存储和呈现。

Push Sequence

 

Pull Sequence

 

当然对于即时消息来说除了收发消息,还需要实现召回消息、转发消息、撤回消息等功能,我们将在后面的章节再介绍这些功能的实现。

 

数据包结构

我们定义一个 Packet 结构来封装我们的数据包,Packet 包含三个主要属性:序列号、名称、负载。

  • 序列号 - 标识每个 Packet 的序号,该序列号唯一标识了一个 Packet 。
  • 名称 - 标记该 Packet 的名称,我们使用名称来表示不同业务操作。
  • 负载 - Packet 携带的数据,这里我们约定负载遵循 JSON 格式。
字段 类型 是否必填 描述
sn long Y 序列号
name string Y 名称
data JSON Y 负载数据

 

数据实体设计

我们定义消息实体对象具有以下字段:

字段 类型 是否必填 描述
id long Y 消息的 ID
from long Y 消息的发送人 ID
to long Y 消息的接收人 ID
source long Y 消息的广播源 ID,一般在群组消息里标记为群组 ID
owner long Y 消息的副本持有人 ID,该字段客户端程序无需识别和使用
lts long Y Local Timestamp 消息本地时间戳
rts long Y Remote Timestamp 消息服务器时间戳
state int Y 消息状态
payload JSON N 消息的负载数据
attachment JSON N 消息的附件

对于消息实体我们约定每条消息对于每个客户端账户使用 ID 字段来唯一标识,这是因为我们这里将采用副本方式来管理消息实体,即每个客户端账户都拥有该 ID 消息的副本。这样做的好处是我们在生成副本后,各个客户端账户各自维护自己的副本及副本状态,有利于消息按照时序进行管理,提高服务器处理消息的读写速度,当然缺点也比较明显,就是不能用一个 ID 来全局标识一条消息,在进行存储管理时需要配合其他字段来进行操作。

 

编写服务器程序

我们的服务器将使用 Java 作为主要的编码语言。

准备工作

在开始实现服务器程序前,我们需要从魔方服务器依赖库下载我们需要的依赖库。

git clone https://gitee.com/shixinhulian/cube-server-dependencies.git

你需要在你服务器工程中指定依赖 cube-server-dependencies 目录下的 cell-2.3.jar 的 JAR 包。这个包封装了我们需要的即时通信的网络实现,该依赖包的文档可以从魔方手册下载:

git clone https://gitee.com/shixinhulian/cube-manual.git

服务器代码

如前所述,服务器需要接收来自客户端的 Push 和 Pull 指令,我们需要在 Cellet 里实现对这两个数据包的接收处理:

public class MessagingCellet extends Cellet {

    private ExecutorService executor;

    public MessagingCellet() {
        super("Messaging");
    }

    @Override
    public boolean install() {
        ...
    }

    @Override
    public void uninstall() {
        ...
    }

    @Override
    public void onListened(TalkContext talkContext, Primitive primitive) {
        ActionDialect dialect = DialectFactory.getInstance().createActionDialect(primitive);
        String action = dialect.getName();

        if (action.equals("push")) {
            // 接收到 Push 指令
            this.executor.execute(new PushTask(this, talkContext, primitive));
        }
        else if (action.equals("pull")) {
            // 接收到 Pull 指令
            this.executor.execute(new PullTask(this, talkContext, primitive));
        }
    }
}

MessagingCellet 中,我们接收 pushpull 两个动作,并使用并发执行器分别执行 PushTaskPullTask 任务。

PushTask 代码:

public class PushTask extends ServiceTask {

    public PushTask(Cellet cellet, TalkContext talkContext, Primitive primitive) {
        super(cellet, talkContext, primitive);
    }

    @Override
    public void run() {
        ActionDialect action = DialectFactory.getInstance().createActionDialect(this.primitive);
        // 创建 Packet
        Packet packet = new Packet(action);

        // 从 Packet 负载里获取消息
        Message message = new Message(packet);

        // 生成消息副本 - 生成 From 的副本
        Message fromCopy = new Message(message);
        fromCopy.setOwner(message.getFrom());

        // 生成消息副本 - 生成 To 的副本
        Message toCopy = new Message(message);
        toCopy.setOwner(message.getTo());

        // 将消息写入时序缓存 - 写入 From 缓存
        messageCache.add(fromCopy.getOwner(), fromCopy);

        // 将消息写入时序缓存 - 写入 To 缓存
        messageCache.add(toCopy.getOwner(), toCopy);

        // 将消息发到实时队列,服务器通过监听方式将消息推送给 To 和 From 的其他设备
        contactEventQueue.publish(toCopy.getOwner(), toCopy);
        contactEventQueue.publish(fromCopy.getOwner(), fromCopy);
    }
}

完整的服务器代码可以在 Gitee 或者 GitHub 上找到。

这里我们仅关注消息数据的流转过程。在消息的处理过程里,我们为一个消息生成了两个副本,并将两个副本分别保存,两个副本将通过全局队列发送给对应的服务器,以便连接到这些服务器上的客户端接收到消息。因此,消息发送人(From)的其他设备也会接收到自己来自其他设备的消息,从而实现多设备消息的同步。

 

编写客户端程序(Web版/ES6)

网络通信实现

在 Web 版的客户端里我们将使用 WebSocket 作为通信协议的实现,我们先创建一个 Pipeline 对象,Pipeline 作为网络数据通道,客户端的数据发送和接收都通过 Pipeline 来操作:

class Pipeline extends cell.TalkListener {
    constructor() {
        this.nucleus = new cell.Nucleus();
        this.nucleus.talkService.addListener(this);
        this.listener = [];
    }

    ...

    /** 发送数据到服务器 */
    send(packet, handleResponse) {
        let primitive = packet.toPrimitive();
        this.nucleus.talkService.speak('Messaging', primitive);
    }

    /** 添加接收数据监听器 */
    addListener(listener) {
        this.listener.push(listener);
    }

    /** 收到消息时的回调 */
    onListened(speaker, cellet, primitive) {
        this.listener.forEach((item) => {
            item(primitive);
        });
    }
}

我们使用了 cell.Nucleus 来管理 WebSocket 连接,cell 库可以从 cell-javascript 下载。

消息程序代码

我们先定义消息服务对象作为消息处理的入口对象:

class MessagingService {
    constructor() {
        this.pipeline = new Pipeline();
    }

    /** 启动服务。 */
    start() {
        this.pipeline.addListener((primitive) => {
            this.onListened(primitive);
        });
    }

    /** 停止服务。 */
    stop() {
        ...
    }

    /** 发送消息。 */
    sendTo(message) {
        let packet = new Packet('push', message.toJSON());
        this.pipeline.send(packet);
    }

    onListened(primitive) {
        // 将网络层的数据结构转为 Packet 结构
        let packet = new Packet(primitive);
        if (packet.name == 'notify') {
             // 接收到新消息
             ...
        }
    }
}
<!doctype html>
<html>
<head>
  <meta charset="utf-8">
  <title>即时消息客户端</title>
</head>
<body>
<script src="js/cell.js"></script>
<script src="js/Pipeline.js"></script>
<script src="js/MessagingService.js"></script>
<script type="text/javascript">
let service = new MessagingService();
service.start();

// 发送人 ID 10000 ,接收人 ID 20000
let message = new Message(10000, 20000, 'Hello! This is a message');

// 发送消息
service.sendTo(message);
</script>
</body>
</html>

客户端的完整代码可以从 Gitee 或者 GitHub 上下载。


至此我们就实现了基本的消息收发功能。但是,在实际项目中对于消息的操作和管理远比上述的演示程序复杂。如果你从 Gitee 或者 GitHub 下载代码并阅读,会发现项目中的代码还需要处理消息超时、消息状态、消息时间戳等问题。而真正应用在产品里时,还需要对消息的内容进行必要的风险管理,以便识别有风险的消息内容。

 

展开阅读全文
打赏
0
1 收藏
分享
加载中
更多评论
打赏
0 评论
1 收藏
0
分享
返回顶部
顶部