Java 聊天室基于Netty

原创

AnieaLanie

发布于 2021-12-11 19:38:40

1K0

发布于 2021-12-11 19:38:40

文章被收录于专栏：铁子的专栏铁子的专栏

1. Netty

1.1 pipeline

Netty在服务断端口绑定和新连接建立的过程中都会建立相应的channel，pipeline就像是一条流水线，被分为许多加工环节，字节流在流水线上加工。

//建立客户端
public static void main(String[] args) throws Exception {
        // 配置SSL证书
        final SslContext sslCtx = SslContextBuilder.forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE).build();

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    // 初始化pipeline
                    .handler(new SecureChatClientInitializer(sslCtx));

            // 启动连接尝试。
            Channel ch = b.connect(HOST, PORT).sync().channel();
            ...

//初始化pipeline
public class SecureChatClientInitializer extends ChannelInitializer<SocketChannel> {

    private final SslContext sslCtx;

    public SecureChatClientInitializer(SslContext sslCtx) {
        this.sslCtx = sslCtx;
    }

    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        //添加SSL处理程序，首先加密和解密一切。
        //在本例中，我们在服务器端使用了一个伪造的证书
        //在客户端接受任何无效的证书。
        //你需要一些更复杂的东西来识别两者
        //和服务器在现实世界。
        pipeline.addLast(sslCtx.newHandler(ch.alloc(), SecureChatClient.HOST, SecureChatClient.PORT));

        //在SSL处理程序的顶部，添加文本行编解码器。
        pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        pipeline.addLast(new StringDecoder());
        pipeline.addLast(new StringEncoder());

        //然后是业务逻辑。
        pipeline.addLast(new SecureChatClientHandler());
    }
}

1.2 Context

安全套接字协议实现SslContext，同时充当SSLEngine和SslHandler的工厂

final SslContext sslCtx = SslContextBuilder.forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE).build();

1.3 Bootstrap

Bootstrap 是 Netty 提供的一个便利的工厂类，它可以轻松引导一个Channel以供客户端使用。

Bootstrap b = new Bootstrap();
b.group(group)
        .channel(NioSocketChannel.class)
        .handler(new SecureChatClientInitializer(sslCtx));

与之相对应的ServerBootstrap是供服务器用的Channel引导类：

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .handler(new LoggingHandler(LogLevel.INFO))
        .childHandler(new SecureChatServerInitializer(sslCtx));

1.4 Channel

Channel是一个与网络套接字的连接，或一个能够进行读、写、连接和绑定等I/O操作的组件。

Channel为用户提供：

通道的当前状态（例如它是否打开？是否已连接？），
通道的配置参数（例如接收缓冲区大小），
通道支持的 I/O 操作（例如读取、写入、连接和绑定），以及
在ChannelPipeline该处理与通道相关联的所有I / O事件和请求。

1.5 Channel的层级

Channel是分层级的，一个Channel可以有一个父级，这取决于Channel的创建方式。比如，一个被ServerSocketChannel接收的SocketChannel将返回一个SocketChannel，这个SocketChannel将以ServerSocketChannel作为父级。例如，可以编写一个新的Channel来创建共享一个套接字连接的子通道。

Channel的操作完成后，完成后调用ChannelOutboundInvoker.close()或ChannelOutboundInvoker.close(ChannelPromise)释放所有资源很重要Channel。这确保所有资源 (文件句柄) 都以正确的方式释放。

1.6 ChannelPipeline

一组ChannelHandler负责处理或拦截一个Channel的入站事件和出站操作。ChannelPipeline实现了拦截过滤器模式的高级形式，使用户可以完全控制事件的处理方式以及控制管道中的这组ChannelHandler之间如何相互交互。

1.7 BossEventLoopGroup和WorkerEventLoopGroup

事件循环组，一旦注册， EventLoop将处理Channel的所有IO操作。而 ServerBootstrap 需要两种类型的EventLoopGroup，即 “boss” 和 “boss”。

BossEventLoopGroup：这个事件循环组寻找并接受传入的连接。
WorkerEventLoopGroup：Boss接受连接并将其注册到worker循环组。worker通过该连接处理通信器件所有的事件。

2. Java聊天室编码

2.1 服务端代码

package JavaIOTest.NettyChatRoom;

import io.netty.bootstrap.ServerBootstrap;
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 io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;

/**
 * Simple SSL chat server modified from {@link TelnetServer}.
 */
public final class SecureChatServer {

    static final int PORT = Integer.parseInt(System.getProperty("port", "8992"));

    public static void main(String[] args) throws Exception {
        SelfSignedCertificate ssc = new SelfSignedCertificate();
        SslContext sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey())
                .build();

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new SecureChatServerInitializer(sslCtx));

            b.bind(PORT).sync().channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

2.2 服务端初始化

package JavaIOTest.NettyChatRoom;


import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.ssl.SslContext;

/**
 * Creates a newly configured {@link ChannelPipeline} for a new channel.
 */
public class SecureChatServerInitializer extends ChannelInitializer<SocketChannel> {

    private final SslContext sslCtx;

