netty4 tcp与protobuf3的整合使用。

野性酷女 2022-06-14 19:52 212阅读 0赞

netty是一个非常棒的NIO框架就不介绍了。protobuf是google提供的一个开源序列化框架，也是一个非常棒的东西。这里不介绍这两个框架了，想要了解的朋友网上一搜一大把。

本文主要介绍 protobuf3与netty4 在tcp协议里的整合使用。个人认为netty与protobuf是绝配的组合。配合使用非常棒，框架提供了粘包拆包等工具类。特别是你要实现如及时通讯功能的时候，能让你省很多功夫。

使用protobuf需要事先预编译，这个过程这里就不介绍了，我的博客里面有，需要的可以看看。

首先在java中使用netty与protobuf需要两个jar包，我这里是：netty-all-4.1.0.CR7.jar、protobuf-java-3.1.0.jar（可以到maven下载，版本可以不同）

直接来看看代码吧。都有详细的注释：

package com.im.socket.netty.tcp;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelId;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.group.ChannelGroup;
    import io.netty.channel.group.DefaultChannelGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    import java.util.concurrent.ConcurrentMap;
    
    import org.jboss.netty.util.internal.ConcurrentHashMap;
    
    /**
     * 建立TCP netty服务
     * 
     * @author kokJuis
     * @version 1.0
     * @date 2016-9-30
     */
    public class ChatServer {
    
    	// 实例一个连接容器保存连接
    	public static ChannelGroup channels = new DefaultChannelGroup(
    			GlobalEventExecutor.INSTANCE);
    
    	// 再搞个map保存与用户的映射关系
    	public static ConcurrentMap<Integer, ChannelId> userSocketMap = new ConcurrentHashMap<Integer, ChannelId>();
    
    	private int port;
    
    	public ChatServer(int port) {
    		this.port = port;
    	}
    
    	public void run() throws Exception {
    
    		EventLoopGroup bossGroup = new NioEventLoopGroup();// boss线程池
    		EventLoopGroup workerGroup = new NioEventLoopGroup();// worker线程池
    		try {
    			ServerBootstrap b = new ServerBootstrap();
    			b.group(bossGroup, workerGroup)
    					.channel(NioServerSocketChannel.class)// 使用TCP
    					.childHandler(new ChatServerInitializer())// 初始化配置的处理器
    					.option(ChannelOption.SO_BACKLOG, 128)// BACKLOG用于构造服务端套接字ServerSocket对象，标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。如果未设置或所设置的值小于1，Java将使用默认值50。
    					.childOption(ChannelOption.SO_KEEPALIVE, true);// 是否启用心跳保活机制。在双方TCP套接字建立连接后（即都进入ESTABLISHED状态）并且在两个小时左右上层没有任何数据传输的情况下，这套机制才会被激活。
    
    			System.out.println("[ChatServer 启动了]");
    
    			// 绑定端口，开始接收进来的连接
    			ChannelFuture f = b.bind(port).sync();
    
    			// 等待服务器 socket 关闭 。
    			// 这不会发生，可以优雅地关闭服务器。
    			f.channel().closeFuture().sync();
    
    		} finally {
    			workerGroup.shutdownGracefully();
    			bossGroup.shutdownGracefully();
    
    			System.out.println("[ChatServer 关闭了]");
    		}
    	}
    
    }

