教你用純Java實現(xiàn)一個即時通訊系統(tǒng)（附源碼）

作者：佚名 2021-05-13 20:20:40

和各位讀者大致介紹下具體場景，線上的小程序中開放一些語音麥克風(fēng)的房間，讓用戶進(jìn)入房間之后可以互相通過語音聊天的方式進(jìn)行互動。

項目背景

這里分享一下相關(guān)的技術(shù)設(shè)計方案。這款系統(tǒng)的核心點(diǎn)設(shè)計在于如何能讓一個用戶發(fā)出的語音通知到其他用戶上邊。語音數(shù)據(jù)在客戶端同事的處理下最終變成了io數(shù)據(jù)流請求到了后端，后端只需要將這些數(shù)據(jù)流傳達(dá)給各個不同的終端即可達(dá)到廣播通知的效果。

單機(jī)版架構(gòu)

最初期上線的時候，為了趕速度，快速試錯，所以簡單地采用了單機(jī)版架構(gòu)去設(shè)計。結(jié)合技術(shù)棧為 SpringBoot，WebSocket，MySQL技術(shù)。

線上一間語音房間的同時在線人數(shù)并不會特別多，大概在15-50人的區(qū)間段內(nèi)，系統(tǒng)核心代碼是通過SpringBoot內(nèi)部的WebSocket技術(shù)去進(jìn)行數(shù)據(jù)的主動推送。

設(shè)計思路

整體的設(shè)計圖比較簡單，基本就是一臺服務(wù)器存儲WebSocket連接，如下圖所示：

用戶進(jìn)行WebSocket初始化連接的時候需要一個連接分配和存儲的過程：

早期的存儲是存放在了服務(wù)器本地的一個Map集合中。

當(dāng)WebSocket進(jìn)行連接的時候就會往內(nèi)存中寫入一條數(shù)據(jù)信息，當(dāng)鏈接斷開的時候，就將內(nèi)存中的數(shù)據(jù)移除。然后進(jìn)行語音廣播的時候需要結(jié)合WebSocket內(nèi)部的廣播發(fā)送功能進(jìn)行通知

看似設(shè)計比較簡單，但是在后期業(yè)務(wù)變得龐大的時候出現(xiàn)了瓶頸。因為隨著參加語音活動用戶的增加，越來越多的WebSocketSession對象需要被存儲到內(nèi)存當(dāng)中，這種有狀態(tài)性的存儲對于單機(jī)擴(kuò)容不靈活。

設(shè)計缺陷

1.假設(shè)原先的服務(wù)器擴(kuò)容到了A，B兩臺機(jī)器，A用戶在A機(jī)器上邊建立了WebSocketSession，B用戶在B機(jī)器上邊建立的WebSocketSession連接。此時如果A想要和B進(jìn)行對話發(fā)送，需要先查找到具體WebSocketSession存放在哪臺機(jī)器上邊。

2.當(dāng)用戶出現(xiàn)了網(wǎng)絡(luò)異常，臨時斷開連接進(jìn)行重連的時候，也可能會出現(xiàn)1所說的問題。

集群架構(gòu)

設(shè)計思路

一旦出現(xiàn)需要發(fā)送語音通知的時候，發(fā)送一條廣播的mq消息，每個機(jī)器都接收到消息之后，觸發(fā)自己的廣播操作即可。

RocketMq的接入系統(tǒng)設(shè)計里面mq采用的是廣播模式，這和我們通常使用的集群模式有一定的區(qū)別。

消息隊列RocketMQ版是基于發(fā)布或訂閱模型的消息系統(tǒng)。消費(fèi)者，即消息的訂閱方訂閱關(guān)注的Topic，以獲取并消費(fèi)消息。由于消費(fèi)者應(yīng)用一般是分布式系統(tǒng)，以集群方式部署，因此消息隊列RocketMQ版約定以下概念：

