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();
		}
}

定義客戶端服務端通訊協議，一下是一個簡單的通訊協議

根據通訊協議，編寫解碼器和編碼器

解碼器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);
		}

	}

}