教你用純Java實現一個即時通訊系統（附源碼）

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

開發后端

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

項目背景

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

這里分享一下相關的技術設計方案。這款系統的核心點設計在于如何能讓一個用戶發出的語音通知到其他用戶上邊。語音數據在客戶端同事的處理下最終變成了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.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內部的消費端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  
    //這個接口需要手動實現順序消費的邏輯 每次獲取到消息隊列的第一條數據  
    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);  
        //消費模型是什么？  
        consumer.setMessageModel(MessageModel.BROADCASTING);  
        //默認一次拉取一條消費  
        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的服務生產者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());  
        //沒有則自動創建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內部發送消息事件的一層函數封裝

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;  
/**  
 * 消息廣播發送端  
 *  
 * @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;  
    /**  
     * 點對點之間的消息發送  
     *  
     * @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;  
    }  
    /**  
     * 廣播消息發送  
     *  
     * @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;  
    }  
}

對消息的訂閱模塊實現代碼如下：

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] 廣播發送消息：觸發----》消息內容為：" + 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] 點對點發送消息；觸發----》消息內容為：" + broadcastMqDTO);  
            }  
        }  
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
    }  
}

整體設計結構如下圖：

于是按照這個結構進行了一版本的緊急開發迭代，原先的單臺服務器擴展為了服務集群。

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

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

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

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

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

責任編輯：龐桂玉來源： Java知音

Java 架構代碼

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

教你用純Java實現一個即時通訊系統（附源碼）

項目背景

單機版架構

集群架構

注意事項

注意事項

關于mq部分的接入代碼