package com.im.socket.netty.tcp;
    
    import java.util.concurrent.TimeUnit;
    
    import com.im.socket.protobuf.MessageOuterClass.Message;
    
    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.protobuf.ProtobufDecoder;
    import io.netty.handler.codec.protobuf.ProtobufEncoder;
    import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
    import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.handler.timeout.IdleStateHandler;
    import io.netty.util.CharsetUtil;
    import io.netty.util.HashedWheelTimer;
    import io.netty.util.Timer;
    
    /**
     * netty处理器配置
     * 
     * @author kokJuis
     * @version 1.0
     * @date 2016-9-30
     */
    public class ChatServerInitializer extends ChannelInitializer<SocketChannel> {
    
    	Timer timer;
    
    	public ChatServerInitializer() {
    		timer = new HashedWheelTimer();
    	}
    
    	@Override
    	public void initChannel(SocketChannel ch) throws Exception {
    		ChannelPipeline pipeline = ch.pipeline();
    
    		// ----Protobuf处理器，这里的配置是关键----
    		pipeline.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());// 用于decode前解决半包和粘包问题（利用包头中的包含数组长度来识别半包粘包）
    		//配置Protobuf解码处理器，消息接收到了就会自动解码，ProtobufDecoder是netty自带的，Message是自己定义的Protobuf类
    		pipeline.addLast("protobufDecoder",
    				new ProtobufDecoder(Message.getDefaultInstance()));
    		// 用于在序列化的字节数组前加上一个简单的包头，只包含序列化的字节长度。
    		pipeline.addLast("frameEncoder",
    				new ProtobufVarint32LengthFieldPrepender());
    		//配置Protobuf编码器，发送的消息会先经过编码
    		pipeline.addLast("protobufEncoder", new ProtobufEncoder());
    		// ----Protobuf处理器END----
    
    		pipeline.addLast("handler", new ChatServerHandler());//自己定义的消息处理器，接收消息会在这个类处理
    		pipeline.addLast("ackHandler", new AckServerHandler());//处理ACK
    		pipeline.addLast("timeout", new IdleStateHandler(100, 0, 0,
    				TimeUnit.SECONDS));// //此两项为添加心跳机制,60秒查看一次在线的客户端channel是否空闲
    		pipeline.addLast(new HeartBeatServerHandler());// 心跳处理handler
    
    	}
    
    	// pipeline.addLast("framer", new DelimiterBasedFrameDecoder(
    	// 2 * 1024, Delimiters.lineDelimiter()));
    	// pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
    	// pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
    }

package com.im.socket.netty.tcp;
    
    import java.util.Map.Entry;
    
    import com.im.socket.enums.EnumMessageType;
    import com.im.socket.protobuf.MessageOuterClass.Message;
    
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelId;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    /**
     * 消息处理类
     * 
     * @author kokJuis
     * @version 1.0
     * @date 2016-9-30
     */
    public class ChatServerHandler extends SimpleChannelInboundHandler<Message> {
    
    	@Override
    	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    		Channel incoming = ctx.channel();
    		System.out.println("[SERVER] - " + incoming.remoteAddress() + " 连接过来\n");
    		
    	}
    
    	@Override
    	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    		Channel incoming = ctx.channel();
    		ChatServer.channels.remove(incoming);
    
    		System.out.println("[SERVER] - " + incoming.remoteAddress() + " 离开\n");
    
    		// A closed Channel is automatically removed from ChannelGroup,
    		// so there is no need to do "channels.remove(ctx.channel());"
    	}
    
    	@Override
    	protected void channelRead0(ChannelHandlerContext ctx, Message msg)
    			throws Exception {
    		
    		
    		//消息会在这个方法接收到，msg就是经过解码器解码后得到的消息，框架自动帮你做好了粘包拆包和解码的工作
    		
    		//处理消息逻辑
    		
    		ctx.fireChannelRead(msg);//把消息交给下一个处理器
    		
    
    	}
    
    	@Override
    	public void channelActive(ChannelHandlerContext ctx) throws Exception { // (5)
    		Channel incoming = ctx.channel();
    		System.out.println("ChatClient:" + incoming.remoteAddress() + "上线");
    
    	}
    
    	@Override
    	public void channelInactive(ChannelHandlerContext ctx) throws Exception { // (6)
    		Channel incoming = ctx.channel();
    		System.out.println("ChatClient:" + incoming.remoteAddress() + "掉线");
    
    
    	}
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (7)
    		Channel incoming = ctx.channel();
    		// 当出现异常就关闭连接
    		System.out.println("ChatClient:" + incoming.remoteAddress()
    				+ "异常,已被服务器关闭");
    		cause.printStackTrace();
    		
    		ctx.close();
    	}
    
    }

package com.im.socket.netty.tcp;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    
    /**
     * 心跳检测
     * 
     * @ClassName HeartBeatServerHandler
     * @Description TODO
     * @author kokjuis 189155278@qq.com
     * @date 2016-9-26
     * @content
     */
    public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter {
    
    	private int loss_connect_time = 0;
    
    	@Override
    	public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
    			throws Exception {
    		if (evt instanceof IdleStateEvent) {
    			IdleStateEvent event = (IdleStateEvent) evt;
    			if (event.state() == IdleState.READER_IDLE) {
    				loss_connect_time++;
    				System.out.println("[60 秒没有接收到客户端" + ctx.channel().id()
    						+ "的信息了]");
    				if (loss_connect_time > 2) {
    					// 超过20秒没有心跳就关闭这个连接
    					System.out.println("[关闭这个不活跃的channel:" + ctx.channel().id()
    							+ "]");
    					ctx.channel().close();
    				}
    			}
    		} else {
    			super.userEventTriggered(ctx, evt);
    		}
    	}
    
    }