通过使用Netty实现RPC

public class SerializationUtil {

	@SuppressWarnings("unchecked")
	public static <T> byte[] serialize(Object obj, Class<T> clazz) {
		Schema<T> schema = (Schema<T>) RuntimeSchema.getSchema(clazz);
		LinkedBuffer buffer = LinkedBuffer.allocate(4096);
		byte[] protostuff = null;
		try {
			protostuff = ProtostuffIOUtil.toByteArray((T)obj, schema, buffer);
		} catch (Exception ex) {
			ex.printStackTrace();
		} finally {
			buffer.clear();
		}
		return protostuff;
	}

	public static <T> T deserialize(byte[] buff, Class<T> clazz) {
		T object = null;
		try {
			object = (T) clazz.newInstance();
			Schema<T> schema = RuntimeSchema.getSchema(clazz);
			ProtostuffIOUtil.mergeFrom(buff, object, schema);
		} catch (InstantiationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		return object;
	}

public class RPCEncoder extends MessageToByteEncoder {

	private Class<?> clazz;

	public RPCEncoder(Class<?> clazz) {
		this.clazz = clazz;
	}
	@Override
	protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
		if (clazz.isInstance(msg)) {
			// 序列化
			byte[] buff = SerializationUtil.serialize(msg, clazz);
			out.writeInt(buff.length);
			out.writeBytes(buff);
		}

	}
}

public class RPCDecoder extends ByteToMessageDecoder {
	private Class<?> clazz;

	public RPCDecoder(Class<?> clazz) {
		this.clazz = clazz;
	}

	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
		if (in.readableBytes() < 4) {
			return;
		}
		in.markReaderIndex();
		int dataLength = in.readInt();
		if (dataLength < 0) {
			ctx.close();
		}
		if (in.readableBytes() < dataLength) {
			in.resetReaderIndex();
		}
		byte[] data = new byte[dataLength];
		in.readBytes(data);
		Object object = null;
		//反序列化
		object = SerializationUtil.deserialize(data, clazz);
		out.add(object);
	}
}

public class StateEvent {

	public String clientId;// 客户端编号，和jewel协议层一致

	private boolean isActice = true;// 通道是可用

	private boolean isClose = false;// 连接关闭不可用

	private boolean isValid = false;// 通道是空闲的

	public boolean isClose() {
		return isClose;
	}

	public void setValid(boolean isValid) {
		this.isValid = isValid;
	}

	public boolean isValid() {
		return this.isValid;
	}

	public void setClose(boolean isClose) {
		this.isClose = isClose;
	}

	private boolean reConnect;

	public void setReConnect(boolean reConnect) {
		this.reConnect = reConnect;
	}

	public void setClientId(String clientId) {
		this.clientId = clientId;
	}

	public String getClientId() {
		return clientId;
	}

	public boolean isReConnect() {
		return reConnect;
	}

	public boolean isActive() {
		return isActice;
	}

	public void setActice(boolean isActice) {
		this.isActice = isActice;
	}

	@Override
	public String toString() {
		return "StateEvent [clientId=" + clientId + ", isActice=" + isActice + ", isClose=" + isClose + ", isVal
				+ reConnect + "]";
	}

}

/**
 * 
 * @Description: netty客户端连接状态监听器类
 * @author maybo
 * @date 2016年5月13日 下午3:12:24
 *
 */
public interface ProtocolActiveListener {
	/**
	 * 
	 * @Description: TODO(事件方法，客户端连接状态)
	 * @param @param stateEvent 状态类
	 * @return void 返回类型
	 */
	public void clientStateEvent(StateEvent stateEvent);

}

public class RPCClient {

	private Logger logger = LoggerFactory.getLogger(this.getClass());

	private String id;// 编号

	public static String HOST = "127.0.0.1";

	public static int PORT = 9999;

	public static int TIMEOUT__SECONDS = 120;// 超时时间

	public static int CONNECT_TIMEOUT_SECONDS = 3;// 连接超时时间

	public static int RECONN_TIME_SECONDS = 3;// 重新建立连接时间

	private ChannelFuture channelFuture;

	private int connectAmount = 0;

	private RPCClient client = null;

	private Bootstrap bootstrap = null;

	private EventLoopGroup group;

