Netty实现长连接简单例子

た入场券 2022-06-15 07:40 665阅读 0赞

# **Netty实现长连接的简单例子** #

# **服务端** #

NettyServerBootstrap提供长连接服务

public class NettyServerBootstrap {
 
 	private static final Log log = LogFactory.getLog(NettyServerBootstrap.class);
 	private Integer port;
 	private SocketChannel socketChannel;
 	public NettyServerBootstrap(Integer port) throws Exception {
 		this.port = port;
 		bind(port);
 	}
 	public Integer getPort() {
 		return port;
 	}
 	public void setPort(Integer port) {
 		this.port = port;
 	}
 	public SocketChannel getSocketChannel() {
 		return socketChannel;
 	}
 	public void setSocketChannel(SocketChannel socketChannel) {
 		this.socketChannel = socketChannel;
 	}
 	private void bind(int serverPort) throws Exception {
 		// 连接处理group
 		EventLoopGroup boss = new NioEventLoopGroup();
 		// 事件处理group
 		EventLoopGroup worker = new NioEventLoopGroup();
 		ServerBootstrap bootstrap = new ServerBootstrap();
 		// 绑定处理group
 		bootstrap.group(boss, worker);
 		bootstrap.channel(NioServerSocketChannel.class);
 		// 保持连接数
 		bootstrap.option(ChannelOption.SO_BACKLOG, 1024 * 1024);
 		// 有数据立即发送
 		bootstrap.option(ChannelOption.TCP_NODELAY, true);
 		// 保持连接
 		bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
 // 处理新连接
 		bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
 			@Override
 			protected void initChannel(SocketChannel sc) throws Exception {
 				// 增加任务处理
 				ChannelPipeline p = sc.pipeline();
 				p.addLast(new MessageDecoder(), new MessageEncoder(), new NettyServerHandler());
 			 }
 		});
 
 		ChannelFuture f = bootstrap.bind(serverPort).sync();
 		if (f.isSuccess()) {
 			log.info("long connection started success");
 		} else {
 			log.error("long connection started fail");
 		}
 	}
 }

启动服务，监听9999端口

public static void main(String[] args) {
    		try {
    			new NettyServerBootstrap(9999);
    		} catch (Exception e) {
    			
    			e.printStackTrace();
    		}
    }

定义客户端服务端通信协议，一下是一个简单的通信协议

<table> 
 <tbody> 
 <tr> 
 <td> 协议分为header和body两部分，都是用网络字节序（BIG ENDIAN） header{ magic 32bit; &nbsp;//校验用固定值0x0CAFFEE0 version 8bit; &nbsp;//版本号 type 8bit; &nbsp;&nbsp;&nbsp;&nbsp;//类型，请求或者响应 seq 32bit; &nbsp;&nbsp;&nbsp;&nbsp;//序号标记一对请求响应 length 32bit; &nbsp;//body长度 } body{ &nbsp; } </td> 
 </tr> 
 </tbody> 
</table>

根据通信协议，编写解码器和编码器

解码器MessageDecoder

public class MessageDecoder extends ByteToMessageDecoder {
 	private static final int MAGIC_NUMBER = 0x0CAFFEE0;
 	public MessageDecoder() {
 
 	}
 	@Override
 	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
 		if (in.readableBytes() < 14) {
 			return;
 		}
 		// 标记开始读取位置
 		in.markReaderIndex();
 
 		int magic_number = in.readInt();
 
 		if (MAGIC_NUMBER != magic_number) {
 			ctx.close();
 			return;
 		}
 
 		@SuppressWarnings("unused")
 		byte version = in.readByte();
 
 		byte type = in.readByte();
 		int squence = in.readInt();
 		int length = in.readInt();
 
 		if (length < 0) {
 			ctx.close();
 			return;
 		}
 
 		if (in.readableBytes() < length) {
 			// 重置到开始读取位置
 			in.resetReaderIndex();
 			return;
 		}
 
 		byte[] body = new byte[length];
 		in.readBytes(body);
 
 		RequestInfoVO req = new RequestInfoVO();
 		req.setBody(new String(body, "utf-8"));
 		req.setType(type);
 		req.setSequence(squence);
 		out.add(req);
 	}
 }

编码器MessageEncoder

public class MessageEncoder extends MessageToByteEncoder<RequestInfoVO> {
 
 	private static final String DEFAULT_ENCODE = "utf-8";
 
 	private static final int MAGIC_NUMBER = 0x0CAFFEE0;
 
