物联网的服务端设计(三)会话管理

    /**
     * 当前服务器预估的最大连接数
     */
    private static final int SERVER_CONNECT_NUM = 60000;

    /**
     * 存储会话,为了防止使用map时进行动态扩容，初始化时直接指定一个预估的单服务器连接数
     */
    private final Map<Channel, SessionInfo> sessions = new ConcurrentHashMap<>(SERVER_CONNECT_NUM);

    /**
     * 存储管道
     */
    private final Map<String, Channel> channels = new ConcurrentHashMap<>(SERVER_CONNECT_NUM);

    /**
     * 增加会话，当终端连接上来就进行注册。
     * 终端发送login包时再更新会话属性
     * @param channel 通道
     */
    public void addSession(Channel channel) {
        SessionInfo session = new SessionInfo();
        session.setChannel(channel);
        sessions.put(channel, session);
    }

    /**
     * 终端登录后补充会话信息,现时增加channel
     * @param channel channel
     * @param devId 设备ID
     * @param version 设备版本
     */
    public void setSession(Channel channel, String devId, String version){
        SessionInfo session = sessions.get(channel);
        session.setVersion(version);
        session.setDevId(devId);

        channels.put(devId, channel);
    }	

	/**
     * 移除会话，当终端断开时请求
     * @param channel channel
     */
    public void removeSession(Channel channel) {

        SessionInfo sessionInfo = this.getSession(channel);
        if(sessionInfo != null) {
            if(sessionInfo.getDevId() != null && !sessionInfo.getDevId().isEmpty()) {
                // 如果设备已经登录过，还需要同时移除channel
                sessions.remove(channel);
                channels.remove(sessionInfo.getDevId());
            } else {
                sessions.remove(channel);
            }
        }
    }

    /**
     * 构造方法用来接收 sessionManager 对象
     * @param sessionManager 会话管理
     */
    public DecoderHandler(SessionManager sessionManager){
        this.sessionManager = sessionManager;
    }

    /**
     * 当有客户端注册时调用
     * @param ctx ChannelHandlerContext
     */
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        ChannelConfig config = ctx.channel().config();
        DefaultSocketChannelConfig socketConfig = (DefaultSocketChannelConfig)config;
        // 此处三个参数决定 延迟情况
        // 连接时间 、往返延迟、 带宽。
        // 这三个参数设置的是权重
        // 因为我的连接会保持住 长连接不会频繁断开，所以 把连接时间权限设置的最低为0
        // 因为我们对往返延迟有一些容忍度，所以 第二参数是1
        // 对于带宽我们会有更大的需求，第三个参数设置为2 这就是目前的权重比
        // 延迟和带宽的性能是互斥的 , 延迟低 , 就意味着很小的包就要发送一次 , 其带宽就低了 , 延迟高了 , 每次积累很多数据才发送 , 其带宽就相应的提高了
        socketConfig.setPerformancePreferences(0,1,2);
        // NioSocketChannel在工作过程中，使用PooledByteBufAllocator来分配内存
        socketConfig.setAllocator(PooledByteBufAllocator.DEFAULT);
        super.channelRegistered(ctx);

        // 增加会话
        sessionManager.addSession(ctx.channel());
    }

@Override
public void run() {
    logger.info(
        "+---------+---------+----------------+---------------+\n" +
        "| session | channel | main disruptor | log disruptor |\n" +
        "+---------+---------+----------------+---------------+\n"+
        "| "+String.format("%7d",sessionCount)+" | "+String.format("%7d",channelCount)+" |        "+String.format("%7d",mainDis)+" |       "+String.format("%7d",logDis)+" |\n"+
        "+---------+---------+----------------+---------------+\n");
    }

// 为worker组设置一个定时器,其中参数2为首次调用等待，参数3为之后每次调用间隔等待，参数4是时间单位
workerGroup.next().scheduleAtFixedRate(statusPringHandler,1, 60, TimeUnit.SECONDS);

/**
 * 空闲检测器
 * 如果增加@Sharable注解，该类必须是线程安全的
 * @author guohai
 */
@Component
@ChannelHandler.Sharable
public class IdleCheckHandler extends ChannelDuplexHandler {
    /**
     * 日志
     */
    private static final Logger logger = LoggerFactory.getLogger(StatusPringHandler.class);

    /**
     * 空闲会话检测
     * @param ctx 管道
     * @param evt 事件对象
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent e = (IdleStateEvent) evt;
            if (e.state() == IdleState.READER_IDLE) {
                // TODO: 读空闲，准备断开客户端,测试阶段先不实现
                logger.debug("读空闲，准备断开客户端");
            } else if (e.state() == IdleState.WRITER_IDLE) {
                logger.debug("写空闲，下行一条心跳保持连接");
                // TODO: 下行数据先写死
                ctx.channel().writeAndFlush(Unpooled.copiedBuffer("{\"msgType\": 20, \"txNo\": \"1234567890123\"}\n", CharsetUtil.UTF_8));
            }
        }
    }
}

	@Override
	public void initChannel(SocketChannel ch) {
		ch.pipeline()
				// 增加空闲检查器，规定读写各30秒没操作时触发
				.addLast(new IdleStateHandler(30,30,0))
				//自定义实现的空闲处理
				.addLast(idleCheckHandler);
	}