栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 软件开发 > 后端开发 > Java

基于netty+springboot+websocket+thymeleaf实现聊天通讯

Java 更新时间: 发布时间: IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

基于netty+springboot+websocket+thymeleaf实现聊天通讯

一、网络IO的基本知识和概念 1、同步、异步、阻塞、非阻塞概念
  • 同步和异步
    同步和异步是针对应用程序和内核的交互而言的,同步指的是用户进程触发IO 操作并等待或者轮询的去查看IO 操作是否就绪,而异步是指用户进程触发IO 操作以后便开始做自己的事情,而当IO 操作已经完成的时候会得到IO 完成的通知。
  • 阻塞和非阻塞
    阻塞和非阻塞是针对于进程在访问数据的时候,根据IO操作的就绪状态来采取的不同方式,说白了是一种读取或者写入操作方法的实现方式,阻塞方式下读取或者写入函数将一直等待,而非阻塞方式下,读取或者写入方法会立即返回一个状态值。

总结: 同步异步与阻塞非阻塞的主要区别是针对对象不同。

同步异步是针对调用者来说的。
同步与异步:针对数据访问的方式,程序是主动去询问操作系统数据准备好了么,还是操作系统在数据准备好的时候通知程序。
阻塞非阻塞是针对被调用者来说的。
阻塞与非阻塞:针对函数(程序)运行的方式,在IO未就绪时,是等待就绪还是直接返回(执行别的操作)。比如: 被调用者收到一个请求后,做完请求任务后才给出反馈就是阻塞,收到请求直接给出反馈再去做任务就是非阻塞。

IO多路复用是同步阻塞模型还是异步阻塞模型?
同步是需要主动等待消息通知,而异步则是被动接收消息通知,通过回调、通知、状态等方式来被动获取消息。IO多路复用在阻塞到select阶段时,用户进程是主动等待并调用select函数获取数据就绪状态消息,并且其进程状态为阻塞。所以,把IO多路复用归为同步阻塞模式。

2、IO模型 阻塞IO

如果数据没有准备就绪,就一直等待,直到数据准备就绪;整个进程会被阻塞。

典型的阻塞IO模型的例子为: data = socket.read();如果数据没有就绪,就会一直阻塞在 read()方法。

非阻塞IO

需不断询问内核是否已经准备好数据,非阻塞虽然不用等待但是一直占用CPU。

典型的非阻塞IO模型一般如下:

while(true){
data = socket.read();
if(data != error){
// 处理数据 break;
}
}

非阻塞IO又一个非常严重的问题,在while 循环中需要不断的去询问内核数据是否就绪,这样会导致CPU占用率非常高,因此一般情况下很少使用while循环这种方式来读取数据。

多路复用IO/NIO

多路复用IO模型是目前使用的比较多的模型。Java NIO实际上就是多路复用IO,在多路复用IO模型中会有一个线程不断去轮询多个socket的状态,只有当socket真正有读写事件时,才真正调用实际的IO读写操作。

在Java NIO中,是通过 selector.select()去查询每个通道是否有达到事件,如果没有事件,则一直阻塞在那里,因此这种方式会导致用户线程的阻塞。

多路复用IO为何比非阻塞IO模型的效率高?是因为在非阻塞IO中,不断的询问socket状态时通过用户线程去进行的,而在多路复用IO中,轮询每个socket状态是内核在进行的,这个效率要比用户线程要高的多。

信号驱动IO模型

在信号驱动IO模型中,当用户线程发起一个IO请求操作,会给对应的socket注册一个信号函数,然后用户线程会继续执行,当内核数据就绪时会发送一个信号给用户线程,用户线程接收到信号之后,便在信号函数中调用IO读写操作来进行实际的IO请求操作。

异步IO模型(asynchronous I/O)

异步IO模型才是最理想的IO模型,在异步IO模型中,当用户线程发起read操作之后,立刻就可以开始去做其它的事情。而另一方面,从内核的角度,当它受到一个asynchronout read之后,它会立刻返回,说明read请求已经成功发起来,因此不会对用户线程产生任何block。然后,内核会等待数据准备完成,然后将数据拷贝到用户线程,当这一切都完成之后,内核会给用户线程发送一个信号,告诉它read操作完成了。
也就是说用户线程完全不需要实际的整个IO操作是如何进行的,只需要发起一个请求,当接收内核返回的成功信号时表示IO操作已完成,可以直接去使用数据了。也就是说在异步IO模型中,IO操作的两个阶段都不会阻塞用户线程,这两个阶段都是由内核自动完成,然后发送一个信号告知用户线程操作已完成。用户线程中不需要再次调用IO函数进行具体的读写。这点是和信号驱动模型有所不同的,在信号驱动模型中,当用户线程接收到信号表示数据已经就绪,然后需要用户线程调用IO函数进行实际的读写操作;而在异步IO模型中,收到信号表示IO操作已经完成,不需要再在用户线程中调用IO函数进行实际的读写操作。注意,异步IO是需要操作系统的底层支持,在Java 7中,提供了Asynchronous IO。

3、NIO 的工作流程


工作原理核心就2个概念 ,一个是 Selector(选择器),一个是通道Channel

  • Selector(选择器)
    其中select 调用可能是阻塞的,也可以是非阻塞的。但是read/write是非阻塞的!

源码关键字: Selector、SelectableChannel

  • 通道Channel
    主要实现类:
    FileChannel:用于读取、写入、映射和操作文件的通道。
    DatagramChannel:通过UDP读写网络中的数据通道。
    SocketChannel:通过tcp读写网络中的数据。
    ServerSocketChannel:可以监听新进来的tcp连接,对每一个连接都创建一个SocketChannel。
