【Netty聊天系统】吃透 SpringBoot + Netty , 还开发不了通讯系统？（三）session

netty是没有session，所需要我们自己创建

所以需要自己创建一个对象的，进行存储，就叫用户传输 

package com.binglian.common;

import com.binglian.message.MessagePro;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Data
public class UserTransmission {


    private String uid;
    private String devId= UUID.randomUUID().toString();
    private String token= UUID.randomUUID().toString();
    private String username;
    PLATTYPE platform = PLATTYPE.MAC;

    // windows,mac,android, ios, web , other
    public enum PLATTYPE {
        WINDOWS, MAC, ANDROID, IOS, WEB, OTHER;
    }

    private String sessionId;


    public void setPlatform(int platform) {
        PLATTYPE[] values = PLATTYPE.values();
        for (int i = 0; i < values.length; i++) {
            if (values[i].ordinal() == platform) {
                this.platform = values[i];
            }
        }

    }



    public static UserTransmission fromMsg(MessagePro.LoginRequest info) {
        UserTransmission userTransmission = new UserTransmission();
        userTransmission.uid = new String(info.getUid());
        userTransmission.devId = new String(info.getDeviceId());
        userTransmission.token = new String(info.getToken());
        userTransmission.setPlatform(info.getPlatform());
        log.info("登录中: {}", userTransmission.toString());
        return userTransmission;

    }

}

这里用了一个容器存放session_key，这个是Netty自带的，唯一性，这里用来绑定通道的，这一篇不会用到，但是后面一篇会用到。

public static final AttributeKey SESSION_KEY =
        AttributeKey.valueOf("SESSION_KEY");

//	获取通道
channel.attr(ServerSession.SESSION_KEY).get();

//	this就是把当前session绑定搭配key中
channel.attr(ServerSession.SESSION_KEY).set(this);

package com.binglian.session;

import com.binglian.common.UserTransmission;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@Slf4j
@Data
public class ServerSession {

    public static final AttributeKey SESSION_KEY = AttributeKey.valueOf("SESSION_KEY");

    //  通道
    private Channel channel;


    private UserTransmission userTransmission;

    private final String sessionId;

    private boolean isLogin = false;

    
    private Map map = new HashMap();

    public ServerSession(Channel channel) {
        this.channel = channel;
        this.sessionId = buildNewSessionId();
    }

    //反向导航
    public static ServerSession getSession(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        return channel.attr(ServerSession.SESSION_KEY).get();
    }

    //关闭连接
    public static void closeSession(ChannelHandlerContext ctx) {
        ServerSession session =
                ctx.channel().attr(ServerSession.SESSION_KEY).get();

        if (null != session && session.isValid()) {
            session.close();
            SessionMap.inst().removeSession(session.getSessionId());
        }
    }

    //和channel 通道实现双向绑定
    public ServerSession bind() {
        log.info(" ServerSession 绑定会话 " + channel.remoteAddress());
        channel.attr(ServerSession.SESSION_KEY).set(this);
        SessionMap.inst().addSession( this);
        isLogin = true;
        return this;
    }


    public ServerSession unbind() {
        isLogin = false;
        SessionMap.inst().removeSession(getSessionId());
        this.close();
        return this;
    }

    public String getSessionId() {
        return sessionId;
    }

    private static String buildNewSessionId() {
        String uuid = UUID.randomUUID().toString();
        return uuid.replaceAll("-", "");
    }

