Netty Server

你的名字 2022-08-27 13:47 125阅读 0赞

**client与server的对话**

public interface NettyRequest<T extends NettyResponse> extends Serializable {
    
    	public String getUid();
    
    	public Class<T> getResponseClass();
    
    }

public class NettyResponse implements Serializable {
    
    	private static final long serialVersionUID = 8576456371876246088L;
    
    	private String uid;
    
    	public String getUid() {
    		return uid;
    	}
    
    	public void setUid(String uid) {
    		this.uid = uid;
    	}
    
    }

**codec**

public class NettyRequestDecoder extends ByteToMessageDecoder {
    
    	@Override
    	protected void decode(ChannelHandlerContext ctx, ByteBuf in,
    			List<Object> out) throws Exception {
    		try {
    			byte[] dst = new byte[in.readableBytes()];
    			in.readBytes(dst);
    			out.add(ObjectUtil.toObject(dst));
    		} catch (Exception e) {
    			in.resetReaderIndex();
    		}
    	}
    
    }

public class NettyRequestEncoder extends MessageToByteEncoder<NettyRequest<?>> {
    
    	@Override
    	protected void encode(ChannelHandlerContext ctx, NettyRequest<?> msg,
    			ByteBuf out) throws Exception {
    		out.writeBytes(ObjectUtil.toByte(msg)); // data
    	}
    
    }

public class NettyResponseDecoder extends ByteToMessageDecoder {
    
    	@Override
    	protected void decode(ChannelHandlerContext ctx, ByteBuf in,
    			List<Object> out) throws Exception {
    		try {
    			byte[] dst = new byte[in.readableBytes()];
    			in.readBytes(dst);
    			out.add(ObjectUtil.toObject(dst));
    		} catch (Exception e) {
    			in.resetReaderIndex();
    		}
    	}
    
    }

public class NettyResponseEncoder extends MessageToByteEncoder<NettyResponse> {
    
    	@Override
    	protected void encode(ChannelHandlerContext ctx, NettyResponse msg,
    			ByteBuf out) throws Exception {
    		out.writeBytes(ObjectUtil.toByte(msg)); // data
    	}
    
    }

**Server**

public interface Server extends Runnable{
    
    	void start();
    
    }

public class NettyServer implements Server {
    
    	private int port;
    	private Channel channel;
    
    	public NettyServer(int port) {
    		this.port = port;
    	}
    
