基于netty的分布式模拟实现

 

1、RpcService,把本地的服务暴露给ZooKeeper

@Component("rpcService")
public class RpcService implements ApplicationContextAware,InitializingBean {

	private String serviceAddress;
	private Map<String,Object> handlerMap;
	private ZookeeperRegister zk;
	
	public RpcService(String serviceAddress, ZookeeperRegister zk) {
		this.serviceAddress = serviceAddress;
		this.zk = zk;
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline()
                                    .addLast(new RpcDecoder(RpcRequest.class)) // 将 RPC 请求进行解码（为了处理请求）
                                    .addLast(new RpcEncoder(RpcResponse.class)) // 将 RPC 响应进行编码（为了返回响应）
                                    .addLast(new RpcHandler(handlerMap)); // 处理 RPC 请求
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
		
            String[] ipMArray = serviceAddress.split(":");
           // bootstrap.bind(ipMArray[0], Integer.parseInt(ipMArray[1]));
            
            ChannelFuture future = bootstrap.bind(ipMArray[0], Integer.parseInt(ipMArray[1])).sync();
            //注册  
            if(zk!=null) {
            	zk.register(serviceAddress);
            }
            
            future.channel().closeFuture().sync();
            
            
        }catch(Exception e) {
        	
        }finally {
        	 workerGroup.shutdownGracefully();
             bossGroup.shutdownGracefully();
        }
	}

	@Override
	public void setApplicationContext(ApplicationContext context) throws BeansException {
		handlerMap = context.getBeansWithAnnotation(RpcServiceAnnotation.class);
		
		Map<String,Object> controllerAnnotation = context.getBeansWithAnnotation(Controller.class);
		
		try {
			for(Object obj:controllerAnnotation.values()) {
				Class<?> clazz = obj.getClass();
				Field[] fields = clazz.getFields();
				for(Field f:fields) {
					boolean flag = f.isAnnotationPresent(Resource.class);
					if(flag) {
						Resource r = f.getAnnotation(Resource.class);
						String name = r.name();
						if("".equals(name)) {
							name = f.getName();
						}
						f.setAccessible(true);
						
						Object proxyObject = RpcProxy.newInstance(f.getClass());
						
						f.set(obj, proxyObject);
					}
				}
			}
		} catch (SecurityException e) {
			e.printStackTrace();
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			e.printStackTrace();
		}
	}
}

2、RpcService处理handler

public class RpcHandler extends SimpleChannelInboundHandler<RpcRequest> {

	private Map<String,Object> handlerMap;
	
	public RpcHandler(Map<String, Object> handlerMap) {
		this.handlerMap = handlerMap;
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, RpcRequest req) throws Exception {
		Object handlerObject = handlerMap.get(req.getBeanName());
		
	    Class handlerObjectClass = handlerObject.getClass();
	    Method method = handlerObjectClass.getMethod(req.getMethodName(), req.getParamTypes());
	    Object result = method.invoke(handlerObject, req.getParamArgs());
	    
	    RpcResponse response = new RpcResponse();
	    response.setResult(result);
	    
	    ctx.writeAndFlush(response);
	}

}

 3、RpcDecoder,channel解码

public class RpcDecoder extends ByteToMessageDecoder {
	private Class<?> decodeClass;
	
	public RpcDecoder(Class<?> decodeClass) {
		this.decodeClass = decodeClass;
	}
	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
		//解码
		if(in.readableBytes()>0) {
			byte[] byteArray = new byte[in.readInt()];
			in.readBytes(byteArray);
			Object decodeObject = SerializationUtil.deserialize(byteArray, decodeClass);
			
			out.add(decodeObject);
		}
	}
}

 4、RpcEncode，发送数据转byte

public class RpcEncoder extends MessageToByteEncoder {

	private Class<?> encodeClass;
	
	public RpcEncoder(Class<?> encodeClass) {
		this.encodeClass = encodeClass;
	}

	@Override
	protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
		//编码
		if(encodeClass.isInstance(msg)) {			
			byte[] msgByte = SerializationUtil.serialize(msg);
			
			out.setInt(0, msgByte.length);
			out.readBytes(msgByte);
		}
	}
}

 5、RpcRequest,请求

public class RpcRequest implements Serializable {

	private static final long serialVersionUID = 1L;
	
	private String methodName;
	private String beanName;
	private Object[] paramArgs;
	private Class<?>[] paramTypes;
	
	public String getMethodName() {
		return methodName;
	}
	public void setMethodName(String methodName) {
		this.methodName = methodName;
	}
	public String getBeanName() {
		return beanName;
	}
	public void setBeanName(String beanName) {
		this.beanName = beanName;
	}
	public Object[] getParamArgs() {
		return paramArgs;
	}
	public void setParamArgs(Object[] paramArgs) {
		this.paramArgs = paramArgs;
	}
	public Class<?>[] getParamTypes() {
		return paramTypes;
	}
	public void setParamTypes(Class<?>[] paramTypes) {
		this.paramTypes = paramTypes;
	}
}

 6、RpcResponse

public class RpcResponse {

	public Object result;
	