集群：使用相同Group ID的消費(fèi)者屬于同一個集群。同一個集群下的消費(fèi)者消費(fèi)邏輯必須完全一致（包括Tag的使用）。
集群消費(fèi)：當(dāng)使用集群消費(fèi)模式時，消息隊列RocketMQ版認(rèn)為任意一條消息只需要被集群內(nèi)的任意一個消費(fèi)者處理即可。
廣播消費(fèi)：當(dāng)使用廣播消費(fèi)模式時，消息隊列RocketMQ版會將每條消息推送給集群內(nèi)所有注冊過的消費(fèi)者，保證消息至少被每個消費(fèi)者消費(fèi)一次。

集群消費(fèi)模式適用場景適用于消費(fèi)端集群化部署，每條消息只需要被處理一次的場景。此外，由于消費(fèi)進(jìn)度在服務(wù)端維護(hù)，可靠性更高。具體消費(fèi)示例如下圖所示。

注意事項

集群消費(fèi)模式下，每一條消息都只會被分發(fā)到一臺機(jī)器上處理。如果需要被集群下的每一臺機(jī)器都處理，請使用廣播模式。
集群消費(fèi)模式下，不保證每一次失敗重投的消息路由到同一臺機(jī)器上。

廣播消費(fèi)模式適用場景適用于消費(fèi)端集群化部署，每條消息需要被集群下的每個消費(fèi)者處理的場景。具體消費(fèi)示例如下圖所示。

注意事項

廣播消費(fèi)模式下不支持順序消息。
廣播消費(fèi)模式下不支持重置消費(fèi)位點(diǎn)。
每條消息都需要被相同訂閱邏輯的多臺機(jī)器處理。
消費(fèi)進(jìn)度在客戶端維護(hù)，出現(xiàn)重復(fù)消費(fèi)的概率稍大于集群模式。
廣播模式下，消息隊列RocketMQ版保證每條消息至少被每臺客戶端消費(fèi)一次，但是并不會重投消費(fèi)失敗的消息，因此業(yè)務(wù)方需要關(guān)注消費(fèi)失敗的情況。
廣播模式下，客戶端每一次重啟都會從最新消息消費(fèi)。客戶端在被停止期間發(fā)送至服務(wù)端的消息將會被自動跳過，請謹(jǐn)慎選擇。
廣播模式下，每條消息都會被大量的客戶端重復(fù)處理，因此推薦盡可能使用集群模式。
廣播模式下服務(wù)端不維護(hù)消費(fèi)進(jìn)度，所以消息隊列RocketMQ版控制臺不支持消息堆積查詢、消息堆積報警和訂閱關(guān)系查詢功能。

這里面的應(yīng)用場景需要對集群內(nèi)部對每個消費(fèi)者都對服務(wù)器內(nèi)存中的socket連接進(jìn)行session是否存在對判斷，因此需要采用mq的廣播模式。

關(guān)于mq部分的接入代碼

Consumer模塊的配置：

package org.idea.web.socket.config;  
import org.springframework.boot.context.properties.ConfigurationProperties;  
/**  
 * @Author linhao  
 * @Date created in 10:30 上午 2021/5/10  
 */  
@ConfigurationProperties(prefix = "rocketmq.consumer")  
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.web.socket.config;  
import org.springframework.boot.context.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內(nèi)部的消費(fèi)端Bean配置

package org.idea.web.socket.mq;  
import lombok.extern.slf4j.Slf4j;  
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;  
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;  
import org.apache.rocketmq.client.exception.MQClientException;  
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;  
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;  
import org.idea.web.socket.config.MqConsumerConfig;  
import org.idea.web.socket.config.MqProducerConfig;  
import org.springframework.boot.autoconfigure.AutoConfigureAfter;  
import org.springframework.boot.autoconfigure.AutoConfigureBefore;  
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;  
import org.springframework.boot.context.properties.EnableConfigurationProperties;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;  
import javax.annotation.Resource;  
/**  
 * @Author linhao  
 * @Date created in 10:34 上午 2021/5/10  
 */  
