欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

基于netty的分布式模拟实现 netty分布式 

程序员文章站 2022-05-28 12:58:00
...

 

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

 

 

 

 

 

 

 

 

 

 

 

 

相关标签: netty 分布式