大家好，关于影视服务器源码分享下载网站很多朋友都还不太明白，今天小编就来为大家分享关于影视服务器搭建教程的知识，希望对各位有所帮助！
项目背景
和各位读者大致介绍下具体场景，线上的小程序中开放一些语音麦克风的房间，让用户进入房间之后可以互相通过语音聊天的方式进行互动。
这里分享一下相关的技术设计方案。这款系统的核心点设计在于如何能让一个用户发出的语音通知到其他用户上边。语音数据在客户端同事的处理下最终变成了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版约定以下概念：
集群：使用相同GroupID的消费者属于同一个集群。同一个集群下的消费者消费逻辑必须完全一致（包括Tag的使用）。
集群消费：当使用集群消费模式时，消息队列RocketMQ版认为任意一条消息只需要被集群内的任意一个消费者处理即可。
广播消费：当使用广播消费模式时，消息队列RocketMQ版会将每条消息推送给集群内所有注册过的消费者，保证消息至少被每个消费者消费一次。
集群消费模式适用场景适用于消费端集群化部署，每条消息只需要被处理一次的场景。此外，由于消费进度在服务端维护，可靠性更高。具体消费示例如下图所示。
注意事项
集群消费模式下，每一条消息都只会被分发到一台机器上处理。如果需要被集群下的每一台机器都处理，请使用广播模式。
集群消费模式下，不保证每一次失败重投的消息路由到同一台机器上。
广播消费模式适用场景适用于消费端集群化部署，每条消息需要被集群下的每个消费者处理的场景。具体消费示例如下图所示。
注意事项
广播消费模式下不支持顺序消息。
广播消费模式下不支持重置消费位点。
每条消息都需要被相同订阅逻辑的多台机器处理。
消费进度在客户端维护，出现重复消费的概率稍大于集群模式。
广播模式下，消息队列RocketMQ版保证每条消息至少被每台客户端消费一次，但是并不会重投消费失败的消息，因此业务方需要关注消费失败的情况。
广播模式下，客户端每一次重启都会从最新消息消费。客户端在被停止期间发送至服务端的消息将会被自动跳过，请谨慎选择。
广播模式下，每条消息都会被大量的客户端重复处理，因此推荐尽可能使用集群模式。
广播模式下服务端不维护消费进度，所以消息队列RocketMQ版控制台不支持消息堆积查询、消息堆积报警和订阅关系查询功能。
这里面的应用场景需要对集群内部对每个消费者都对服务器内存中的socket连接进行session是否存在对判断，因此需要采用mq的广播模式。
关于mq部分的接入代码
Consumer模块的配置：
packageorg.idea.web.socket.config;
importorg.springframework.boot.context.properties.ConfigurationProperties;
/**
*@Authorlinhao
*@Datecreatedin10:30上午2021/5/10
*/
@ConfigurationProperties(prefix=&34;)
publicclassMqConsumerConfig{
privatebooleanisOn;
privateStringgroupName;
privateStringnameSrvAddr;
privateStringtopics;
privateIntegerconsumeThreadMin;
privateIntegerconsumeThreadMax;
privateIntegerconsumeMessageBatchMaxSize;
/**
getter和setter部分省略
**/
}
Producer模块的配置展示：
packageorg.idea.web.socket.config;
importorg.springframework.boot.context.properties.ConfigurationProperties;
/**
*@Authorlinhao
*@Datecreatedin10:26上午2021/5/10
*/
@ConfigurationProperties(prefix=&34;)
publicclassMqProducerConfig{
privatebooleanisOn;
privateStringgroupName;
privateStringnameSrvAddr;
privateIntegermaxMessageSize;
privateIntegersendMsgTimeout;
privateIntegerretryTimesWhenSendFailed;
/**
getter和setter部分省略
**/
}
RocketMq内部的消费端Bean配置
packageorg.idea.web.socket.mq;
importlombok.extern.slf4j.Slf4j;
importorg.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
importorg.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
importorg.apache.rocketmq.client.exception.MQClientException;
importorg.apache.rocketmq.common.consumer.ConsumeFromWhere;
importorg.apache.rocketmq.common.protocol.heartbeat.MessageModel;
importorg.idea.web.socket.config.MqConsumerConfig;
importorg.idea.web.socket.config.MqProducerConfig;
importorg.springframework.boot.autoconfigure.AutoConfigureAfter;
importorg.springframework.boot.autoconfigure.AutoConfigureBefore;
importorg.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
importorg.springframework.boot.context.properties.EnableConfigurationProperties;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Configuration;
importjavax.annotation.Resource;
/**
*@Authorlinhao
*@Datecreatedin10:34上午2021/5/10
*/
@Configuration
@Slf4j
@EnableConfigurationProperties({MqConsumerConfig.class})
publicclassMqConsumerAutoConfig{
@Resource
privateMqConsumerConfigmqConsumerConfig;
@Resource
//这个接口需要手动实现顺序消费的逻辑每次获取到消息队列的第一条数据
privateMessageListenerHandlermessageListenerConcurrently;
@Bean
@ConditionalOnMissingBean
publicDefaultMQPushConsumerdefaultMQPushConsumer(){
DefaultMQPushConsumerconsumer=newDefaultMQPushConsumer();
consumer.setNamesrvAddr(mqConsumerConfig.getNameSrvAddr());
consumer.setConsumerGroup(mqConsumerConfig.getGroupName());
consumer.setConsumeThreadMin(mqConsumerConfig.getConsumeThreadMin());
consumer.setConsumeThreadMax(mqConsumerConfig.getConsumeThreadMax());
consumer.registerMessageListener(messageListenerConcurrently);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//消费模型是什么？
consumer.setMessageModel(MessageModel.BROADCASTING);
//默认一次拉取一条消费
consumer.setConsumeMessageBatchMaxSize(mqConsumerConfig.getConsumeMessageBatchMaxSize());
//*表示订阅所有的tag
try{
consumer.subscribe(mqConsumerConfig.getTopics(),&34;);
consumer.start();
log.info(&34;);
}catch(Exceptione){
log.error(&34;,e);
}
returnconsumer;
}
}
RocketMq的服务生产者Bean配置
packageorg.idea.web.socket.mq;
importlombok.extern.slf4j.Slf4j;
importorg.apache.rocketmq.client.producer.DefaultMQProducer;
importorg.idea.web.socket.config.MqProducerConfig;
importorg.springframework.boot.autoconfigure.AutoConfigureAfter;
importorg.springframework.boot.autoconfigure.AutoConfigureBefore;
importorg.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
importorg.springframework.boot.context.properties.EnableConfigurationProperties;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Configuration;
importjavax.annotation.Resource;
/**
*@Authorlinhao
*@Datecreatedin11:05上午2021/5/10
*/
@Configuration
@Slf4j
@EnableConfigurationProperties({MqProducerConfig.class})
publicclassMqProducerAutoConfig{
@Resource
privateMqProducerConfigmqProducerConfig;
@Bean
@ConditionalOnMissingBean
//意味着DefaultMQProducer的配置可以被覆盖
publicDefaultMQProducerdefaultMQProducer(){
DefaultMQProducerproducer=newDefaultMQProducer(mqProducerConfig.getGroupName());
producer.setNamesrvAddr(mqProducerConfig.getNameSrvAddr());
//没有则自动创建topic的key
//producer.setCreateTopicKey(&34;);
producer.setMaxMessageSize(mqProducerConfig.getMaxMessageSize());
producer.setSendMsgTimeout(mqProducerConfig.getSendMsgTimeout());
producer.setRetryTimesWhenSendFailed(mqProducerConfig.getRetryTimesWhenSendFailed());
try{
producer.start();
log.info(&34;);
}catch(Exceptione){
log.error(&34;,e);
}
returnproducer;
}
}
然后是对RocketMq内部发送消息事件的一层函数封装
packageorg.idea.web.socket.mq;
importcom.alibaba.fastjson.JSON;
importlombok.extern.slf4j.Slf4j;
importorg.apache.commons.lang3.StringUtils;
importorg.apache.rocketmq.client.producer.DefaultMQProducer;
importorg.apache.rocketmq.client.producer.SendResult;
importorg.apache.rocketmq.common.message.Message;
importorg.apache.rocketmq.remoting.common.RemotingHelper;
importorg.idea.web.socket.config.MqProducerConfig;
importorg.idea.web.socket.dto.BroadcastMqDTO;
importorg.springframework.stereotype.Component;
importjavax.annotation.Resource;
importjava.io.UnsupportedEncodingException;
/**
*消息广播发送端
*
*@Authorlinhao
*@Datecreatedin10:43下午2021/5/9
*/
@Component
@Slf4j
publicclassBroadcastMqProducer{
@Resource
privateDefaultMQProducerdefaultMQProducer;
@Resource
privateMqProducerConfigmqProducerConfig;
privatestaticStringTOPIC=&34;;
privatestaticStringTAGS=&34;;
publicstaticIntegerALL_USER_RECEIVE_TYPE=1;
publicstaticIntegerONE_USER_RECEIVE_TYPE=2;
/**
*点对点之间的消息发送
*
*@paramdestSessionKey
*@parammsg
*@return
*/
publicSendResultsendWebSocketToUser(StringdestSessionKey,Stringmsg){
if(StringUtils.isEmpty(msg)){
log.error(&34;);
returnnull;
}
Messagemessage=null;
SendResultsendResult=null;
try{
BroadcastMqDTObroadcastMqDTO=newBroadcastMqDTO();
broadcastMqDTO.setEventType(ONE_USER_RECEIVE_TYPE);
broadcastMqDTO.setMessage(msg);
broadcastMqDTO.setSessionKey(destSessionKey);
message=newMessage(TOPIC,TAGS,(JSON.toJSONString(broadcastMqDTO)).getBytes(RemotingHelper.DEFAULT_CHARSET));
sendResult=defaultMQProducer.send(message);
}catch(Exceptione){
log.error(&34;,e);
}
returnsendResult;
}
/**
*广播消息发送
*
*@parammsg
*@return
*/
publicSendResultsendWebSocketBroadcastMsg(Stringmsg){
if(StringUtils.isEmpty(msg)){
log.error(&34;);
returnnull;
}
Messagemessage=null;
SendResultsendResult=null;
try{
BroadcastMqDTObroadcastMqDTO=newBroadcastMqDTO();
broadcastMqDTO.setEventType(ALL_USER_RECEIVE_TYPE);
broadcastMqDTO.setMessage(msg);
message=newMessage(TOPIC,TAGS,(JSON.toJSONString(broadcastMqDTO)).getBytes(RemotingHelper.DEFAULT_CHARSET));
sendResult=defaultMQProducer.send(message);
}catch(Exceptione){
log.error(&34;,e);
}
returnsendResult;
}
}
对消息的订阅模块实现代码如下：
packageorg.idea.web.socket.mq;
importcom.alibaba.fastjson.JSON;
importcom.oracle.tools.packager.Log;
importlombok.extern.slf4j.Slf4j;
importorg.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
importorg.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
importorg.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
importorg.apache.rocketmq.common.message.MessageExt;
importorg.idea.web.socket.dto.BroadcastMqDTO;
importorg.idea.web.socket.manager.SocketManager;
importorg.springframework.messaging.simp.SimpMessagingTemplate;
importorg.springframework.stereotype.Component;
importorg.springframework.util.CollectionUtils;
importorg.springframework.web.socket.WebSocketSession;
importjavax.annotation.Resource;
importjava.util.List;
importstaticorg.idea.web.socket.mq.BroadcastMqProducer.ALL_USER_RECEIVE_TYPE;
importstaticorg.idea.web.socket.mq.BroadcastMqProducer.ONE_USER_RECEIVE_TYPE;
/**
*@Authorlinhao
*@Datecreatedin10:59上午2021/5/10
*/
@Component
@Slf4j
publicclassMessageListenerHandlerimplementsMessageListenerConcurrently{
@Resource
privateSocketManagersocketManager;
@Resource
privateSimpMessagingTemplatetemplate;
@Override
publicConsumeConcurrentlyStatusconsumeMessage(List<MessageExt>list,ConsumeConcurrentlyContextconsumeConcurrentlyContext){
if(CollectionUtils.isEmpty(list)){
Log.info(&34;);
returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
MessageExtmessageExt=list.get(0);
byte[]bytes=messageExt.getBody();
Stringjson=newString(bytes);
BroadcastMqDTObroadcastMqDTO=JSON.parseObject(json,BroadcastMqDTO.class);
log.info(&34;+broadcastMqDTO);
if(ALL_USER_RECEIVE_TYPE.equals(broadcastMqDTO.getEventType())){
log.info(&34;+broadcastMqDTO);
template.convertAndSend(&34;,broadcastMqDTO);
}elseif(ONE_USER_RECEIVE_TYPE.equals(broadcastMqDTO.getEventType())){
StringsessionKey=broadcastMqDTO.getSessionKey();
WebSocketSessionwebSocketSession=socketManager.get(sessionKey);
if(webSocketSession!=null){
template.convertAndSendToUser(sessionKey,&34;,broadcastMqDTO.getMessage());
log.info(&34;+broadcastMqDTO);
}
}
returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
整体设计结构如下图：
于是按照这个结构进行了一版本的紧急开发迭代，原先的单台服务器扩展为了服务集群。
业务拓展后续产品经理提出一个需求，要求支持在同一间房内的两个用户之间发送悄悄话功能。这就需要我们进行一个点对点之间传输通讯的功能了。因此需要在mq通知到每台机器的时候加一个本地Session遍历的逻辑，如果当前机器存有用户token对应的session变量，那么就单独针对那个Session进行WebSocket的发送通知。
设计弊端一旦某台机器出现了异常崩溃，那么就意味着这台机器上的所有语音连接可能会出现中断情况。目前这一块的问题也在考虑解决，计划是将WebSocketSession存入到分布式缓存的redis中保证数据可靠存储，但是在后续尝试的时候发现WebSocketSession对象没有实现序列化接口，在存储到Redis的时候会出现异常。目前这个问题还在寻找解决思路中，不知道各位读者朋友们有什么好的思路。
好了，文章到这里就结束啦，如果本次分享的影视服务器源码分享下载网站和影视服务器搭建教程问题对您有所帮助，还望关注下本站哦！