 	public MessageEncoder() {
 	}
 
 	@Override
 	protected void encode(ChannelHandlerContext ctx, RequestInfoVO msg, ByteBuf out) throws Exception {
 
 		@SuppressWarnings("resource")
 		ByteBufOutputStream writer = new ByteBufOutputStream(out);
 		byte[] body = null;
 
 		if (null != msg && null != msg.getBody() && "" != msg.getBody()) {
 			body = msg.getBody().getBytes(DEFAULT_ENCODE);
 		}
 
 		writer.writeInt(MAGIC_NUMBER);
 
 		writer.writeByte(1);
 
 		writer.writeByte(msg.getType());
 
 		writer.writeInt(msg.getSequence());
 
 		if (null == body || 0 == body.length) {
 			writer.writeInt(0);
 		} else {
 			writer.writeInt(body.length);
 			writer.write(body);
 		}
 	}
 
 }

服务端事件处理NettyServerHandler

@Sharable
 public class NettyServerHandler extends SimpleChannelInboundHandler<RequestInfoVO> {
 	private static final Log log = LogFactory.getLog(NettyServerHandler.class);
 
 	@Override
 	protected void channelRead0(ChannelHandlerContext ctx, RequestInfoVO msg) throws Exception {
 		//
 	}
 
 	@Override
 	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 		
 	}
 
 	@Override
 	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
 		
 	}
 
 
 	@Override
 	public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
 		
 	}
 
 }

# **客户端** #

NettyClientBootstrap连接长连接服务

public class NettyClientBootstrap {
 	private int port;
 	private String host;
 	private SocketChannel socketChannel;
 	public NettyClientBootstrap(int port, String host) throws Exception {
 		this.host = host;
 		this.port = port;
 		start();
 	}
 	private void start() throws Exception {
 		EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
 		Bootstrap bootstrap = new Bootstrap();
 		bootstrap.channel(NioSocketChannel.class);
 		bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
 		bootstrap.option(ChannelOption.TCP_NODELAY, true);
 		bootstrap.group(eventLoopGroup);
 		bootstrap.remoteAddress(this.host, this.port);
 		bootstrap.handler(new ChannelInitializer<SocketChannel>() {
 			@Override
 			protected void initChannel(SocketChannel socketChannel) throws Exception {
 				socketChannel.pipeline().addLast(new MessageDecoder(), new MessageEncoder(), new NettyClientHandler());
 			}
 		});
 		ChannelFuture future = bootstrap.connect(this.host, this.port).sync();
 		if (future.isSuccess()) {
 			socketChannel = (SocketChannel) future.channel();
 			System.out.println("connect server success|");
 		}
 	}
 	public int getPort() {
 		return this.port;
 	}
 	public void setPort(int port) {
 		this.port = port;
 	}
 
 	public SocketChannel getSocketChannel() {
 		return socketChannel;
 	}
 	public void setSocketChannel(SocketChannel socketChannel) {
 		this.socketChannel = socketChannel;
 	}
 	public String getHost() {
 		return host;
 	}
 	public void setHost(String host) {
 		this.host = host;
 	}
 }

初始化客户端

public static void main(String[] args) throws Exception {
    		NettyClientBootstrap bootstrap = new NettyClientBootstrap(9999, "127.0.0.1");
    		int i = 1;
    
    		while (true) {
    			TimeUnit.SECONDS.sleep(2);
    			RequestInfoVO req = new RequestInfoVO();
    			req.setSequence(123456);
    			req.setType((byte) 1);
    			req.setSequence(0);
    			req.setBody(String.valueOf((new Date()).getTime()));
    			bootstrap.getSocketChannel().writeAndFlush(req);
    			i++;
    		}
    	}

根据协议编写编码器，解码器，同服务端编码器、解码器

客户端事件处理

public class NettyClientHandler extends SimpleChannelInboundHandler<RequestInfoVO> {
 
 	@Override
 	protected void messageReceived(ChannelHandlerContext ctx, RequestInfoVO msg) throws Exception {
 		System.out.println(msg.getBody());
 		RequestInfoVO req = new RequestInfoVO();
 		req.setSequence(msg.getSequence());
 		req.setType(msg.getType());
 		if (2 == msg.getType()) {
 			req.setBody("client");
 			ctx.channel().writeAndFlush(req);
 		} else if (3 == msg.getType()) {
 			req.setBody("zpksb");
 			ctx.channel().writeAndFlush(req);
 		}
 
 	}
 
 }

至此，实现了简单的客户端服务端长连接。