Java IO流(五)Netty实战[TCP|Http|心跳检测|Websocket]

这篇具有很好参考价值的文章主要介绍了Java IO流(五)Netty实战[TCP|Http|心跳检测|Websocket]。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

Netty入门代码示例(基于TCP服务)

Server端

package com.bierce.io.netty.simple;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.CharsetUtil;
public class NettyServer {
    public static void main(String[] args) throws InterruptedException {
        //创建BossGroup和WorkerGroup线程池组,均属于自旋状态
        EventLoopGroup bossGroup = new NioEventLoopGroup(); //负责连接请求处理
        EventLoopGroup workerGroup = new NioEventLoopGroup(); //进行业务处理
        try {
            //创建服务器端启动,通过链式编程配置相关参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class) //设置为服务端通道
                    .option(ChannelOption.SO_BACKLOG,128) //设置线程队列等待连接的个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true) //设置保持活动连接状态
                    .childHandler(new ChannelInitializer<SocketChannel>() { //匿名创建通道初始对象
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new NettyServerHandler()); //为workerGroup下的NioEventLoop对应管道pipeline设置自定义处理器
                        }
                    });
            System.out.println("Server is start Successful !!!");
            ChannelFuture cf = bootstrap.bind(6668).sync(); //绑定指定端口并同步处理
            cf.channel().closeFuture().sync(); //监听关闭通道方法
        }finally { //关闭线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}
class NettyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //读取客户端发送的数据
        //ctx:上下文对象,包含管道pipeline,通道等信息
        //msg:即客户端发送的数据
        ByteBuf buf = (ByteBuf) msg; //ByteBuf是Netty提供的缓冲区,性能更高
        System.out.println("客户端发送过来的msg = " + buf.toString((CharsetUtil.UTF_8)));
        System.out.println("客户端地址 = " + ctx.channel().remoteAddress());
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { //读取客户端信息完成后进行的业务处理
//        super.channelReadComplete(ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello, Client!",CharsetUtil.UTF_8)); //将数据写到缓存并刷新
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close(); //处理异常需要关闭通道
    }
}

 Client

package com.bierce.io.netty.simple;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.CharsetUtil;
public class NettyClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(); //客户端事件循环组
        try {
            Bootstrap bootstrap = new Bootstrap(); //客户端启动对象
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class) //客户端通道
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new NettyClientHandler()); //添加自定义处理器
                        }
                    });
            System.out.println("Client Start Successful!!!");
            ChannelFuture sync = bootstrap.connect("127.0.0.1", 6668).sync();
            sync.channel().closeFuture().sync(); //监听关闭通道
        }finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}
class NettyClientHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception { //通道就绪会触发该方法
        ctx.writeAndFlush(Unpooled.copiedBuffer("Hello, Server!", CharsetUtil.UTF_8)); //将数据写到缓存并刷新
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //读取服务端返回信息
        ByteBuf buf = (ByteBuf) msg; //ByteBuf是Netty提供的缓冲区,性能更高
        System.out.println("服务端发送过来的msg = " + buf.toString((CharsetUtil.UTF_8)));
        System.out.println("服务端地址 = " + ctx.channel().remoteAddress());
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close(); //处理异常需要关闭通道
    }
}

运行结果

Java IO流(五)Netty实战[TCP|Http|心跳检测|Websocket],# Java,java,nio,网络
Server
Java IO流(五)Netty实战[TCP|Http|心跳检测|Websocket],# Java,java,nio,网络
Client

Netty入门代码示例(基于HTTP服务)

package com.bierce.io.netty.http;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;

import java.net.URI;

public class TestServer {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //创建服务器端启动,通过链式编程配置相关参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class) //设置为服务端通道
                    .childHandler(new TestServerInitializer()); //设置为自定义的初始化
            System.out.println("Server is start Successful !!!");
            ChannelFuture cf = bootstrap.bind(9999).sync(); //绑定指定端口并同步处理
            cf.channel().closeFuture().sync(); //监听关闭通道方法
        }finally { //关闭线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}
class TestServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        //HttpServerCodec是Netty提供的处理Http的编-解码器 使用io.netty:netty-all:4.1.20Final版本,其他版本不支持会报错
        pipeline.addLast("MyHttpServerCodec", new HttpServerCodec());
        //增加自定义的handler
        pipeline.addLast("MyTestHttpServerHandler", new TestHttpServerHandler());
    }
}
class TestHttpServerHandler extends SimpleChannelInboundHandler<HttpObject> {
    //读取客户端数据
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpObject httpObject) throws Exception {
        if (httpObject instanceof HttpRequest){
            System.out.println("httpObject Type = " + httpObject.getClass());
            System.out.println("Client Address = " + channelHandlerContext.channel().remoteAddress());
            //对特定资源进行过滤
            HttpRequest httpRequest = (HttpRequest)httpObject;
            URI uri = new URI(httpRequest.getUri());
            if ("/favicon.ico".equals(uri.getPath())){
                System.out.println("favicon.ico资源不做响应");
                return;
            }
            //回复浏览器信息(http协议)
            ByteBuf content = Unpooled.copiedBuffer("Hello, I'm Server", CharsetUtil.UTF_8);
            FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
            response.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/plain");
            response.headers().set(HttpHeaderNames.CONTENT_LENGTH,content.readableBytes());
            channelHandlerContext.writeAndFlush(response);
        }
    }
}

