即时通讯系统服务器端设计，教你⽤纯Java实现⼀个即时通讯
系统（附源码）
项⽬背景
和各位读者⼤致介绍下具体场景，线上的⼩程序中开放⼀些语⾳麦克风的房间，让⽤户进⼊房间之后可以互相通过语⾳聊天的⽅式进⾏互动。
这⾥分享⼀下相关的技术设计⽅案。这款系统的核⼼点设计在于如何能让⼀个⽤户发出的语⾳通知到其他⽤户上边。语⾳数据在客户端同事的处理下最终变成了io数据流请求到了后端，后端只需要将这些数据流传达给各个不同的终端即可达到⼴播通知的效果。 单机版架构
最初期上线的时候，为了赶速度，快速试错，所以简单地采⽤了单机版架构去设计。结合技术栈为 SpringBoot，WebSocket，MySQL技术。 线上⼀间语⾳房间的同时在线⼈数并不会特别多，⼤概在15-50⼈的区间段内，系统核⼼代码是通过SpringBoot内部的WebSocket技术去进⾏数据的主动推送。
设计思路
整体的设计图⽐较简单，基本就是⼀台服务器存储WebSocket连接，如下图所⽰：
⽤户进⾏WebSocket初始化连接的时候需要⼀个连接分配和存储的过程：
早期的存储是存放在了服务器本地的⼀个Map集合中。
当WebSocket进⾏连接的时候就会往内存中写⼊⼀条数据信息，当链接断开的时候，就将内存中的数据移除。然后进⾏语⾳⼴播的时候需要结合WebSocket内部的⼴播发送功能进⾏通知
看似设计⽐较简单，但是在后期业务变得庞⼤的时候出现了瓶颈。因为随着参加语⾳活动⽤户的增加，越来越多的WebSocketSession对象需要被存储到内存当中，这种有状态性的存储对于单机扩容不灵活。
设计缺陷
1.假设原先的服务器扩容到了A，B两台机器，A⽤户在A机器上边建⽴了WebSocketSession，B⽤户在B机器上边建⽴的WebSocketSession连接。此时如果A想要和B进⾏对话发送，需要先查到具体
WebSocketSession存放在哪台机器上边。
2.当⽤户出现了⽹络异常，临时断开连接进⾏重连的时候，也可能会出现1所说的问题。
集架构
设计思路
⼀旦出现需要发送语⾳通知的时候，发送⼀条⼴播的mq消息，每个机器都接收到消息之后，触发⾃⼰的⼴播操作即可。 RocketMq的接⼊系统设计⾥⾯mq采⽤的是⼴播模式，这和我们通常使⽤的集模式有⼀定的区别。
消息队列RocketMQ版是基于发布或订阅模型的消息系统。消费者，即消息的订阅⽅订阅关注的Topic，以获取并消费消息。由于消费者应⽤⼀般是分布式系统，以集⽅式部署，因此消息队列RocketMQ版约定以下概念： 集：使⽤相同Group ID的消费者属于同⼀个集。同⼀个集下的消费者消费逻辑必须完全⼀致(包括Tag的使⽤)。
集消费：当使⽤集消费模式时，消息队列RocketMQ版认为任意⼀条消息只需要被集内的任意⼀个消费者处理即可。
⼴播消费：当使⽤⼴播消费模式时，消息队列RocketMQ版会将每条消息推送给集内所有注册过的消费者，保证消息⾄少被每个消费者消费⼀次。
集消费模式适⽤场景 适⽤于消费端集化部署，每条消息只需要被处理⼀次的场景。此外，由于消费进度在服务端维护，可靠性更⾼。具体消费⽰例如下图所⽰。
注意事项
集消费模式下，每⼀条消息都只会被分发到⼀台机器上处理。如果需要被集下的每⼀台机器都处理，请使⽤⼴播模式。
集消费模式下，不保证每⼀次失败重投的消息路由到同⼀台机器上。
⼴播消费模式适⽤场景 适⽤于消费端集化部署，每条消息需要被集下的每个消费者处理的场景。具体消费⽰例如下图所⽰。
注意事项
⼴播消费模式下不⽀持顺序消息。
⼴播消费模式下不⽀持重置消费位点。
每条消息都需要被相同订阅逻辑的多台机器处理。
消费进度在客户端维护，出现重复消费的概率稍⼤于集模式。
⼴播模式下，消息队列RocketMQ版保证每条消息⾄少被每台客户端消费⼀次，但是并不会重投消费失败的消息，因此业务⽅需要关注消费失败的情况。
⼴播模式下，客户端每⼀次重启都会从最新消息消费。客户端在被停⽌期间发送⾄服务端的消息将会被⾃动跳过，请谨慎选择。
⼴播模式下，每条消息都会被⼤量的客户端重复处理，因此推荐尽可能使⽤集模式。
⼴播模式下服务端不维护消费进度，所以消息队列RocketMQ版控制台不⽀持消息堆积查询、消息堆积报警和订阅关系查询功能。
这⾥⾯的应⽤场景需要对集内部对每个消费者都对服务器内存中的socket连接进⾏session是否存在对判断，因此需要采⽤mq的⼴播模式。
关于mq部分的接⼊代码
Consumer模块的配置：
package org.idea.fig;
import org.t.properties.ConfigurationProperties;
/**
刀模
* @Author linhao
* @Date created in 10:30 上午 2021/5/10
*/
@ConfigurationProperties(prefix="sumer")
public class MqConsumerConfig {
模板的制作
private boolean isOn;
private String groupName;
private String nameSrvAddr;
private String topics;
private Integer consumeThreadMin;
private Integer consumeThreadMax;
private Integer consumeMessageBatchMaxSize;
/**
getter 和 setter部分省略
**/
}
Producer模块的配置展⽰：微波热疗机
package org.idea.fig;
import org.t.properties.ConfigurationProperties;
/**
蚊帐架* @Author linhao
* @Date created in 10:26 上午 2021/5/10
*/
@ConfigurationProperties(prefix="rocketmq.producer")
public class MqProducerConfig {
private boolean isOn;
private String groupName;
private String nameSrvAddr;
private Integer maxMessageSize;
private Integer sendMsgTimeout;
private Integer retryTimesWhenSendFailed;
/**
getter 和 setter部分省略
防臭鞋
**/
}
RocketMq内部的消费端Bean配置
package org.idea.web.socket.mq;
slf4j.Slf4j;
import sumer.DefaultMQPushConsumer;
import sumer.listener.MessageListenerConcurrently; import ption.MQClientException;
import sumer.ConsumeFromWhere;声音设备
import ketmqmon.protocol.heartbeat.MessageModel;
import org.idea.fig.MqConsumerConfig;
import org.idea.fig.MqProducerConfig;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
豫公网安备41160202000603
站长QQ:729038198
关于我们
投诉建议