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