一、群聊系统
实例要求:
1、编写?个 Netty 群聊系统,实现服务器端和客户端之间的数据简单通讯(?阻塞)
2、实现多?群聊
3、服务器端:可以监测?户上线,离线,并实现消息转发功能
4、客户端:通过 channel 可以?阻塞发送消息给其它所有?户,同时可以接受其它?户发送的消息(有服务器转发得到)
5、?的:进?步理解 Netty ?阻塞?络编程机制
1、服务端代码
这里主要就是启动一个ServerBootstrap,然后配置各类信息,主要就是绑定一个handler。
public class GroupChatServer { private int port; //监听端口 public GroupChatServer(int port) { this.port = port; } //编写run方法,处理客户端的请求 public void run() throws Exception { //创建两个线程组 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop try { ServerBootstrap b = new ServerBootstrap(); b.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 ch) throws Exception { //获取到pipeline ChannelPipeline pipeline = ch.pipeline(); //向pipeline加入解码器 pipeline.addLast("decoder", new StringDecoder()); //向pipeline加入编码器 pipeline.addLast("encoder", new StringEncoder()); //加入自己的业务处理handler pipeline.addLast(new GroupChatServerHandler()); } }); System.out.println("netty 服务器启动"); ChannelFuture channelFuture = b.bind(port).sync(); //监听关闭 channelFuture.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { new GroupChatServer(7000).run(); } }
服务端Handler:
代码逻辑:继承SimpleChannelInboundHandler,重写其中的方法,其中channelActive和channelInactive分别表示channel的活动状态,是活动还是不活动,那么这两个方法中,直接在服务器中显示上线和下线即可。handlerAdded和handlerRemoved表示当前channel是加入channelGroup还是离开channelGroup,那么就可以使用channelGroup的writeAndFlush刷新到客户端。
public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> { //public static List<Channel> channels = new ArrayList<Channel>(); //使用一个hashmap 管理 //public static Map<String, Channel> channels = new HashMap<String,Channel>(); //定义一个channle 组,管理所有的channel //GlobalEventExecutor.INSTANCE) 是全局的事件执行器,是一个单例 private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); //handlerAdded 表示连接建立,一旦连接,第一个被执行 //将当前channel 加入到 channelGroup @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); //将该客户加入聊天的信息推送给其它在线的客户端 /* 该方法会将 channelGroup 中所有的channel 遍历,并发送消息, 我们不需要自己遍历 */ channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 加入聊天" + sdf.format(new java.util.Date()) + " \n"); channelGroup.add(channel); } //断开连接, 将xx客户离开信息推送给当前在线的客户 @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 离开了\n"); System.out.println("channelGroup size" + channelGroup.size()); } //表示channel 处于活动状态, 提示 xx上线 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println(ctx.channel().remoteAddress() + " 上线了~"); } //表示channel 处于不活动状态, 提示 xx离线了 @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { System.out.println(ctx.channel().remoteAddress() + " 离线了~"); } //读取数据 @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { //获取到当前channel Channel channel = ctx.channel(); //这时我们遍历channelGroup, 根据不同的情况,回送不同的消息 channelGroup.forEach(ch -> { if (channel != ch) { //不是当前的channel,转发消息 ch.writeAndFlush("[客户]" + channel.remoteAddress() + " 发送了消息" + msg + "\n"); } else {//回显自己发送的消息给自己 ch.writeAndFlush("[自己]发送了消息" + msg + "\n"); } }); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //关闭通道 ctx.close(); } }
客户端:
这里没什么可说的,就是创建并启动一个Bootstrap,对其绑定一个handler。
public class GroupChatClient { //属性 private final String host; private final int port; public GroupChatClient(String host, int port) { this.host = host; this.port = port; } public void run() throws Exception { EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap() .group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { //得到pipeline ChannelPipeline pipeline = ch.pipeline(); //加入相关handler pipeline.addLast("decoder", new StringDecoder()); pipeline.addLast("encoder", new StringEncoder()); //加入自定义的handler pipeline.addLast(new GroupChatClientHandler()); } }); ChannelFuture channelFuture = bootstrap.connect(host, port).sync(); //得到channel Channel channel = channelFuture.channel(); System.out.println("-------" + channel.localAddress() + "--------"); //客户端需要输入信息,创建一个扫描器 Scanner scanner = new Scanner(System.in); while (scanner.hasNextLine()) { String msg = scanner.nextLine(); //通过channel 发送到服务器端 channel.writeAndFlush(msg + "\r\n"); } } finally { group.shutdownGracefully(); } } public static void main(String[] args) throws Exception { new GroupChatClient("127.0.0.1", 7000).run(); } }
客户端handler:
这里继承了SimpleChannelInboundHandler,重新了channelRead0方法,直接输出了服务端的msg。
public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> { @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { System.out.println(msg.trim()); } }
二、?跳检测机制案例
当 IdleStateEvent 触发后 , 就会传递给管道的下一个handler去处理,由后续的handler来决定应该怎么处理。
public class MyHeartBeatServer { public static void main(String[] args) throws Exception { //创建两个线程组 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop try { ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup, workerGroup); serverBootstrap.channel(NioServerSocketChannel.class); serverBootstrap.handler(new LoggingHandler(LogLevel.INFO)); serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //加入一个netty 提供 IdleStateHandler /* 说明 1. IdleStateHandler 是netty 提供的处理空闲状态的处理器 2. long readerIdleTime : 表示多长时间没有读, 就会发送一个心跳检测包检测是否连接 3. long writerIdleTime : 表示多长时间没有写, 就会发送一个心跳检测包检测是否连接 4. long allIdleTime : 表示多长时间没有读写, 就会发送一个心跳检测包检测是否连接 5. 文档说明 triggers an {@link IdleStateEvent} when a {@link Channel} has not performed * read, write, or both operation for a while. * 6. 当 IdleStateEvent 触发后 , 就会传递给管道 的下一个handler去处理 * 通过调用(触发)下一个handler 的 userEventTiggered , 在该方法中去处理 IdleStateEvent(读空闲,写空闲,读写空闲) */ pipeline.addLast(new IdleStateHandler(7000, 7000, 10, TimeUnit.SECONDS)); //加入一个对空闲检测进一步处理的handler(自定义) pipeline.addLast(new MyServerHandler()); } }); //启动服务器 ChannelFuture channelFuture = serverBootstrap.bind(7000).sync(); channelFuture.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }
2、服务端handler
当 IdleStateEvent 触发后 , 就会传递给管道的下一个handler去处理,由后续的handler来决定应该怎么处理。而handler实际的处理就是断开连接。
public class MyServerHandler extends ChannelInboundHandlerAdapter { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { //将 evt 向下转型 IdleStateEvent IdleStateEvent event = (IdleStateEvent) evt; String eventType = null; switch (event.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("服务器做相应处理.."); //如果发生空闲,我们关闭通道 // ctx.channel().close(); } } }
三、Netty 通过 WebSocket 编程实现服务器和客户端?连接
public class MyWebSocketServer { public static void main(String[] args) throws Exception { //创建两个线程组 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop try { ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup, workerGroup); serverBootstrap.channel(NioServerSocketChannel.class); serverBootstrap.handler(new LoggingHandler(LogLevel.INFO)); serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //因为基于http协议,使用http的编码和解码器 pipeline.addLast(new HttpServerCodec()); //是以块方式写,添加ChunkedWriteHandler处理器 pipeline.addLast(new ChunkedWriteHandler()); /* 说明 1. http数据在传输过程中是分段, HttpObjectAggregator ,就是可以将多个段聚合 2. 这就就是为什么,当浏览器发送大量数据时,就会发出多次http请求 */ pipeline.addLast(new HttpObjectAggregator(8192)); /* 说明 1. 对应websocket ,它的数据是以 帧(frame) 形式传递 2. 可以看到WebSocketFrame 下面有六个子类 3. 浏览器请求时 ws://localhost:7000/hello 表示请求的uri 4. WebSocketServerProtocolHandler 核心功能是将 http协议升级为 ws协议 , 保持长连接 5. 是通过一个 状态码 101 */ pipeline.addLast(new WebSocketServerProtocolHandler("/hello2")); //自定义的handler ,处理业务逻辑 pipeline.addLast(new MyTextWebSocketFrameHandler()); } }); //启动服务器 ChannelFuture channelFuture = serverBootstrap.bind(7000).sync(); channelFuture.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }
2、handler
public class MyTextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { @Override protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { System.out.println("服务器收到消息 " + msg.text()); //回复消息 ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间" + LocalDateTime.now() + " " + msg.text())); } //当web客户端连接后, 触发方法 @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { //id 表示唯一的值,LongText 是唯一的 ShortText 不是唯一 System.out.println("handlerAdded 被调用" + ctx.channel().id().asLongText()); System.out.println("handlerAdded 被调用" + ctx.channel().id().asShortText()); } @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { System.out.println("handlerRemoved 被调用" + ctx.channel().id().asLongText()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println("异常发生 " + cause.getMessage()); ctx.close(); //关闭连接 } }