	public static Map<String, Object> responseMap = new ConcurrentHashMap<String, Object>();

	public ProtocolActiveListener activeListener = null;

	public ProtocolActiveListener getActiveListener() {
		return activeListener;
	}

	public RPCClient(String id) {
		this.id = id;
	}

	/**
	 * 初始化Bootstrap
	 * 
	 * @return
	 */
	private Bootstrap getBootstrap() {
		group = new NioEventLoopGroup();
		Bootstrap b = new Bootstrap();
		b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, CONNECT_TIMEOUT_SECONDS * 1000);
		b.group(group).channel(NioSocketChannel.class);
		b.handler(new ChannelInitializer<Channel>() {
			@Override
			protected void initChannel(Channel ch) throws Exception {
				ChannelPipeline pipeline = ch.pipeline();
				pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));// 防止丢包
				pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
				pipeline.addLast(new RPCEncoder(RPCRequest.class));// 编码
				pipeline.addLast(new RPCDecoder(RPCResponse.class));// 转码
				pipeline.addLast(new RPCClientHandler(responseMap, client));// 处理
			}
		});
		b.option(ChannelOption.SO_KEEPALIVE, true);
		return b;
	}

	/**
	 * 
	 * @Description: TODO(初始化建立连接)
	 * @param @param id
	 * @param @return 设定文件
	 * @return RPCClient 返回类型
	 */
	public static RPCClient init(String id) {
		RPCClient client = new RPCClient(id);
		client.client = client;
		client.bootstrap = client.getBootstrap();
		return client;
	}

	public RPCClient getClient() {
		return client;
	}

	public boolean isActive() {
		return client.getChannelFuture().channel().isActive();
	}

	public RPCClient doConnection() throws InterruptedException {
		client.connection(HOST, PORT);
		return client;
	}

	/**
	 * 
	 * @Description: TODO(连接)
	 * @param @param host
	 * @param @param port
	 * @param @return 设定文件
	 * @return RPCClient 返回类型
	 * @throws InterruptedException 
	 */
	public RPCClient connection(String host, int port) throws InterruptedException {
		
			channelFuture = client.bootstrap.connect(host, port).sync();
			channelFuture.addListener(new ChannelFutureListener() {

				public void operationComplete(ChannelFuture future) throws Exception {// 重连机制
					if (future.isSuccess()) {
						// 连接成功连接指数为0
						connectAmount = 0;

						if (null != activeListener) {// 接听客户状态
							StateEvent stateEvent = new StateEvent();
							stateEvent.setActice(true);
							stateEvent.setClientId(id);
							stateEvent.setClose(false);
							stateEvent.setReConnect(true);
							activeListener.clientStateEvent(stateEvent);
						}
						logger.info("客户端连接成功。");

					} else {
						connectAmount++;
						if (connectAmount == 3) {// 连接数大于3次停止连接
							connectAmount = 0;
							shutdown();// 关闭连接
						} else {
							future.channel().eventLoop().schedule(new Runnable() {
								public void run() {
									try {
										doConnection();
									} catch (InterruptedException e) {
										e.printStackTrace();
										logger.error("------------重新连接服务器失败--------");
									}
								}
							}, RECONN_TIME_SECONDS, TimeUnit.SECONDS);

							if (null != activeListener) {// 接听客户状态
								StateEvent stateEvent = new StateEvent();
								stateEvent.setActice(false);
								stateEvent.setClientId(id);
								stateEvent.setClose(false);
								stateEvent.setReConnect(false);
								activeListener.clientStateEvent(stateEvent);
							}
						}
					}
				}
			});
		
		return client;
	}

	public ChannelFuture getChannelFuture() {
		return channelFuture;
	}

	/**
	 * 
	 * @Description: TODO(发送消息)
	 * @param @param request
	 * @param @return
	 * @param @throws Exception 设定文件
	 * @return RPCResponse 返回类型
	 */
	public RPCResponse sendMsg(RPCRequest request) throws Exception {
		if (null == request) {
			throw new NullPointerException();
		}
		try {
			RPCResponse response = new RPCResponse();
			responseMap.put(request.getRequestId(), response);

			if (channelFuture.channel().isOpen() && channelFuture.channel().isActive()) {
				if (null != this.activeListener) {// 发送监听通道
					StateEvent stateEvent = new StateEvent();
					stateEvent.setActice(true);
					stateEvent.setClientId(this.id);
					stateEvent.setClose(false);
					stateEvent.setReConnect(false);
					stateEvent.setValid(false);
					activeListener.clientStateEvent(stateEvent);
				}
				channelFuture.channel().writeAndFlush(request);
				synchronized (response) {
					if (null == response.getResponseId()) {
						response.wait(TIMEOUT__SECONDS * 1000);
					} else {
						response.notifyAll();
					}
					if (null == response.getResponseId()) {
						channelFuture.channel().close();
						response = new RPCResponse();
						response.setDate(new Date());
						response.setResponseId(request.getRequestId());
						response.setState("305");
						response.setError(new RequestTimeOutException("请求超时"));
						this.shutdown();// 关闭连接

					} else {
						if (null != this.activeListener) {// 发送监听通道
							StateEvent stateEvent = new StateEvent();
							stateEvent.setActice(true);
							stateEvent.setClientId(this.id);
							stateEvent.setClose(false);
							stateEvent.setReConnect(false);
							stateEvent.setValid(true);
							activeListener.clientStateEvent(stateEvent);
						}
						response = (RPCResponse) responseMap.get(request.getRequestId());
						System.out.println(response.toString());
						responseMap.remove(request.getRequestId());
					}
				}

			}

			return response;
		} finally {

		}
	}

	public void shutdown() throws InterruptedException {
		channelFuture.channel().close();
		group.shutdownGracefully();

		if (null != activeListener) {// 接听客户状态
			StateEvent stateEvent = new StateEvent();
			stateEvent.setActice(false);
			stateEvent.setClientId(this.id);
			stateEvent.setClose(true);
			activeListener.clientStateEvent(stateEvent);
		}
		logger.info("客户端关闭连接。");
	}

	public void addActiveListener(ProtocolActiveListener listener) {
		this.activeListener = listener;
	}

	public String id() {
		return id;
	}

	public void setId(String id) {
		this.id = id;
	}

}

