Netty读书笔记：netty实现WebSocket服务(web's IM)

目录

  • [HttpRequsetHandler][1] - 管理HTTP请求和响应
  • TextWebSocketFrameHandler[2] - 处理聊天的WebSocket帧
  • ChatForWebSocketServerInitailizer [3] - 初始化PipelineChannel
  • ChatWebSocketServer[4]-启动程序
  • 增加SSL- SecurityChatWebSocketServerInitailizer [5]
  • 增加SSL- SecurityChatWebSocketServer[6]

HttpRequsetHandler

import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedNioFile;
import java.io.File;
import java.io.RandomAccessFile;
import java.net.URISyntaxException;
import java.net.URL;

public class HttpRequsetHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    private final String wsUri;
    private static File INDEX;
    static{
        URL location = HttpRequsetHandler.class.getProtectionDomain().getCodeSource().getLocation();
        try{
            String path = location.toURI() + "index.html";
            path = !path.contains("file:")?path:path.substring(5);
            INDEX = new File(path);
        }catch (URISyntaxException e){
            e.printStackTrace();;
        }
    }

    public HttpRequsetHandler(String wsUri) {
        this.wsUri = wsUri;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
        if (wsUri.equalsIgnoreCase(request.getUri())){
            //如果请求了websocket协议升级，则增加引用计数(调用retain()方法),并传给下一个ChannelInboundHandler
            ctx.fireChannelRead(request.retain());
        }else{
            if (HttpHeaders.is100ContinueExpected(request)){ //读取is100ContinueExpected请求，符合http1.1版本
                send100Continue(ctx);
            }
            RandomAccessFile file = new RandomAccessFile(INDEX,"r");
            HttpResponse response = new DefaultFullHttpResponse(
                    request.getProtocolVersion(), HttpResponseStatus.OK
            );
            response.headers().set(
                    HttpHeaders.Names.CONTENT_TYPE,"text/html;charset=UTF-8"
            );
            boolean keepAlive = HttpHeaders.isKeepAlive(request);
            if (keepAlive){ //如果请求了keep-alive，则添加所需要的HTTP头信息
                response.headers().set(
                        HttpHeaders.Names.CONTENT_LENGTH,file.length()
                );
                response.headers().set(
                        HttpHeaders.Names.CONNECTION,HttpHeaders.Values.KEEP_ALIVE
                );
            }
            ctx.write(response);//将httpresponse写入到客户端
            if (ctx.pipeline().get(SslHandler.class)==null){//将index.html写到客户端
                ctx.write(new DefaultFileRegion(file.getChannel(),0,file.length()));
            }else{
                ctx.write(new ChunkedNioFile(file.getChannel()));
            }

            ChannelFuture future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
            if (!keepAlive){
                future.addListener(
                        ChannelFutureListener.CLOSE
                );
            }
        }
    }

    private void send100Continue(ChannelHandlerContext ctx) {
        FullHttpResponse response = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_0,HttpResponseStatus.CONTINUE
        );

        ctx.writeAndFlush(response);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
        cause.printStackTrace();
        ctx.close();
    }
}

TextWebSocketFrameHandler-处理聊天的WebSocket帧


public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    private final ChannelGroup group;
    public TextWebSocketFrameHandler(ChannelGroup group) {
        this.group = group;
    }
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt== WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE){
            //握手成功，则从该ChannelPipeline移出HttpRequsetHandler，因为它将接收不到任何消息了
            ctx.pipeline().remove(HttpRequsetHandler.class);//
            //通知所有已连接的WebSocket客户端，新客户端已上线
            group.writeAndFlush(new TextWebSocketFrame("CLient "+ctx.channel()) +"joined");
            group.add(ctx.channel());
        }else{
            super.userEventTriggered(ctx,evt);
        }
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        group.writeAndFlush(msg.retain());//增加消息的引用计数，并将它写到ChannelGroup中所有已连接的客户端
    }
}

初始化PipelineChannel

public class ChatForWebSocketServerInitailizer extends ChannelInitializer<Channel> {
    private final ChannelGroup group;

    public ChatForWebSocketServerInitailizer(ChannelGroup congrouptext) {
        this.group = congrouptext;
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast(
                new HttpServerCodec(), //字节解码为httpreqeust,httpcontent,last-httpcontent
                new ChunkedWriteHandler(),//写入一个文本
                new HttpObjectAggregator(64*1024), //HttpMessage与多个HttpConent聚合
                new HttpRequsetHandler("/ws"), //处理FullJttpRequest——不往/ws上发的请求
                new WebSocketServerProtocolHandler("/ws"), //处理websocket升级握手、PingWebSocketFrame、PongWebSocketFrame、CloseWebSocketFrame
                new TextWebSocketFrameHandler(group), //处理TextWebSocketFrameHandler和握手完成事件,如握手成功，则所需要的Handler将会被添加到ChannelPipeline，不需要的则会移除              
        );//如果请求端是websocket则升级该握手
    }
}

