Netty（七）之聊天室小小小案例

2023-02-23 64

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： Netty（七）之聊天室小小小案例

需求

1）上线或者下线给其它人员通知

2）A发送消息其它人员都可见

设计思路

客户端与服务端建立连接后会触发 serverHandler中的 channelActive 方法，把channel保存到ChannelGroup中，当客户端给服务端发送消息时，把channelGroup中的每一个channel都把消息发送一遍，就实现群发功能

代码实现(亲测可用)

pom

<dependencies>
        <!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.25.Final</version>
        </dependency>
    </dependencies>

MyChatServerHandler

package mychat;
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.util.concurrent.GlobalEventExecutor;
/**
 * @author CBeann
 * @create 2019-10-16 15:55
 */
public class MyChatServerHandler extends SimpleChannelInboundHandler<String> {
    //用一个ChannelGroup保存所有连接到服务器的客户端通道
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        Channel channel = channelHandlerContext.channel();
        //服务器收到消息
        // "[服务端]   " + channel.remoteAddress() + "通道关闭";
        String body = s;
        //群发
        channelGroup.forEach((x) -> {
            if (x != channel) {
                x.writeAndFlush(channel.remoteAddress() + "说===>" + s);
            } else {
                x.writeAndFlush("自己说===>" + s);
            }
        });
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        String notice = "[服务端]   " + channel.remoteAddress() + "通道激活";
        System.out.println(notice);
        channelGroup.writeAndFlush(notice);
        //添加建立连接的channel
        channelGroup.add(channel);
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        //删除失效的channel
        channelGroup.remove(channel);
        String notice = "[服务端]   " + channel.remoteAddress() + "通道关闭";
        channelGroup.writeAndFlush(notice);
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel channel = ctx.channel();
        System.out.println("[服务端]   " + channel.remoteAddress() + "出现异常");
        ctx.close();
    }
}

MyChatServer

package mychat;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
/**
 * @author CBeann
 * @create 2019-10-16 15:51
 */
public class MyChatServer {
    public static void main(String[] args) {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //TimeClientHandler是自己定义的方法
                            socketChannel.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
                            socketChannel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
                            socketChannel.pipeline().addLast(new MyChatServerHandler());
                        }
                    });
            //绑定端口
            ChannelFuture f = b.bind(8888).sync();
            //等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        } catch (
                Exception e) {
        } finally {
            //优雅关闭，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

MyChatClientHandler

package mychat;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
/**
 * @author CBeann
 * @create 2019-10-16 21:23
 */
public class MyChatClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        //收到服务端发送的消息
        String body = s;
        System.out.println(body);
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel channel = ctx.channel();
        System.out.println("[客户端出现异常");
        ctx.close();
    }
}

MyChatClient

package mychat;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;
import java.util.Scanner;
/**
 * @author CBeann
 * @create 2019-10-16 21:23
 */
public class MyChatClient {
    public static void main(String[] args) throws Exception {
        int port = 8888;
        String host = "127.0.0.1";
        //配置客户端NIO线程组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
                            socketChannel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
                            //TimeClientHandler是自己定义的方法
                            socketChannel.pipeline().addLast(new MyChatClientHandler());
                        }
                    });
            //发起异步连接操作
            ChannelFuture f = b.connect(host, port).sync();
//            //发送数据
            Scanner reader = new Scanner(System.in);
            String body = reader.nextLine();
            while (!"exit".equals(body)) {
                f.channel().writeAndFlush(body);
                body = reader.nextLine();
            }
            //等待客户端链路关闭
            f.channel().closeFuture().sync();
        } catch (Exception e) {
        } finally {
            //优雅关闭
            group.shutdownGracefully();
        }
    }
}

常见问题

1）IDEA怎么把一个启动类同时运行多次

先运行一下程序，在按照下面的操作进行