Netty入門4之----如何實現長連接

​ 前面三章介紹了Netty的一些基本用法，這一章介紹怎么使用Netty來實現一個簡單的長連接demo。

關於長連接的背景知識，可以參考《如何使用Socket實現長連接》

​ 一個簡單的長連接demo分為以下幾個步驟：

長連接流程

1. 創建連接（Channel）
2. 發心跳包
3. 發消息，並通知其他用戶
4. 一段時間沒收到心跳包或者用戶主動關閉之后關閉連接

​ 看似簡單的步驟，里面有兩個技術難點：

1. 如何保存已創建的Channel

   這里我們是將Channel放在一個Map中，以Channel.hashCode()作為key

   其實這樣做有一個劣勢，就是不適合水平擴展，每個機器都有一個連接數的上線，如果需要實現多用戶實時在線，對機器的數量要求會很高，在這里我們不多做討論，不同的業務場景，設計方案也是不同的，可以在長連接方案和客戶端輪詢方案中進行選擇。

2. 如何自動關閉沒有心跳的連接

   Netty有一個比較好的Feature，就是ScheduledFuture，他可以通過ChannelHandlerContext.executor().schedule()創建，支持延時提交，也支持取消任務，這就給我們心跳包的自動關閉提供了一個很好的實現方案。

開始動手

​ 首先，我們需要用一個JavaBean來封裝通信的協議內容，在這里我們只需要三個數據就行了：

1. type : byte，表示消息的類型，有心跳類型和內容類型
2. length : int，表示消息的長度
3. content : String，表示消息的內容（心跳包在這里沒有內容）

​ 然后，因為我們需要將Channel和ScheduledFuture緩存在Map里面，所以需要將兩個對象組合成一個JavaBean。

​ 接着，需要完成輸入輸出流的解析和轉換，我們需要重寫Decoder和Encoder，具體可以參考Netty筆記3-Decoder和Encoder。

​ 最后，就是需要完成ChannelHandler了，代碼如下：

package com.dz.netty.live;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.concurrent.ScheduledFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/** * Created by RoyDeng on 17/7/20. */
public class LiveHandler extends SimpleChannelInboundHandler<LiveMessage> { // 1

    private static Map<Integer, LiveChannelCache> channelCache = new HashMap<>();
    private Logger logger = LoggerFactory.getLogger(LiveHandler.class);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, LiveMessage msg) throws Exception {
        Channel channel = ctx.channel();
        final int hashCode = channel.hashCode();
        System.out.println("channel hashCode:" + hashCode + " msg:" + msg + " cache:" + channelCache.size());

        if (!channelCache.containsKey(hashCode)) {
            System.out.println("channelCache.containsKey(hashCode), put key:" + hashCode);
            channel.closeFuture().addListener(future -> {
                System.out.println("channel close, remove key:" + hashCode);
                channelCache.remove(hashCode);
            });
            ScheduledFuture scheduledFuture = ctx.executor().schedule(
                    () -> {
                        System.out.println("schedule runs, close channel:" + hashCode);
                        channel.close();
                    }, 10, TimeUnit.SECONDS);
            channelCache.put(hashCode, new LiveChannelCache(channel, scheduledFuture));
        }

        switch (msg.getType()) {
            case LiveMessage.TYPE_HEART: {
                LiveChannelCache cache = channelCache.get(hashCode);
                ScheduledFuture scheduledFuture = ctx.executor().schedule(
                        () -> channel.close(), 5, TimeUnit.SECONDS);
                cache.getScheduledFuture().cancel(true);
                cache.setScheduledFuture(scheduledFuture);
                ctx.channel().writeAndFlush(msg);
                break;
            }
            case LiveMessage.TYPE_MESSAGE: {
                channelCache.entrySet().stream().forEach(entry -> {
                    Channel otherChannel = entry.getValue().getChannel();
                    otherChannel.writeAndFlush(msg);
                });
                break;
            }
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        logger.debug("channelReadComplete");
        super.channelReadComplete(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.debug("exceptionCaught");
        if(null != cause) cause.printStackTrace();
        if(null != ctx) ctx.close();
    }
}

​ 寫完服務端之后，我們需要有客戶端連接來測試這個項目，教程參考如何使用Socket在客戶端實現長連接，代碼如下：

package com.dz.test;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.Scanner;

/** * Created by RoyDeng on 18/2/3. */
public class LongConnTest {

    private Logger logger = LoggerFactory.getLogger(LongConnTest.class);

    String host = "localhost";
    int port = 8080;

    public void testLongConn() throws Exception {
        logger.debug("start");
        final Socket socket = new Socket();
        socket.connect(new InetSocketAddress(host, port));
        Scanner scanner = new Scanner(System.in);
        new Thread(() -> {
            while (true) {
                try {
                    byte[] input = new byte[64];
                    int readByte = socket.getInputStream().read(input);
                    logger.debug("readByte " + readByte);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        int code;
        while (true) {
            code = scanner.nextInt();
            logger.debug("input code:" + code);
            if (code == 0) {
                break;
            } else if (code == 1) {
                ByteBuffer byteBuffer = ByteBuffer.allocate(5);
                byteBuffer.put((byte) 1);
                byteBuffer.putInt(0);
                socket.getOutputStream().write(byteBuffer.array());
                logger.debug("write heart finish!");
            } else if (code == 2) {
                byte[] content = ("hello, I'm" + hashCode()).getBytes();
                ByteBuffer byteBuffer = ByteBuffer.allocate(content.length + 5);
                byteBuffer.put((byte) 2);
                byteBuffer.putInt(content.length);
                byteBuffer.put(content);
                socket.getOutputStream().write(byteBuffer.array());
                logger.debug("write content finish!");
            }
        }
        socket.close();
    }

    // 因為Junit不支持用戶輸入,所以用main的方式來執行用例
    public static void main(String[] args) throws Exception {
        new LongConnTest().testLongConn();
    }
}

運行main方法之后，輸入1表示發心跳包，輸入2表示發content，5秒內不輸入1則服務端會自動斷開連接。