    public SecureChatServerInitializer(SslContext sslCtx) {
        this.sslCtx = sslCtx;
    }

    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        //添加SSL处理程序，首先加密和解密一切。
        //在本例中，我们在服务器端使用了一个伪造的证书
        //在客户端接受任何无效的证书。
        //你需要一些更复杂的东西来识别两者
        //和服务器在现实世界。
        pipeline.addLast(sslCtx.newHandler(ch.alloc()));

        //在SSL处理程序的顶部，添加文本行编解码器。
        pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        pipeline.addLast(new StringDecoder());
        pipeline.addLast(new StringEncoder());

        //然后是业务逻辑。
        pipeline.addLast(new SecureChatServerHandler());
    }
}

2.3 服务端事件处理

package JavaIOTest.NettyChatRoom;


import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.net.InetAddress;

/**
 * 处理服务端通道
 */
public class SecureChatServerHandler extends SimpleChannelInboundHandler<String> {

    static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    public void channelActive(final ChannelHandlerContext ctx) {
        //一旦会话安全，发送问候并注册通道到全局通道
        //列表以便通道从其他接收消息。
        ctx.pipeline().get(SslHandler.class).handshakeFuture().addListener(
                new GenericFutureListener<Future<Channel>>() {
                    @Override
                    public void operationComplete(Future<Channel> future) throws Exception {
                        ctx.writeAndFlush(
                                "Welcome to " + InetAddress.getLocalHost().getHostName() + " secure chat service!\n");
                        ctx.writeAndFlush(
                                "Your session is protected by " +
                                        ctx.pipeline().get(SslHandler.class).engine().getSession().getCipherSuite() +
                                        " cipher suite.\n");

                        channels.add(ctx.channel());
                    }
                });
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        //将接收到的消息发送到除当前通道外的所有通道。
        for (Channel c: channels) {
            if (c != ctx.channel()) {
                c.writeAndFlush("[" + ctx.channel().remoteAddress() + "] " + msg + '\n');
            } else {
                c.writeAndFlush("[you] " + msg + '\n');
            }
        }

        //如果客户端发送了'bye'，请关闭连接。
        if ("bye".equals(msg.toLowerCase())) {
            ctx.close();
        }
    }

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

2.4 客户端代码

package JavaIOTest.NettyChatRoom;


import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;

import java.io.BufferedReader;
import java.io.InputStreamReader;

/**
 * Simple SSL chat client
 */
public final class SecureChatClient {

    static final String HOST = System.getProperty("host", "127.0.0.1");
    static final int PORT = Integer.parseInt(System.getProperty("port", "8992"));

    public static void main(String[] args) throws Exception {
        // 配置SSL证书
        final SslContext sslCtx = SslContextBuilder.forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE).build();

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new SecureChatClientInitializer(sslCtx));

            // 启动连接尝试。
            Channel ch = b.connect(HOST, PORT).sync().channel();

            // 从stdin中读取命令。
            ChannelFuture lastWriteFuture = null;
            BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
            for (;;) {
                String line = in.readLine();
                if (line == null) {
                    break;
                }

                // 将接收到的行发送到服务器。
                lastWriteFuture = ch.writeAndFlush(line + "\r\n");

                //如果用户输入了'bye'命令，请等待服务器关闭
                // 连接。
                if ("bye".equals(line.toLowerCase())) {
                    ch.closeFuture().sync();
                    break;
                }
            }

            //等待所有消息都刷新后才关闭通道。
            if (lastWriteFuture != null) {
                lastWriteFuture.sync();
            }
        } finally {
            //连接在关闭时自动关闭。
            group.shutdownGracefully();
        }
    }
}

2.5 客户端初始化

package JavaIOTest.NettyChatRoom;


import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.ssl.SslContext;

/**
 * Creates a newly configured {@link ChannelPipeline} for a new channel.
 */
public class SecureChatClientInitializer extends ChannelInitializer<SocketChannel> {

    private final SslContext sslCtx;

    public SecureChatClientInitializer(SslContext sslCtx) {
        this.sslCtx = sslCtx;
    }

    @Override
    public void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        //添加SSL处理程序，首先加密和解密一切。
        //在本例中，我们在服务器端使用了一个伪造的证书
        //在客户端接受任何无效的证书。
        //你需要一些更复杂的东西来识别两者
        //和服务器在现实世界。
        pipeline.addLast(sslCtx.newHandler(ch.alloc(), SecureChatClient.HOST, SecureChatClient.PORT));

        //在SSL处理程序的顶部，添加文本行编解码器。
        pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
        pipeline.addLast(new StringDecoder());
        pipeline.addLast(new StringEncoder());

        //然后是业务逻辑。
        pipeline.addLast(new SecureChatClientHandler());
    }
}

2.6 客户端事件处理

package JavaIOTest.NettyChatRoom;


import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * 处理客户端通道。
 */
public class SecureChatClientHandler extends SimpleChannelInboundHandler<String> {

    @Override
    public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.err.println(msg);
    }

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

通信效果：

客戶端1发送hello并接受hi：

客戶端2接受hello并发送hi：

3. 参考

[1] springboot和netty整合的聊天室--群聊

[2] Netty安全聊天

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

java

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

java

登录后参与评论

0 条评论

热度