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); } }
相关推荐
客户端与服务端连接 客户端处理服用户输入,发送消息到服务端 ...实现Server端分布式 分布式部署 消息群发 路由端转发消息 客户端上下线处理 服务端宕机,客户端重连处理 未来扩展方向 下载直接可以用
一个基于netty封装的快速分布式任务开发框架,目标:简单,简单,再简单! easynetty-demo是用spring-boot,spring-shell实现的示例
做Java开发,现在很多场合需要分布式应用,很多通信框架的底层实现都包含Netty技术,为了更好了解,值得有兴趣的人研究一下
封装NETTY来实现一个分布式文件传输服务,目标是提供定时的可靠安全文件传输。同时提供一种简便的方式,只需将数据配置好
JAVA版基于netty的物联网高并发智能网关 JAVA版基于netty的物联网高并发智能网关 JAVA版基于netty的物联网高并发智能网关 JAVA版基于netty的物联网高并发智能网关 JAVA版基于netty的物联网高并发智能网关 JAVA...
基于Netty实现的命令行斗地主游戏,为划水摸鱼而生~ 基于Netty实现的命令行斗地主游戏,为划水摸鱼而生~ 基于Netty实现的命令行斗地主游戏,为划水摸鱼而生~ 基于Netty实现的命令行斗地主游戏,为划水摸鱼而生...
基于Netty实现的内网穿透&反向代理的工具 (支持TCP上层协议和HTTP的穿透式反向代理).zip
尊敬的用户您好!本资源中的源码已经过本地编译测试,下载后按照文档配置好环境就可以运行。资源项目的难度比较适中,内容都是经过助教老师审定过的,应该能够满足学习、使用需求,如果有需要的话可以放心下载使用。...
一个基于netty实现web框架,或者mvc框架,实现基于netty的web框架,你说netty强不强,文中有不对的地方,欢迎大牛指正
netty-websocket-example 基于netty的websocket实现示例 高并发 附jmx压力测试
分布式锁,基于Netty长连接实现,自定义协议,高性能锁,内置等待队列,服务下线立刻感知,无需等待超时.支持优雅关闭.规避Redis分布式锁缺陷
基于Netty实现了dubbo rpc
基于netty框架编写的socket服务器
使用Netty进行网络通信,完成分布式的数据采集任务,可以采集500以上数据节点。
1、基于netty+websocket+springboot的实时聊天系统项目源码.zip 2、该资源包括项目的全部源码,下载可以直接使用! 3、本项目适合作为计算机、数学、电子信息等专业的课程设计、期末大作业和毕设项目,作为参考资料...
基于Netty框架开发的Modbus源代码。支持 * READ COILS | 0x01 * READ DISCRETE INPUTS | 0x02 * READ HOLDING REGISTERS | 0x03 * READ INPUT REGISTERS | 0x04 * WRITE SINGLE COIL | 0x05 * WRITE SINGLE REGISTER...
基于netty与Springboot实现的编程游戏服务器_源码
Android基于Netty框架实现通信
基于 Java Netty实现的可用于内网穿透的代理工具.zip基于 Java Netty实现的可用于内网穿透的代理工具.zip基于 Java Netty实现的可用于内网穿透的代理工具.zip基于 Java Netty实现的可用于内网穿透的代理工具.zip基于...
netty案例,netty4.1中级拓展篇十三《Netty基于SSL实现信息传输过程中双向加密验证》源码 ...