ChatWebSocketServer-启动程序

public class ChatWebSocketServer {
    //创建DefaultChannelGroup，将其保存为所有已经连接的WebSocket Channel
    private final ChannelGroup channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);

    private final EventLoopGroup group = new NioEventLoopGroup();
    private Channel channel;

    public ChannelFuture start(InetSocketAddress address){
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(group)
                .channel(NioServerSocketChannel.class)
                .childHandler(createInitializer(channelGroup));
        ChannelFuture future = bootstrap.bind(address);
        future.syncUninterruptibly();
        channel = future.channel();
        return future;
    }

    protected ChannelHandler createInitializer(ChannelGroup channelGroup) {
        return new ChatWebSocketServerInitailizer(channelGroup);
    }


    public void destory(){
        if (channel!=null){
            channel.close();
        }
        channelGroup.close();
        group.shutdownGracefully();
    }

    public static void main(String[] args) {
        if (args.length!=1){
            System.out.print("Please give port as argument!");
            System.exit(1);
        }

        int port = Integer.parseInt(args[0]);
        final ChatWebSocketServer endpoint = new ChatWebSocketServer();
        ChannelFuture future = endpoint.start(new InetSocketAddress(port));
        Runtime.getRuntime().addShutdownHook(new Thread(){
            @Override
            public void run(){
                endpoint.destory();
            }
        });

        future.channel().closeFuture().syncUninterruptibly();
    }
}

加密

SecurityChatWebSocketServerInitailizer

public class SecurityChatWebSocketServerInitailizer extends ChatWebSocketServerInitailizer {
    private final SslContext context;
    public SecurityChatWebSocketServerInitailizer(ChannelGroup congrouptext, SslContext context) {
        super(congrouptext);
        this.context = context;
    }

    @Override
    protected  void initChannel(Channel ch) throws Exception{
        super.initChannel(ch);

        SSLEngine engine = context.newEngine(ch.alloc());
        engine.setUseClientMode(false);

        ch.pipeline().addFirst(new SslHandler(engine)); //将SslHandler添加到ChannelPipeline中
    }
}

启动程序 - SecurityChatWebSocketServer

public class SecurityChatWebSocketServer extends ChatWebSocketServer{
    private final SslContext context;
    public SecurityChatWebSocketServer(SslContext context) {
        this.context = context;
    }

    public ChannelFuture start(InetSocketAddress address){
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(group)
                .channel(NioServerSocketChannel.class)
                .childHandler(createInitializer(channelGroup));
        ChannelFuture future = bootstrap.bind(address);
        future.syncUninterruptibly();
        channel = future.channel();
        return future;
    }

    protected ChannelHandler createInitializer(ChannelGroup channelGroup) {
        //return new ChatWebSocketServerInitailizer(channelGroup);
        return new SecurityChatWebSocketServerInitailizer(channelGroup,context);
    }


    public void destory(){
        if (channel!=null){
            channel.close();
        }
        channelGroup.close();
        group.shutdownGracefully();
    }

    public static void main(String[] args) throws SSLException, CertificateException {
        if (args.length!=1){
            System.out.print("Please give port as argument!");
            System.exit(1);
        }


        int port = Integer.parseInt(args[0]);
        SelfSignedCertificate cert = new SelfSignedCertificate();
        SslContext context = SslContext.newServerContext(cert.certificate(),cert.privateKey());
        final SecurityChatWebSocketServer endpoint = new SecurityChatWebSocketServer(context);
        ChannelFuture future = endpoint.start(new InetSocketAddress(port));
        Runtime.getRuntime().addShutdownHook(new Thread(){
            @Override
            public void run(){
                endpoint.destory();
            }
        });

        future.channel().closeFuture().syncUninterruptibly();
    }
}
  1. ↩
  2. ↩
  3. ↩
  4. ↩
  5. ↩
  6. ↩
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 220,002评论 6赞 509
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 93,777评论 3赞 396
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 166,341评论 0赞 357
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 59,085评论 1赞 295
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 68,110评论 6赞 395
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 51,868评论 1赞 308
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,528评论 3赞 420
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 39,422评论 0赞 276
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 45,938评论 1赞 319
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 38,067评论 3赞 340
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,199评论 1赞 352
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,877评论 5赞 347
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,540评论 3赞 331
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 32,079评论 0赞 23
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,192评论 1赞 272
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 48,514评论 3赞 375
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 45,190评论 2赞 357