Netty+SpringBoot+FastDFS+Html5實現聊天App(五)

Netty+SpringBoot+FastDFS+Html5實現聊天App，項目介紹。

Netty+SpringBoot+FastDFS+Html5實現聊天App，項目github連接。

本章完整代碼連接。

本章主要講的是聊天App_PigChat中關於聊天功能的實現。

移除方法與處理異常方法的重寫

在ChatHandler中重寫其移除channel的方法handlerRemoved，以及處理異常的方法exceptionCaught。

@Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        
        String channelId = ctx.channel().id().asShortText();
        System.out.println("客戶端被移除，channelId爲：" + channelId);
        
        // 當觸發handlerRemoved，ChannelGroup會自動移除對應客戶端的channel
        users.remove(ctx.channel());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        // 發生異常以後關閉鏈接（關閉channel），隨後從ChannelGroup中移除
        ctx.channel().close();
        users.remove(ctx.channel());
    }

定義消息的實體類

public class ChatMsg implements Serializable {

    private static final long serialVersionUID = 3611169682695799175L;
    
    private String senderId;        // 發送者的用戶id    
    private String receiverId;        // 接受者的用戶id
    private String msg;                // 聊天內容
    private String msgId;            // 用於消息的簽收
    
    public String getSenderId() {
        return senderId;
    }
    public void setSenderId(String senderId) {
        this.senderId = senderId;
    }
    public String getReceiverId() {
        return receiverId;
    }
    public void setReceiverId(String receiverId) {
        this.receiverId = receiverId;
    }
    public String getMsg() {
        return msg;
    }
    public void setMsg(String msg) {
        this.msg = msg;
    }
    public String getMsgId() {
        return msgId;
    }
    public void setMsgId(String msgId) {
        this.msgId = msgId;
    }
    
}

對實體類再作一層包裝

public class DataContent implements Serializable {

    private static final long serialVersionUID = 8021381444738260454L;

    private Integer action;        // 動做類型
    private ChatMsg chatMsg;    // 用戶的聊天內容entity
    private String extand;        // 擴展字段
    
    public Integer getAction() {
        return action;
    }
    public void setAction(Integer action) {
        this.action = action;
    }
    public ChatMsg getChatMsg() {
        return chatMsg;
    }
    public void setChatMsg(ChatMsg chatMsg) {
        this.chatMsg = chatMsg;
    }
    public String getExtand() {
        return extand;
    }
    public void setExtand(String extand) {
        this.extand = extand;
    }
}

定義發送消息的動做的枚舉類型

public enum MsgActionEnum {
    
    CONNECT(1, "第一次(或重連)初始化鏈接"),
    CHAT(2, "聊天消息"),    
    SIGNED(3, "消息簽收"),
    KEEPALIVE(4, "客戶端保持心跳"),
    PULL_FRIEND(5, "拉取好友");
    
    public final Integer type;
    public final String content;
    
    MsgActionEnum(Integer type, String content){
        this.type = type;
        this.content = content;
    }
    
    public Integer getType() {
        return type;
    }  
}

定義記錄用戶與channel關係的類

/**
 * @Description: 用戶id和channel的關聯關係處理
 */
public class UserChannelRel {

    private static HashMap<String, Channel> manager = new HashMap<>();

    public static void put(String senderId, Channel channel) {
        manager.put(senderId, channel);
    }
    
    public static Channel get(String senderId) {
        return manager.get(senderId);
    }
    
    public static void output() {
        for (HashMap.Entry<String, Channel> entry : manager.entrySet()) {
            System.out.println("UserId: " + entry.getKey() 
                            + ", ChannelId: " + entry.getValue().id().asLongText());
        }
    }
}

接受與處理消息方法的重寫

重寫ChatHandler讀取消息的channelRead0方法。

具體步驟以下：

（1）獲取客戶端發來的消息；

（2）判斷消息類型，根據不一樣的類型來處理不一樣的業務；

（2.1）當websocket 第一次open的時候，初始化channel，把用的channel和userid關聯起來；

（2.2）聊天類型的消息，把聊天記錄保存到數據庫，同時標記消息的簽收狀態[未簽收]；
而後實現消息的發送，首先從全局用戶Channel關係中獲取接受方的channel，而後當receiverChannel不爲空的時候，從ChannelGroup去查找對應的channel是否存在，若用戶在線，則使用writeAndFlush方法向其發送消息；