    	@Override
    	public void start() {
    		try {
    			new Thread(this).start();
    			Thread.currentThread().join(); // just wait..
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    
    	@Override
    	public void run() {
    		EventLoopGroup bossGroup = new NioEventLoopGroup();
    		EventLoopGroup workerGroup = new NioEventLoopGroup();
    		try {
    			ServerBootstrap b = new ServerBootstrap();
    			b.group(bossGroup, workerGroup);
    			b.channel(NioServerSocketChannel.class);
    			b.childHandler(new ChannelInitializer<SocketChannel>() {
    				@Override
    				public void initChannel(SocketChannel ch) throws Exception {
    					// 解码用
    					// decode byte to request, then goto next Inbound handler
    					ch.pipeline().addLast("requestDecoder",
    							new NettyRequestDecoder());
    
    					// encode response to byte, then send to client;
    					ch.pipeline().addLast("responseEncoder",
    							new NettyResponseEncoder());
    
    					// 业务逻辑用
    					ch.pipeline().addLast("handler", new ServiceDispatcher());
    				}
    			});
    			b.option(ChannelOption.SO_BACKLOG, 1024);
    			b.option(ChannelOption.TCP_NODELAY, true);
    			b.option(ChannelOption.SO_KEEPALIVE, true);
    			ChannelFuture bind = b.bind(port);
    			ChannelFuture f = bind.sync();
    			f.channel().closeFuture().sync();
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			workerGroup.shutdownGracefully();
    			bossGroup.shutdownGracefully();
    		}
    	}
    
    	public void stop() {
    		if (channel != null) {
    			channel.close();
    		}
    	}
    
    }

public class ServiceDispatcher extends
    		SimpleChannelInboundHandler<NettyRequest<?>> {
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
    			throws Exception {
    		cause.printStackTrace();
    		super.exceptionCaught(ctx, cause);
    	}
    
    	@Override
    	protected void channelRead0(ChannelHandlerContext ctx, NettyRequest<?> msg)
    			throws Exception {
    		NettyResponse response = msg.getResponseClass().newInstance();
    		response.setUid(msg.getUid());
    		// 回传消息
    		ctx.channel().writeAndFlush(response);
    	}
    
    	@Override
    	public void channelActive(ChannelHandlerContext ctx) throws Exception {
    		System.err.println(ctx.channel().remoteAddress().toString()
    				+ " is actived");
    	}
    }

**client**

public interface Client {
    
    	public <T extends NettyResponse> T execute(NettyRequest<T> request);
    
    }

public class ClientService implements Runnable {
    
    	private volatile Channel channel;
    
    	private String rpcUri;
    
    	private CountDownLatch startCDL;
    
    	private ChannelHandler clientDispatcher;
    
    	public ClientService(String rpcUri, ChannelHandler clientDispatcher,
    			CountDownLatch startCDL) {
    		this.rpcUri = rpcUri;
    		this.startCDL = startCDL;
    		this.clientDispatcher = clientDispatcher;
    	}
    
    	@Override
    	public void run() {
    		EventLoopGroup workerGroup = new NioEventLoopGroup();
    		SocketAddress serviceAddr = parseRpcUri(rpcUri);
    		try {
    			Bootstrap b = new Bootstrap();
    			b.group(workerGroup);
    			b.channel(NioSocketChannel.class);
    			b.option(ChannelOption.TCP_NODELAY, true);
    			b.option(ChannelOption.SO_KEEPALIVE, true);
    			b.handler(new ChannelInitializer<SocketChannel>() {
    				@Override
    				public void initChannel(SocketChannel ch) throws Exception {
    					// decode server byte to response, then goto to Inbound
    					// handler;
    					ch.pipeline().addLast("responseDecoder",
    							new NettyResponseDecoder());
    					// encode request to byte , then send to server
    					ch.pipeline().addLast("requestEncoder",
    							new NettyRequestEncoder());
    
    					// 业务逻辑用
    					ch.pipeline().addLast("handler", clientDispatcher);
    				}
    			});
    			ChannelFuture future = b.connect(serviceAddr).sync();
    			startCDL.countDown();
    			channel = future.channel();
    			channel.closeFuture().sync();
    		} catch (Exception e) {
    			startCDL.countDown();
    			channel = null;
    			throwExceptionIfInitfail();
    		} finally {
    			workerGroup.shutdownGracefully();
    		}
    	}
    
    	public void write(Object msg) {
    		write(msg, 0);
    	}
    
    	public void write(Object msg, long requestTimeout) {
    		if (requestTimeout > 0) {
    			channel.writeAndFlush(msg).awaitUninterruptibly(requestTimeout);
    		} else {
    			channel.writeAndFlush(msg);
    		}
    	}
    
    	/**
    	 * 
    	 * @param rpcUri
    	 * @return
    	 */
    	public SocketAddress parseRpcUri(String rpcUri) {
    		try {
    			URI uri = new URI("rpc://" + rpcUri);
    			String host = uri.getHost();
    			int port = uri.getPort();
    			InetSocketAddress address = new InetSocketAddress(host, port);
    			return address;
    		} catch (URISyntaxException e) {
    			throw new RuntimeException(e);
    		}
    	}
    
    	private void throwExceptionIfInitfail() {
    		if (channel == null) {
    			String message = String.format("connect %s fail!", rpcUri);
    			System.err.println(message);
    		} else {
    			String message = String.format("connect %s success!", rpcUri);
    			System.err.println(message);
    		}
    	}
    
    	public void close() {
    		channel.close();
    	}
    
    	public boolean isRunning() {
    		return channel == null ? false : channel.isActive();
    	}
    
    }

public class ClientDispatcher extends
    		SimpleChannelInboundHandler<NettyResponse> {
    
    	private HandlerFactory handlerFactory;
    
    	public ClientDispatcher(HandlerFactory handlerFactory) {
    		super();
    		this.handlerFactory = handlerFactory;
    	}
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    		StringWriter out = new StringWriter();
    		cause.printStackTrace(new PrintWriter(out));
    		if (ctx.channel() != null) {
    			ctx.channel().close().addListener(ChannelFutureListener.CLOSE);
    		}
    	}
    
    	@Override
    	protected void channelRead0(ChannelHandlerContext ctx,
    			NettyResponse response) throws Exception {
    		ResponseHandler handler = handlerFactory.getHandler(response.getUid());
    		if (handler != null) {
    			handler.onResponse(response);
    			handlerFactory.remove(response.getUid());
    		} else {
    			System.err.println("null ,id:" + response.getUid());
    		}
    	}
    }

public class NettyClient implements Client {
    