public class RPCClientHandler extends SimpleChannelInboundHandler<RPCResponse> {

	private RPCClient client = null;

	private Logger logger = LoggerFactory.getLogger(RPCClientHandler.class);

	public void setClient(RPCClient client) {
		this.client = client;
	}

	private Map<String, Object> responseMap;

	public RPCClientHandler(Map<String, Object> responseMap) {
		this.responseMap = responseMap;
	}

	public RPCClientHandler(Map<String, Object> responseMap, RPCClient client) {
		this.responseMap = responseMap;
		this.client = client;
	}

	@Override
	protected void messageReceived(ChannelHandlerContext ctx, RPCResponse response) throws Exception {
		if (response.getType() == 1) {// 发送心跳回文

			if (null != this.client.getActiveListener()) {// 监听通道空闲
				StateEvent stateEvent = new StateEvent();
				stateEvent.setActice(true);
				stateEvent.setClientId(this.client.id());
				stateEvent.setClose(false);
				stateEvent.setReConnect(false);
				stateEvent.setValid(true);
				this.client.getActiveListener().clientStateEvent(stateEvent);
			}

			RPCRequest request = new RPCRequest();
			request.setType(1);
			ctx.channel().writeAndFlush(request);
		} else {// 接收消息
			RPCResponse rpcResponse = (RPCResponse) responseMap.get(response.getResponseId());
			synchronized (rpcResponse) {
				BeanUtils.copyProperties(rpcResponse, response);
				rpcResponse.notifyAll();
			}
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
		e.printStackTrace();
		super.exceptionCaught(ctx, e);
		if (e instanceof IOException) {// 发生异常关闭通道
			ctx.channel().close();
			client.shutdown();
		}
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		super.channelInactive(ctx);
		// 重新连接服务器
		ctx.channel().eventLoop().schedule(new Runnable() {
			public void run() {
				try {
					client.doConnection();
				} catch (InterruptedException e) {
					logger.error("断开连接后重新建立连接出现异常."+e.getMessage());
					e.printStackTrace();
				}
			}
		}, RPCClient.RECONN_TIME_SECONDS, TimeUnit.SECONDS);
	}

}