    //关闭连接
    public synchronized void close() {
        ChannelFuture future = channel.close();
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    log.error("CHANNEL_CLOSED error ");
                }
            }
        });
    }
    public synchronized void set(String key, Object value) {
        map.put(key, value);
    }


    public synchronized  T get(String key) {
        return (T) map.get(key);
    }


    public boolean isValid() {
        return getUser() != null ? true : false;
    }

    //写Protobuf数据帧
    public synchronized void writeAndFlush(Object pkg) {
        //当系统水位过高时，系统应不继续发送消息，防止发送队列积压
        //写Protobuf数据帧

        if (channel.isWritable()) //低水位
        {
            channel.writeAndFlush(pkg);
        } else {   //高水位时
            log.debug("通道很忙，消息被暂存了");
            //写入消息暂存的分布式存储，如果mongo
            //等channel空闲之后，再写出去
        }

    }




    public UserTransmission getUser() {
        return userTransmission;
    }

    public void setUserTransmission(UserTransmission user) {
        this.userTransmission = user;
        user.setSessionId(sessionId);
    }
}

sessionMap存放，根据sessionID的，添加map，这里用到了ConcurrentHashMap，进行存放，因为它是线程安全的。

package com.binglian.session;

import com.binglian.common.UserTransmission;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;


import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Slf4j
@Data
public final class SessionMap {
    private SessionMap() {
    }

    private static SessionMap singleInstance = new SessionMap();

    //  会话集合
    private ConcurrentHashMap map =
            new ConcurrentHashMap();

    //  单例模式，获取对象
    public static SessionMap inst() {
        return singleInstance;
    }

    
    public void addSession(ServerSession s) {
        map.put(s.getSessionId(), s);
        log.info("用户登录:id= " + s.getUser().getUid()
                + "   在线总数: " + map.size());

    }

    
    public ServerSession getSession(String sessionId) {
        if (map.containsKey(sessionId)) {
            return map.get(sessionId);
        } else {
            return null;
        }
    }

    
    public List getSessionsBy(String userId) {

        List list = map.values()
                .stream()
                .filter(s -> s.getUser().getUid().equals(userId))
                .collect(Collectors.toList());
        return list;
    }

    
    public void removeSession(String sessionId) {
        if (!map.containsKey(sessionId)) {
            return;
        }
        ServerSession s = map.get(sessionId);
        map.remove(sessionId);
        log.info("用户下线:id= " + s.getUser().getUid()
                + "   在线总数: " + map.size());
    }



}

Netty有自带的一些编码，下面，只列举两个

StringEncoder，对字符串数据进行编码
ObjectEncoder，对 Java 对象进行编码

对应

StringDecoder, 对字符串数据进行解码

ObjectDecoder，对 Java 对象进行解码

这里我们用的是自定义的

extends MessageToByteEncoder就用到了，我们上面表格所说的。

这里用到了魔术和版本、长度、内容，按照此规范，传输到服务端，服务端那边也会按照编码的规范进行解析信息判断。

package com.binglian.common;

import com.binglian.message.MessagePro;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.extern.slf4j.Slf4j;


@Slf4j
public class ProtobufEncoder extends MessageToByteEncoder {

    @Override
    protected void encode(ChannelHandlerContext ctx,
                          MessagePro.Message msg, ByteBuf out)
            throws Exception {
        encode0(msg, out);
    }

    public static void encode0(
            MessagePro.Message msg, ByteBuf out) {

        //  将魔数和版本都写进去
        out.writeShort(ProtoInstant.MAGIC_CODE);
        out.writeShort(ProtoInstant.VERSION_CODE);


        // 将对象转换为byte
        byte[] bytes = msg.toByteArray();

        // 加密消息体
        int length = bytes.length;// 读取消息的长度

        log.info("message:---" + bytes.toString());

        // 先将消息长度写入，也就是消息头
        out.writeInt(length);
        // 消息体中包含我们要发送的数据
        out.writeBytes(bytes);
    }

}

这里就集成了，自定义ByteToMessageDecoder，解码按照编码规定的格式

获取魔数是否匹配枚举类的获取版本是否匹配获取长度然后在获取内容

package com.binglian.common;

import com.binglian.message.MessagePro;
import com.google.protobuf.InvalidProtocolBufferException;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.List;


@Slf4j
public class ProtobufDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List