	public Object getResult() {
		return result;
	}

	public void setResult(Object result) {
		this.result = result;
	}
	
}

 

7、RpcServiceAnnotation用于提供service的类

/**
 * RPC接口注解
 */
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface RpcServiceAnnotation {

	public String value();
	
}

 8、BizProxy，用于Controller类属性注解

/**
 * 用于Controller类下面的属性注解
 */
@Target(ElementType.FIELD)
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface BizProxy {

	public String value();
	
}

 9、SerializationUtil

public class SerializationUtil {

	private static Map<Class<?>, Schema<?>> cachedSchema = new ConcurrentHashMap<>();

    private static Objenesis objenesis = new ObjenesisStd(true);

    private SerializationUtil() {
    }

    @SuppressWarnings("unchecked")
    private static <T> Schema<T> getSchema(Class<T> cls) {
        Schema<T> schema = (Schema<T>) cachedSchema.get(cls);
        if (schema == null) {
            schema = RuntimeSchema.createFrom(cls);
            if (schema != null) {
                cachedSchema.put(cls, schema);
            }
        }
        return schema;
    }

    @SuppressWarnings("unchecked")
    public static <T> byte[] serialize(T obj) {
        Class<T> cls = (Class<T>) obj.getClass();
        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
        try {
            Schema<T> schema = getSchema(cls);
            return ProtostuffIOUtil.toByteArray(obj, schema, buffer);
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        } finally {
            buffer.clear();
        }
    }

    public static <T> T deserialize(byte[] data, Class<T> cls) {
        try {
            T message = (T) objenesis.newInstance(cls);
            Schema<T> schema = getSchema(cls);
            ProtostuffIOUtil.mergeFrom(data, message, schema);
            return message;
        } catch (Exception e) {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }
}

 10、ZookeeperRegister,注册ZooKeeper

@Component("zk")
public class ZookeeperRegister implements Watcher {

	private String registerAddress;
	private String childService = "nettyServiceIp";
	private CountDownLatch countDownLatch;
	private ZooKeeper zk;

	public ZookeeperRegister(String registerAddress) {
		this.registerAddress = registerAddress;
		connect();
	}
	