运行结果

Java IO流(五)Netty实战[TCP|Http|心跳检测|Websocket],# Java,java,nio,网络

Netty心跳检测机制

package com.bierce.io.netty.heartbeat;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;
public class MyServer {
    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //IdleStateHandler:Netty提供的处理空闲状态的处理器
                            //readerIdleTime:多长时间没有读操作,会发送心跳检测包检测是否连接
                            //writerIdleTime:多长时间没有写操作,会发送心跳检测包检测是否连接
                            //allIdleTime:多长时间没有读写操作,会发送心跳检测包检测是否连接
                            pipeline.addLast(new IdleStateHandler(3,5,7, TimeUnit.SECONDS));
                            //IdleStateHandler触发后,将传递给下一个Handler的userEventTriggered方法去处理
                            //通过自定义的Handler对空闲状态进一步处理
                            pipeline.addLast(new MyServerHandler());
                        }
                    });
            ChannelFuture sync = bootstrap.bind(9999).sync().channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}
class MyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent){
            IdleStateEvent IdleStateEvent = (IdleStateEvent) evt;
            String eventType = null;
            switch (IdleStateEvent.state()){
                case READER_IDLE:
                    eventType = "读空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
            }
            System.out.println(ctx.channel().remoteAddress() + "-已超时,超时类型为: " + eventType );
            System.out.println("Server will deal with it instantly...");
            //发生空闲则关闭当前通道
            ctx.close();
        }
    }
}
class NettyClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(); //客户端事件循环组
        try {
            Bootstrap bootstrap = new Bootstrap(); //客户端启动对象
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class) //客户端通道
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new NettyClientHandler()); //添加自定义处理器
                        }
                    });
            System.out.println("Client Start Successful!!!");
            ChannelFuture sync = bootstrap.connect("127.0.0.1", 9999).sync();
            sync.channel().closeFuture().sync(); //监听关闭通道
        }finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}

Java IO流(五)Netty实战[TCP|Http|心跳检测|Websocket],# Java,java,nio,网络

 注意: 需要调整readerIdleTime|writerIdleTime|allIdleTime参数才会显示对应超时信息

Netty入门代码示例(基于WebSocket协议)

服务端

package com.bierce.websocket;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

import java.time.LocalDateTime;

public class MyWebsocketServer {
    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup,workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //基于Http协议,所以需要http解码编码
                            pipeline.addLast(new HttpServerCodec());
                            pipeline.addLast(new ChunkedWriteHandler()); //处理块方式的写操作
                            //http传输过程数据量非常大时会分段,而HttpObjectAggregator可以将多个分段聚合
                            pipeline.addLast(new HttpObjectAggregator(8192));
                            //webSocket采用帧方式传输数据
                            //WebSocketServerProtocolHandler作用是将http协议升级为ws协议,且保持长连接
                            pipeline.addLast(new WebSocketServerProtocolHandler("/hello"));
                            pipeline.addLast(new MyWebsocketServerHandler());
                        }
                    });
            ChannelFuture sync = bootstrap.bind(9999).sync().channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}
class MyWebsocketServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        System.out.println("Server receive the Info: " + msg.text());
        ctx.channel().writeAndFlush(new TextWebSocketFrame("Server time " + LocalDateTime.now() + " --- " + msg.text()));
    }
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        System.out.println("有客户端连接成功 --" + ctx.channel().id().asLongText()); //asLongText唯一值
    }
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println("有客户端已经离开 --" + ctx.channel().id().asLongText()); //asLongText唯一值
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("异常Info:" + cause.getMessage());
        ctx.close();
    }
}

客户端(浏览器)

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Websocket</title>
</head>
<body>
<script>
    var socket;
    if (window.WebSocket) {
        socket = new WebSocket("ws://localhost:9999/hello");
        //相当于channelRead0,读取服务器端的消息
        socket.onmessage = function(ev){
            var rt = document.getElementById("responseText");
            rt.value = rt.value + "\n" + ev.data;
        }
        //开启连接
        socket.onopen = function(ev){
            var rt = document.getElementById("responseText");
            rt.value = "开启连接成功!";
        }
        //连接关闭
        socket.onclose = function(ev){
            var rt = document.getElementById("responseText");
            rt.value = rt.value + "\n" + "连接关闭成功!";
        }
    }
    //发送消息给服务器
    function send(msg){
        if(!window.socket){ //是否已创建socket
            return;
        }
        if(socket.readyState == WebSocket.OPEN){
            socket.send(msg);
        }else{
            alert("socket未连接");
        }
    }
