netty实现消息群发

netty是什么

我所理解的netty，是一个比较底层的网络编程的框架，它和tomcat的区别是什么呢？就是tomcat是一个已经封装好的容器，你可以直接使用，而netty是可以写出像tomcat这种容器的。而且tomcat支持的网络协议是http，但是使用netty，可以写出支持任何协议的容易。

当然，由于所学还不够深入，暂时使用netty实现简单的功能。网上很多大神对netty的分析都很深入很到位，优点也说了很多，简单来说，使用netty可以自己搭建并实现一套高性能的服务端通讯服务，自由灵活，可扩展性好，并能支撑较好的高并发性能，国内许多知名的互联网大厂，在一些通讯组件的底层上，选用的就是netty，比如阿里的,dubbo和rocketMq，读过源码的同学应该知道就是使用netty进行封装的，所以效率才会那么高；

实例分析

关于netty的一些基本概念和里面的一些常用组件的了解，建议可以先查阅相关资料进行学习，比如netty的编程模型，各种handler的理解，以及客户端与服务端建立连接时的channel的生命周期等概念，都是学习netty必要的基础知识；

下面有这样一个简单的需求，也是实际生活中会遇到的：用netty实现一个群聊的功能，每个用户上线和下线都能检测到，每个用户都可以发消息，每个用户发出去的消息都能被群里的其他用户接收到，下面用代码来简单实现一下，

显而易见，既然是群聊，必然有服务端和客户端，由于是后端代码，没有页面，我们将直接通过控制台发消息模拟效果即可，

服务端代码，在上一篇中我们简单分析了netty的编程特点，3个步骤，初始化服务端配置，加上一个自定义的handler即可，

1、服务端配置

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;/*** 服务端基本参数配置*/
public class ChatServer {private int port;public ChatServer(int port) {this.port = port;}public void startServer() throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChatServerInitializer()).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);System.out.println("Server 启动");ChannelFuture f = b.bind(port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();System.out.println("Server 关闭");}}public static void main(String[] args) throws Exception {int port;if (args.length > 0) {port = Integer.parseInt(args[0]);} else {port = 2333;}new ChatServer(port).startServer();}}

2、服务端自定义ChatServerInitializer

这部分的代码格式比较固定，可以这样理解，在客户端和服务端进行消息通讯的时候，两端都需要对进入channel的消息做一些前置的处理，比如编解码的操作，格式化输入输出校验，在netty中是通过诸如pipeline.addLast的代码实现的，也就是java中的一个责任链模式的具体实现，即把你需要的handler（处理具体某一种功能的代码块）加入到处理链中使其生效，比如这个里面最后添加了我们自定义的ChatServerHandler