	public void connect() {
		try {
			countDownLatch = new CountDownLatch(1);
			zk = new ZooKeeper(registerAddress, ZookeeperContant.SESSION_TIME_OUT, this);
			countDownLatch.await();
			Stat stat = zk.exists(ZookeeperContant.ROOT, false);
			if(stat==null) {
				zk.create(ZookeeperContant.ROOT, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	public void register(String serviceAddress) {
		try {
			//创建子节点
			zk.create(ZookeeperContant.ROOT+"/"+childService, serviceAddress.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void process(WatchedEvent arg0) {
		if (arg0.getState() == Event.KeeperState.SyncConnected) {
			countDownLatch.countDown();
        }
	}
}

 11、ZookeeperContant

public interface ZookeeperContant {

	public String CHILD_SERVICE = "nettyServiceIp";
	
	public String ROOT = "netty";
	
	public int SESSION_TIME_OUT = 2000;
}

 以下是客户端的类，用于获取服务地址和发送请求

 

12、RpcClient,发送请求

public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {

	private ServiceDiscovery serviceDiscovery;
	
	private RpcResponse response;
	
	private Object object = new Object();
	
	public RpcClient() {
	}

	public RpcClient(ServiceDiscovery serviceDiscovery) {
		this.serviceDiscovery = serviceDiscovery;
	}
	
	public RpcResponse send(RpcRequest req) {
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		try {
			Bootstrap bootstrap = new Bootstrap();
			bootstrap.group(bossGroup).channel(NioSocketChannel.class)
			        .handler(new ChannelInitializer<SocketChannel>() {
			            @Override
			            public void initChannel(SocketChannel channel) throws Exception {
			                channel.pipeline()
			                        .addLast(new RpcDecoder(RpcRequest.class)) // 将 RPC 请求进行解码（为了处理请求）
			                        .addLast(new RpcEncoder(RpcResponse.class)) // 将 RPC 响应进行编码（为了返回响应）
			                        .addLast(RpcClient.this); // 处理 RPC 请求
			            }
			        })
			        .option(ChannelOption.SO_KEEPALIVE, true);
			
			String addressIp = serviceDiscovery.discovery();
			String[] ipArray = addressIp.split(":");
			//发送数据
			ChannelFuture future = bootstrap.connect(ipArray[0], Integer.parseInt(ipArray[1])).sync();
			future.channel().writeAndFlush(req);
			//等待返回值
			synchronized (object) {
				object.wait();
			}
			
			
		} catch (Exception e) {
			e.printStackTrace();
		}finally {
            bossGroup.shutdownGracefully();
		}
		return response;
	}
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, RpcResponse res) throws Exception {
		response = res;
		synchronized (object) {
			object.notifyAll();
		}
	}
}

 13、RpcProxy，初始化对象

public class RpcProxy {
	private Object target;
	public static Object newInstance(final Class<?> cls) {
		//RpcServiceAnnotation annotation = cls.getAnnotation(RpcServiceAnnotation.class);
		//final String beanName = annotation.value();		
		return Proxy.newProxyInstance(cls.getClassLoader(), cls.getInterfaces(), new InvocationHandler() {
			@Override
			public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
				RpcRequest req = new RpcRequest();
				req.setBeanName(cls.getName());       //需要处理
				req.setMethodName(method.getName());
				req.setParamArgs(args);
				req.setParamTypes(method.getParameterTypes());
				
				RpcClient client = new RpcClient();
				RpcResponse response = client.send(req);
				
				return response;
			}
		});
	}
}

 14、ServiceDiscovery和ServiceDiscoveryImpl

public interface ServiceDiscovery {

	public String discovery();
	
}

@Component("serviceDiscovery")
public class ServiceDiscoveryImpl implements ServiceDiscovery,Watcher {
	private String serviceDiscoveryAddress;
	private CountDownLatch latch = new CountDownLatch(1);
	private ZooKeeper zk;	
	public ServiceDiscoveryImpl(String serviceDiscoveryAddress) {
		this.serviceDiscoveryAddress = serviceDiscoveryAddress;
		connect();
	}	
	public void connect() {
		try {			
			zk = new ZooKeeper(serviceDiscoveryAddress,ZookeeperContant.SESSION_TIME_OUT,this);
			latch.await();
		}catch(Exception e) {
			e.printStackTrace();
		}
	}
	@Override
	public String discovery() {
		try {
			List<String> serviceIpList = new ArrayList<String>();
			List<String> listAddress = zk.getChildren(ZookeeperContant.ROOT+"/"+ZookeeperContant.CHILD_SERVICE, false);
			for(String str:listAddress) {
				byte[] ipBytes = zk.getData(str, false, new Stat());
				String address = new String(ipBytes);
				serviceIpList.add(address);
			}
			if(serviceIpList.size()==1) {
				return serviceIpList.get(0);
			}
			Random random = new Random();
			
			String addressIp = serviceIpList.get(random.nextInt(serviceIpList.size()));
			
			return addressIp;
			
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return null;
	}

	@Override
	public void process(WatchedEvent event) {
		if(event.getState().equals(Event.KeeperState.SyncConnected)) {
			latch.countDown();
		}
	}
}

 15、BeanFactoryProcessTest，Scan，BeanFactoryBeanTest

@Component("beanFactoryProcess")
public class BeanFactoryProcessTest implements ApplicationContextAware,BeanFactoryPostProcessor {

	private ApplicationContext applicationContext;
	
	@Override
	public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
		Scan san = new Scan((BeanDefinitionRegistry)beanFactory);
		san.setResourceLoader(applicationContext);
	    san.scan("com.paic.egis.*");
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		this.applicationContext = applicationContext;
	}
}

 

public class BeanFactoryBeanTest<T> implements FactoryBean<T> {

	private String innerClassName;
	
	public BeanFactoryBeanTest(String innerClassName) {
		this.innerClassName = innerClassName;
	}

	@Override
	public T getObject() throws Exception {
		final Class<?> className = Class.forName(innerClassName);
		if(className.isInterface()) {
			return (T)RpcProxy.newInstance(className);
		}else {
			 Enhancer enhancer = new Enhancer();
	         enhancer.setSuperclass(className);
	         enhancer.setNamingPolicy((NamingPolicy) SpringNamingPolicy.INSTANCE);
	         enhancer.setCallback(new MethodInterceptor() {
				
				@Override
				public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable {
					
					RpcRequest req = new RpcRequest();
					req.setBeanName(className.getName());       //需要处理
					req.setMethodName(method.getName());
					req.setParamArgs(args);
					req.setParamTypes(method.getParameterTypes());
					
					RpcClient client = new RpcClient();
					RpcResponse response = client.send(req);
					
					return response;
				}
			});
	         return (T) enhancer.create();
		}
	}

	@Override
	public Class<?> getObjectType() {
		try {
			return Class.forName(innerClassName);
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
		return null;
	}

	@Override
	public boolean isSingleton() {
		
		return true;
	}
	
}

 

public class Scan extends ClassPathBeanDefinitionScanner {

	public Scan(BeanDefinitionRegistry registry) {
		super(registry);
	}

	@Override
	protected Set<BeanDefinitionHolder> doScan(String... basePackages) {
		Set<BeanDefinitionHolder> definitionHolders = super.doScan(basePackages);
		for(BeanDefinitionHolder holder:definitionHolders) {
			GenericBeanDefinition bean = (GenericBeanDefinition)holder.getBeanDefinition();
			bean.getPropertyValues().add("innerClass", bean.getBeanClassName());
			bean.setBeanClass(BeanFactoryBeanTest.class);
		}
		return definitionHolders;
	}

	@Override
	protected boolean isCandidateComponent(AnnotatedBeanDefinition beanDefinition) {
		
		 return super.isCandidateComponent(beanDefinition) && beanDefinition.getMetadata()
				 .hasAnnotation(RpcServiceAnnotation.class.getName());
		//return super.isCandidateComponent(beanDefinition);
	}
}