@Configuration  
@Slf4j  
@EnableConfigurationProperties({MqConsumerConfig.class})  
public class MqConsumerAutoConfig {  
    @Resource  
    private MqConsumerConfig mqConsumerConfig;  
    @Resource  
    //這個接口需要手動實現(xiàn)順序消費(fèi)的邏輯 每次獲取到消息隊列的第一條數(shù)據(jù)  
    private MessageListenerHandler messageListenerConcurrently;  
    @Bean  
    @ConditionalOnMissingBean  
    public DefaultMQPushConsumer defaultMQPushConsumer() {  
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();  
        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);  
        //消費(fèi)模型是什么？  
        consumer.setMessageModel(MessageModel.BROADCASTING);  
        //默認(rèn)一次拉取一條消費(fèi)  
        consumer.setConsumeMessageBatchMaxSize(mqConsumerConfig.getConsumeMessageBatchMaxSize());  
        //*表示訂閱所有的tag  
        try {  
            consumer.subscribe(mqConsumerConfig.getTopics(), "*");  
            consumer.start();  
            log.info("【 MqConsumerAutoConfig 】mq consumer is started!");  
        } catch (Exception e) {  
            log.error("mq start fail,e is ", e);  
        }  
        return consumer;  
    }  
}

RocketMq的服務(wù)生產(chǎn)者Bean配置

package org.idea.web.socket.mq;  
import lombok.extern.slf4j.Slf4j;  
import org.apache.rocketmq.client.producer.DefaultMQProducer;  
import org.idea.web.socket.config.MqProducerConfig;  
import org.springframework.boot.autoconfigure.AutoConfigureAfter;  
import org.springframework.boot.autoconfigure.AutoConfigureBefore;  
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;  
import org.springframework.boot.context.properties.EnableConfigurationProperties;  
import org.springframework.context.annotation.Bean;  
import org.springframework.context.annotation.Configuration;  
import javax.annotation.Resource;  
/**  
 * @Author linhao  
 * @Date created in 11:05 上午 2021/5/10  
 */  
@Configuration  
@Slf4j  
@EnableConfigurationProperties({MqProducerConfig.class})  
public class MqProducerAutoConfig {  
    @Resource  
    private MqProducerConfig mqProducerConfig;  
    @Bean  
    @ConditionalOnMissingBean  
    //意味著DefaultMQProducer的配置可以被覆蓋  
    public DefaultMQProducer defaultMQProducer() {  
        DefaultMQProducer producer = new DefaultMQProducer(mqProducerConfig.getGroupName());  
        producer.setNamesrvAddr(mqProducerConfig.getNameSrvAddr());  
        //沒有則自動創(chuàng)建topic的key 
 //        producer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY");  
        producer.setMaxMessageSize(mqProducerConfig.getMaxMessageSize()); 
        producer.setSendMsgTimeout(mqProducerConfig.getSendMsgTimeout());  
        producer.setRetryTimesWhenSendFailed(mqProducerConfig.getRetryTimesWhenSendFailed());  
        try {  
            producer.start();  
            log.info("【 MqProducerAutoConfig 】mq producer is started!");  
        } catch (Exception e) {  
            log.error("[MqProducerAutoConfig] start fail, e is ", e);  
        }  
        return producer;  
    }  
}

然后是對RocketMq內(nèi)部發(fā)送消息事件的一層函數(shù)封裝