</script>
    <form onsubmit="return false">
        <textarea name="message" style="height:300px;width:300px"></textarea>
        <input type="button" value="Send" onclick="send(this.form.message.value)">
        <textarea id="responseText" style="height:300px;width:300px"></textarea>
        <input type="button" value="Clear" onclick="document.getElementById('responseText').value=''">
    </form>
</body>
</html>

效果图

Java IO流(五)Netty实战[TCP|Http|心跳检测|Websocket],# Java,java,nio,网络

Java IO流(五)Netty实战[TCP|Http|心跳检测|Websocket],# Java,java,nio,网络文章来源地址https://www.toymoban.com/news/detail-668157.html

到了这里,关于Java IO流(五)Netty实战[TCP|Http|心跳检测|Websocket]的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • websocket断线重连&&心跳检测

    封装websocket 实现断线重连跟心态检测,使用的typeScript去封装 在nodejs 安装ws库 代码如下(示例):  服务端实现ws 创建一个server.js 文件 运行ws服务   node .server.js  客户端实现websocket 创建一个socket.ts 文件 vue 页面使用 断开ws服务 断线  启动服务后 自动重连

    2024年01月19日
    浏览(39)
  • WebSocket心跳检测和重连机制

    心跳和重连的目的用一句话概括就是客户端和服务端保证彼此还活着,避免丢包发生。 websocket 连接断开有以下两证情况: 前端断开 在使用 websocket 过程中,可能会出现网络断开的情况,比如信号不好,或者网络临时关闭,这时候websocket的连接已经断开,而不同浏览器有不同

    2024年01月21日
    浏览(39)
  • websocket实时通讯和socket.io实时通信库的使用;心跳机制与断线重连

    https://zh.javascript.info/websocket WebSocket 是一种网络通信协议,就类似于 HTTP 也是一种通信协议。 为什么需要 WebSocket? 因为 HTTP 协议有一个缺陷:通信只能由客户端发起。 代码解析: 创建WebSocket实例:通过 new WebSocket() 创建一个WebSocket实例。在括号中传入服务器的URL,该URL指定了

    2024年02月16日
    浏览(41)
  • Java netty发送接收(TCP、UDP)

    最下方附项目地址 项目地址 https://gitee.com/xn-mg/netty_kafka

    2024年02月16日
    浏览(40)
  • 为什么WebSocket需要前端心跳检测,有没有原生的检测机制?

    本文代码 github、gitee、npm 在web应用中,WebSocket是很常用的技术。通过浏览器的WebSocket构造函数就可以建立一个WebSocket连接。但当需要应用在具体项目中时,几乎都会进行心跳检测。 设置心跳检测,一是让通讯双方确认对方依旧活跃,二是浏览器端及时检测当前网络线路可用

    2024年02月03日
    浏览(58)
  • 【netty】java如何作为websocket客户端 对服务端发起请求

    是的 本文介绍java如何作为客户端 发起websocket请求 博主不做标题党 不会服务端客户端分不清就写个标题 乱写文章 为什么会使用java作为websocket客户端? 虽说websocket协议 本意是web与服务端之间的通讯协议,那假设有一天 我们的供应商 或者是甲方大爷 只提供了websocket接口呢?

    2024年02月05日
    浏览(49)
  • WebSocket实战之六心跳重连机制

    WebSocket应用部署到生产环境,我们除了会碰到因为经过代理服务器无法连接的问题(注:该问题可以通过搭建WSS来解决,具体配置请看 WebSocket实战之四WSS配置 ),另外一个问题就是外网环境不稳定经常会断开或者服务器重启或者网络中间服务器当发现一个长连接长时间没有

    2024年02月07日
    浏览(46)
  • Vue 2 中 WebSocket 模块实现与应用(包含心跳检测、自动重连)

    WebSocket 技术是一种在 Web 开发中常用的实时通信方式,它允许客户端和服务器之间建立持久性的双向连接,以便实时地传输数据。在 Vue.js 项目中,使用 WebSocket 可以轻松实现实时消息推送、即时通讯等功能。在这篇博客中,我们将介绍一个基于 Vue.js 的 WebSocket 模块的实现,

    2024年02月03日
    浏览(37)
  • websocket报错:java.io.EOFException: null

    提示:我这里是websocket在断开连接时就会提示这个错误,但websocket连接的时候没问题,不太清楚怎么回事,如有大佬清楚,希望可以指教一下,谢谢 既然提示没有为它配置错误处理,那我们就为它配置一下错误处理 注意上面两个参数是一点要的,不要的话启动会报错,不要

    2024年02月11日
    浏览(38)
  • Golang 实现http协议的心跳检测程序

    本文介绍如何使用Golang实现心跳程序。 实现心跳程序,其他应用可以简单集成。客户端程序通过HTTP协议进行检测,返回当前程序状态、版本ID以及已运行时间。 首先定义了两个变量,CommitHash、StartTime,然后定义结构体HeartbeatMessage封装返回值。 接着在init方法中给StartTime变量

    2023年04月09日
    浏览(28)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包