public class ChatServerInitializer extends ChannelInitializer<SocketChannel> {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//以"\n"或者"\r\n"作为分隔符pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));pipeline.addLast("decoder", new StringDecoder());pipeline.addLast("encoder", new StringEncoder());pipeline.addLast("handler", new ChatServerHandler());System.out.println("ChatClient:" + ch.remoteAddress() + "  channel_id :" + ch.id() + "  连接上");}}

3、服务端自定义ChatServerHandler
ChatServerHandler是一个很重要的类，netty监控channel的生命周期就是在这个类里面完成的，下图可以看到，继承了SimpleChannelInboundHandler这个类之后有很多方法，在这些方法中，可以理解为netty管理具体的channel的生命周期的方法，可以根据业务的需要对这些方法进行重写，

见名知意，比如说channelRegistered（）这个方法，可以理解为当channel刚注册的时候方法被触发；channelActive（）当某个客户端连接进来的时候被触发；exceptionCaught()，当channel出现异常的时候被触发，不同的回调方法都有其特定的用途，在实际生产中都是可以派上用场的，

要特别说明的是，继承了上述的SimpleChannelInboundHandler这个类之后，有一个方法必须被复写，在4.X版本中，我这里显示的是messageReceived()，但在有一些版本中是channelRead()，但大家在学习的时候，一样的理解，里面的参数和功能是一样的，本人已经验证过了；

理解了上述的概念，下面直接贴上handler的代码，具体的使用都有备注

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;public class ChatServerHandler extends SimpleChannelInboundHandler<String> {/*** 管理channel的组，可以理解为channel的池*/public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);// 客户端的 Channel 存入 ChannelGroup列表中，并通知列表中的其他客户端 Channel@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {Channel currchannel = ctx.channel();channels.writeAndFlush("[服务器] - " + currchannel.remoteAddress() + " channel_id :" + currchannel.id() + " 加入\n");channels.add(ctx.channel());}//每当从服务端收到客户端断开时，客户端的 Channel 移除 ChannelGroup 列表中，并通知列表中的其他客户端 Channel@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {Channel currchannel = ctx.channel();channels.writeAndFlush("[服务器] - " + currchannel.remoteAddress() + " 离开\n");}//每当从服务端读到客户端写入信息时，将信息转发给其他客户端的 Channel。// 其中如果你使用的是 Netty 5.x 版本时，需要把 channelRead0() 重命名为messageReceived()@Overrideprotected void messageReceived(ChannelHandlerContext channelHandlerContext, String msg) {Channel incoming = channelHandlerContext.channel();for (Channel channel : channels) {//遍历ChannelGroup中的channelif (channel != incoming){//找到加入到ChannelGroup中的channel后，将录入的信息回写给除去发送信息的客户端channel.writeAndFlush("[" + incoming.remoteAddress() + "]" + msg + "\n");}else {channel.writeAndFlush("[服务器消息]" + msg + "\n");}}}//服务端监听到客户端活动@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception { // (5)Channel incoming = ctx.channel();System.out.println("服务器:" + incoming.remoteAddress() + "上线");}//服务端监听到客户端不活动@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception { // (6)Channel incoming = ctx.channel();System.out.println("服务器:" + incoming.remoteAddress() + "掉线");}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {Channel incoming = ctx.channel();System.out.println("服务器:" + incoming.remoteAddress() + "异常");ctx.close();}}

4、客户端配置
客户端的基本连接配置和服务端差不多，不过这里的不再是ServerBootstrap，因为它是服务端专用的，这里将会变成Bootstrap，其他的都差不多，绑定端口，加入自定义的handler即可，

public class SimpleChatClient {public static void main(String[] args) throws Exception{new SimpleChatClient("localhost", 2333).run();}private final String host;private final int port;public SimpleChatClient(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 ChatClientInitializer());Channel channel = bootstrap.connect(host, port).sync().channel();//录入信息BufferedReader in = new BufferedReader(new InputStreamReader(System.in));while(true){//死循环，模拟不断向服务端发送消息channel.writeAndFlush(in.readLine() + "\r\n");}} catch (Exception e) {e.printStackTrace();} finally {group.shutdownGracefully();}}
}

5、客户端ChatClientInitializer

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;public class ChatClientInitializer extends ChannelInitializer<SocketChannel> {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//使用netty自带的string类型的编解码器pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));pipeline.addLast("decoder", new StringDecoder());pipeline.addLast("encoder", new StringEncoder());pipeline.addLast("handler", new ChatClientHandler());}
}

6、客户端自定义handler

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;public class ChatClientHandler extends SimpleChannelInboundHandler<String> {@Overrideprotected void messageReceived(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {//客户端读取服务端发送回来的消息，只是进行显示System.out.println(msg);}}

代码的部分基本上就是这些，下面我们运行一下程序看一下效果，首先启动服务端，

启动第一个客户端，客户端启动完毕，在服务端的控制台打印如下信息，说明channel的相关回调方法生效了，

再启动一个客户端，有两个客户端的连接了，同时第一个客户端收到了一条消息，即被通知另一个客户端上线了，

某个客户端的控制台输入一条消息，由于服务端是转发，因此在另一个客户端那边，收到了hello world 这条消息，

到这里，我们用一段简单的程序演示了netty模拟消息群发的效果，感兴趣的同学可以继续做深入的研究，本篇到此结束，最后感谢观看！