package org.idea.web.socket.mq;  
import com.alibaba.fastjson.JSON;  
import lombok.extern.slf4j.Slf4j;  
import org.apache.commons.lang3.StringUtils;  
import org.apache.rocketmq.client.producer.DefaultMQProducer;  
import org.apache.rocketmq.client.producer.SendResult;  
import org.apache.rocketmq.common.message.Message;  
import org.apache.rocketmq.remoting.common.RemotingHelper;  
import org.idea.web.socket.config.MqProducerConfig;  
import org.idea.web.socket.dto.BroadcastMqDTO;  
import org.springframework.stereotype.Component;  
import javax.annotation.Resource;  
import java.io.UnsupportedEncodingException;  
/**  
 * 消息廣播發(fā)送端  
 *  
 * @Author linhao  
 * @Date created in 10:43 下午 2021/5/9  
 */  
@Component  
@Slf4j  
public class BroadcastMqProducer {  
    @Resource  
    private DefaultMQProducer defaultMQProducer;  
    @Resource  
    private MqProducerConfig mqProducerConfig;  
    private static String TOPIC = "ws-topic";  
    private static String TAGS = "ws-tag";  
    public static Integer ALL_USER_RECEIVE_TYPE = 1;  
    public static Integer ONE_USER_RECEIVE_TYPE = 2;  
    /**  
     * 點(diǎn)對點(diǎn)之間的消息發(fā)送  
     *  
     * @param destSessionKey  
     * @param msg  
     * @return  
     */  
    public SendResult sendWebSocketToUser(String destSessionKey,String msg) {  
        if (StringUtils.isEmpty(msg)) {  
            log.error("[sendWebSocketToUser] msg can not be null!");  
            return null;  
        }  
        Message message = null;  
        SendResult sendResult = null;  
        try {  
            BroadcastMqDTO broadcastMqDTO = new BroadcastMqDTO();  
            broadcastMqDTO.setEventType(ONE_USER_RECEIVE_TYPE);  
            broadcastMqDTO.setMessage(msg);  
            broadcastMqDTO.setSessionKey(destSessionKey);  
            message = new Message(TOPIC, TAGS, (JSON.toJSONString(broadcastMqDTO)).getBytes(RemotingHelper.DEFAULT_CHARSET));  
            sendResult = defaultMQProducer.send(message);  
        } catch (Exception e) { 
             log.error("[sendWebSocketBroadcastMsg] e is ", e);  
        }  
        return sendResult;  
    }  
    /**  
     * 廣播消息發(fā)送  
     *  
     * @param msg 
     * @return  
     */  
    public SendResult sendWebSocketBroadcastMsg(String msg) {  
        if (StringUtils.isEmpty(msg)) {  
            log.error("[sendWebSocketBroadcastMsg] msg can not be null!");  
            return null;  
        }  
        Message message = null;  
        SendResult sendResult = null;  
        try {  
            BroadcastMqDTO broadcastMqDTO = new BroadcastMqDTO();  
            broadcastMqDTO.setEventType(ALL_USER_RECEIVE_TYPE);  
            broadcastMqDTO.setMessage(msg);  
            message = new Message(TOPIC, TAGS, (JSON.toJSONString(broadcastMqDTO)).getBytes(RemotingHelper.DEFAULT_CHARSET));  
            sendResult = defaultMQProducer.send(message);  
        } catch (Exception e) {  
            log.error("[sendWebSocketBroadcastMsg] e is ", e);  
        }  
        return sendResult;  
    }  
}

對消息的訂閱模塊實現(xiàn)代碼如下：

package org.idea.web.socket.mq;  
import com.alibaba.fastjson.JSON;  
import com.oracle.tools.packager.Log;  
import lombok.extern.slf4j.Slf4j;  
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;  
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;  
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;  
import org.apache.rocketmq.common.message.MessageExt;  
import org.idea.web.socket.dto.BroadcastMqDTO;  
import org.idea.web.socket.manager.SocketManager;  
import org.springframework.messaging.simp.SimpMessagingTemplate;  
import org.springframework.stereotype.Component;  
import org.springframework.util.CollectionUtils;  
import org.springframework.web.socket.WebSocketSession;  
import javax.annotation.Resource;  
import java.util.List;  
import static org.idea.web.socket.mq.BroadcastMqProducer.ALL_USER_RECEIVE_TYPE;  
import static org.idea.web.socket.mq.BroadcastMqProducer.ONE_USER_RECEIVE_TYPE;  
/**  
 * @Author linhao  
 * @Date created in 10:59 上午 2021/5/10  
 */  