    	/**
    	 * 10秒超时
    	 */
    	private static final long DEFAULT_REQUEST_TIMEOUT = 100 * 1000;
    
    	private ClientService clientService;
    	private long requestTimeout;
    	private String rpcUri;
    
    	private HandlerFactory handlerFactory;
    
    	private ChannelHandler clientDispatcher;
    
    	public NettyClient(String rpcUri) {
    		this(rpcUri, DEFAULT_REQUEST_TIMEOUT);
    	}
    
    	public NettyClient(String rpcUri, long requestTimeout) {
    		this.rpcUri = rpcUri;
    		this.requestTimeout = requestTimeout;
    		this.handlerFactory = new HandlerFactory();
    		clientDispatcher = new ClientDispatcher(handlerFactory);
    		connectService();
    	}
    
    	private void connectService() {
    		CountDownLatch connectCDL = new CountDownLatch(1);
    		clientService = new ClientService(rpcUri, clientDispatcher, connectCDL);
    		Thread clientThread = new Thread(clientService);
    		clientThread.start();
    		try {
    			connectCDL.await();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    
    	@Override
    	public <T extends NettyResponse> T execute(NettyRequest<T> request) {
    		TimeoutMonitor<T> timeoutMonitor = new TimeoutMonitor<T>(clientService,
    				handlerFactory, request, requestTimeout);
    		Thread thread = new Thread(timeoutMonitor);
    		thread.start();
    		try {
    			thread.join();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		return timeoutMonitor.getResponse();
    	}
    
    	public void close() {
    		clientService.close();
    	}
    
    	/**
    	 * 是否存活
    	 * 
    	 * @return
    	 */
    	public boolean isRunning() {
    		return clientService.isRunning();
    	}
    
    }

public class TimeoutMonitor<T extends NettyResponse> implements Runnable {
    
    	/**
    	 * 计时器超时时间
    	 */
    	private long timeout;
    	private NettyRequest<?> request;
    	private T response;
    	private ClientService clientService;
    
    	private HandlerFactory handlerFactory;
    
    	/**
    	 * 构造器
    	 * 
    	 * @param timeout
    	 *            指定超时的时间
    	 */
    	public TimeoutMonitor(ClientService clientService,
    			HandlerFactory handlerFactory, NettyRequest<?> request, long timeout) {
    		this.clientService = clientService;
    		this.handlerFactory = handlerFactory;
    		this.request = request;
    		this.timeout = timeout;
    	}
    
    	/**
    	 * 启动超时计时器
    	 */
    	@Override
    	public void run() {
    		Thread thread = Thread.currentThread();
    		ResponseHandler syncHandler = new ResponseHandler(thread);
    		handlerFactory.registHandler(request.getUid(), syncHandler);
    		clientService.write(request);
    		try {
    			Thread.sleep(timeout);
    		} catch (InterruptedException e) {
    			// 服务器返回的时候会打断睡眠
    			this.response = (T) syncHandler.getResponse();
    		}
    	}
    
    	public T getResponse() {
    		return response;
    	}
    }

**线程与服务器响应的绑定类**

public class ResponseHandler {
    
    	private NettyResponse response;
    	private final Thread requestThread;
    
    	public ResponseHandler(Thread requestThread) {
    		this.requestThread = requestThread;
    	}
    
    	public NettyResponse getResponse() {
    		return response;
    	}
    
    	public void onResponse(NettyResponse response) {
    		this.response = response;
    		// 当服务器响应时，中断睡眠  
    		requestThread.interrupt();
    	}
    
    }