二、netty

Netty 是由 JBOSS 提供的一个 Java 开源框架。Netty 提供异步的、基于事件驱动的网络应用程序框架,用以快速开发高性能、高可靠性的网络 IO 程序,是目前最流行的 NIO 框架,Netty 在互联网领域、大数据分布式计算领域、游戏行业、通信行业等获得了广泛的应用,知名的 Elasticsearch 、Dubbo 框架内部都采用了 Netty。

三、demo


1、netty服务端

package com.yy.nettydemo.server;

import com.yy.nettydemo.config.WebSocketChannelConfig;
import com.yy.nettydemo.eneity.Netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;


@Component
@Slf4j
public class NettyWebSocketServer implements Runnable {

    @Autowired
    private Netty netty;

    @Autowired
    private WebSocketChannelConfig webSocketChannelConfig;

    
    private EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    
    private EventLoopGroup workerGroup = new NioEventLoopGroup();

    
    @PreDestroy
    public void close() {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

    @Override
    public void run() {
        try {
            //创建服务端启动助手
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup);
            serverBootstrap.channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.DEBUG))
                    .childHandler(webSocketChannelConfig);
            //启动
            ChannelFuture channelFuture = serverBootstrap.bind(netty.getPort()).sync();
            log.info("【-----Netty服务端启动成功-----】");
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

2、netty实体

package com.yy.nettydemo.eneity;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;


@Component
@ConfigurationProperties(prefix = "netty")
@Data
public class Netty {
    
    private int port;
    
    private String path;
}

3、netty初始化

package com.yy.nettydemo.config;


import com.yy.nettydemo.eneity.Netty;
import com.yy.nettydemo.handler.WebSocketHandler;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;


@Component
public class WebSocketChannelConfig extends ChannelInitializer {

    @Resource
    private Netty netty;
    @Resource
    private WebSocketHandler webSocketHandler;

    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        // 对http协议的支持.
        pipeline.addLast(new HttpServerCodec());
        // 对大数据流的支持,因为http通信事以块方式写,添加ChunkedWriteHandler处理器
        pipeline.addLast(new ChunkedWriteHandler());
        // HttpObjectAggregator将多个信息转化成单一的request或者response对象
        
        pipeline.addLast(new HttpObjectAggregator(8000));
        // 将http协议升级为ws协议. 对websocket的支持,他的数据是以帧(frame)形式传递
        //2:可以看到webSocketFrame下面有六个子类
        //3:浏览器请求时 ws://localhost:8088/netty 表示请求的uri
        //4: WebSocketClientProtocolHandler 核心功能是将http协议升级为ws协议,保持长连接
        pipeline.addLast(new WebSocketServerProtocolHandler(netty.getPath()));
        // 自定义处理handler
        pipeline.addLast(webSocketHandler);
    }
}

4、处理器

package com.yy.nettydemo.handler;



import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
@Component
@Slf4j

@ChannelHandler.Sharable
public class WebSocketHandler extends SimpleChannelInboundHandler {
    public static List channelList = new ArrayList<>();

    
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        //当有新的客户端连接的时候, 将通道放入集合
        channelList.add(channel);
        log.info("有新的连接加入。。。");
    }

    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        //当有客户端断开连接的时候,就移除对应的通道
        channelList.remove(channel);
        log.info("有连接已经断开。。。");
    }

    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception {
        String msg = textWebSocketFrame.text();
        log.info("msg:{}", msg);
        //当前发送消息的通道, 当前发送的客户端连接
        Channel channel = ctx.channel();
        for (Channel channel1 : channelList) {
            //排除自身通道
            if (channel != channel1) {
                channel1.writeAndFlush(new TextWebSocketFrame(msg));
            }
        }
    }

    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        Channel channel = ctx.channel();
        //移除集合
        channelList.remove(channel);
    }
}

5、启动类

package com.yy.nettydemo;

import com.yy.nettydemo.server.NettyWebSocketServer;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import javax.annotation.Resource;

@SpringBootApplication
public class NettyDemoApplication implements CommandLineRunner {

    @Resource
    private NettyWebSocketServer nettyWebSocketServer;

    public static void main(String[] args) {
        SpringApplication.run(NettyDemoApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception{
        new Thread(nettyWebSocketServer).start();
    }
}

6、pom



    4.0.0
    
        org.springframework.boot
        spring-boot-starter-parent
        2.1.6.RELEASE
         
    
    com.yy
    netty-demo
    0.0.1-SNAPSHOT
    netty-demo
    Demo project for Spring Boot
    
        1.8
    
    
        
            org.springframework.boot
            spring-boot-starter-web
        
        
            org.springframework.boot
            spring-boot-devtools
            runtime
            true
        

        
            org.springframework.boot
            spring-boot-starter-test
            test
        

        
            io.netty
            netty-all
            4.1.36.Final
        

        
            org.springframework.boot
            spring-boot-starter-thymeleaf
        

        
        
            com.alibaba
            fastjson
            1.2.83
        

        
            org.projectlombok
            lombok
        

        
            commons-lang
            commons-lang
            2.3
        

        
            com.github.pagehelper
            pagehelper
            5.3.0
        
    

    
        
            
                org.springframework.boot
                spring-boot-maven-plugin
            
        
    



7、配置文件

server:
  port: 9090
netty:
  port: 9091
  path: /chat
spring:
  thymeleaf:
    cache: false
    checktemplatelocation: true
    enabled: true
    encoding: UTF-8
    mode: HTML5
    prefix: classpath:/templates/
    suffix: .html
  resources:
    static-locations: classpath:/templates/


8、前端chat.html





    
    
    聊天室
    
    
    



9、测试

转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/1007806.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号