@Component  
@Slf4j  
public class MessageListenerHandler implements MessageListenerConcurrently {  
    @Resource  
    private SocketManager socketManager;  
    @Resource  
    private SimpMessagingTemplate template;  
    @Override  
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {  
        if (CollectionUtils.isEmpty(list)) {  
            Log.info("receive empty msg");  
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
        }  
        MessageExt messageExt = list.get(0);  
        byte[] bytes = messageExt.getBody();  
        String json = new String(bytes);  
        BroadcastMqDTO broadcastMqDTO = JSON.parseObject(json, BroadcastMqDTO.class);  
        log.info("[MessageListenerHandler] broadcastMqDTO is " + broadcastMqDTO);  
        if (ALL_USER_RECEIVE_TYPE.equals(broadcastMqDTO.getEventType())) {  
            log.info("[consumeMessage] 廣播發(fā)送消息：觸發(fā)----》消息內(nèi)容為：" + broadcastMqDTO);  
            template.convertAndSend("/topic/sendTopic", broadcastMqDTO);  
        } else if (ONE_USER_RECEIVE_TYPE.equals(broadcastMqDTO.getEventType())) {  
            String sessionKey = broadcastMqDTO.getSessionKey();  
            WebSocketSession webSocketSession = socketManager.get(sessionKey);  
            if (webSocketSession != null) {  
                template.convertAndSendToUser(sessionKey, "/queue/sendUser", broadcastMqDTO.getMessage());  
                log.info("[consumeMessage] 點(diǎn)對點(diǎn)發(fā)送消息；觸發(fā)----》消息內(nèi)容為：" + broadcastMqDTO);  
            }  
        }  
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
    }  
}

整體設(shè)計結(jié)構(gòu)如下圖：

于是按照這個結(jié)構(gòu)進(jìn)行了一版本的緊急開發(fā)迭代，原先的單臺服務(wù)器擴(kuò)展為了服務(wù)集群。

業(yè)務(wù)拓展后續(xù)產(chǎn)品經(jīng)理提出一個需求，要求支持在同一間房內(nèi)的兩個用戶之間發(fā)送悄悄話功能。這就需要我們進(jìn)行一個點(diǎn)對點(diǎn)之間傳輸通訊的功能了。因此需要在mq通知到每臺機(jī)器的時候加一個本地Session遍歷的邏輯，如果當(dāng)前機(jī)器存有用戶token對應(yīng)的session變量，那么就單獨(dú)針對那個Session進(jìn)行WebSocket的發(fā)送通知。

設(shè)計弊端一旦某臺機(jī)器出現(xiàn)了異常崩潰，那么就意味著這臺機(jī)器上的所有語音連接可能會出現(xiàn)中斷情況。目前這一塊的問題也在考慮解決，計劃是將WebSocketSession存入到分布式緩存的redis中保證數(shù)據(jù)可靠存儲，但是在后續(xù)嘗試的時候發(fā)現(xiàn)WebSocketSession對象沒有實現(xiàn)序列化接口，在存儲到Redis的時候會出現(xiàn)異常。目前這個問題還在尋找解決思路中，不知道各位讀者朋友們有什么好的思路。

遇到的問題點(diǎn)用戶請求直接訪問到了我們的內(nèi)部服務(wù)器，如果在請求的中間加入一臺nginx做負(fù)載均衡則需要在nginx中配置一些額外信息。

項目的源代碼比較多，這里我把核心部分的代碼整理了一份，感興趣的朋友可以到我的gitee上邊去下載：

https://gitee.com/IdeaHome_admin/socket-framework

責(zé)任編輯：龐桂玉來源： Java知音