（2.3）簽收消息類型，針對具體的消息進行簽收，修改數據庫中對應消息的簽收狀態[已簽收]；

（2.4）心跳類型的消息

// 用於記錄和管理全部客戶端的channle
    public static ChannelGroup users = 
            new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) 
            throws Exception {
        System.out.println("read..........");
        // 獲取客戶端傳輸過來的消息
        String content = msg.text();
        
        Channel currentChannel = ctx.channel();

        // 1. 獲取客戶端發來的消息
        DataContent dataContent = JsonUtils.jsonToPojo(content, DataContent.class);
        Integer action = dataContent.getAction();
        // 2. 判斷消息類型，根據不一樣的類型來處理不一樣的業務

        if (action == MsgActionEnum.CONNECT.type) {
            //     2.1  當websocket 第一次open的時候，初始化channel，把用的channel和userid關聯起來
            String senderId = dataContent.getChatMsg().getSenderId();
            UserChannelRel.put(senderId, currentChannel);
            
            // 測試
            for (Channel c : users) {
                System.out.println(c.id().asLongText());
            }
            UserChannelRel.output();
        } else if (action == MsgActionEnum.CHAT.type) {
            //  2.2  聊天類型的消息，把聊天記錄保存到數據庫，同時標記消息的簽收狀態[未簽收]
            ChatMsg chatMsg = dataContent.getChatMsg();
            String msgText = chatMsg.getMsg();
            String receiverId = chatMsg.getReceiverId();
            String senderId = chatMsg.getSenderId();
            
            // 保存消息到數據庫，而且標記爲 未簽收
            UserService userService = (UserService)SpringUtil.getBean("userServiceImpl");
            String msgId = userService.saveMsg(chatMsg);
            chatMsg.setMsgId(msgId);
            
            DataContent dataContentMsg = new DataContent();
            dataContentMsg.setChatMsg(chatMsg);
            
            // 發送消息
            // 從全局用戶Channel關係中獲取接受方的channel
            Channel receiverChannel = UserChannelRel.get(receiverId);
            if (receiverChannel == null) {
                // TODO channel爲空表明用戶離線，推送消息（JPush，個推，小米推送）
            } else {
                // 當receiverChannel不爲空的時候，從ChannelGroup去查找對應的channel是否存在
                Channel findChannel = users.find(receiverChannel.id());
                if (findChannel != null) {
                    // 用戶在線
                    receiverChannel.writeAndFlush(
                            new TextWebSocketFrame(
                                    JsonUtils.objectToJson(dataContentMsg)));
                } else {
                    // 用戶離線 TODO 推送消息
                }
            }
            
        } else if (action == MsgActionEnum.SIGNED.type) {
            //  2.3  簽收消息類型，針對具體的消息進行簽收，修改數據庫中對應消息的簽收狀態[已簽收]
            UserService userService = (UserService)SpringUtil.getBean("userServiceImpl");
            // 擴展字段在signed類型的消息中，表明須要去簽收的消息id，逗號間隔
            String msgIdsStr = dataContent.getExtand();
            String msgIds[] = msgIdsStr.split(",");
            
            List<String> msgIdList = new ArrayList<>();
            for (String mid : msgIds) {
                if (StringUtils.isNotBlank(mid)) {
                    msgIdList.add(mid);
                }
            }
            
            System.out.println(msgIdList.toString());
            
            if (msgIdList != null && !msgIdList.isEmpty() && msgIdList.size() > 0) {
                // 批量簽收
                userService.updateMsgSigned(msgIdList);
            }
            
        } else if (action == MsgActionEnum.KEEPALIVE.type) {
            //  2.4  心跳類型的消息
            System.out.println("收到來自channel爲[" + currentChannel + "]的心跳包...");
        }
    }

獲取未簽收的消息列表的接口

在controller中添加獲取未簽收的消息列表的接口getUnReadMsgList。

/**
     * 
     * @Description: 用戶手機端獲取未簽收的消息列表
     */
    @PostMapping("/getUnReadMsgList")
    public IMoocJSONResult getUnReadMsgList(String acceptUserId) {
        // 0. userId 判斷不能爲空
        if (StringUtils.isBlank(acceptUserId)) {
            return IMoocJSONResult.errorMsg("");
        }
        
        // 查詢列表
        List<com.imooc.pojo.ChatMsg> unreadMsgList = userService.getUnReadMsgList(acceptUserId);
        
        return IMoocJSONResult.ok(unreadMsgList);
    }

測試