文档章节

Spring+Netty+Protostuff+ZooKeeper实现轻量级RPC服务

L
 Listen_ing
发布于 2016/05/26 18:37
字数 4199
阅读 2685
收藏 28
  • RPC简介

  • RPC,即 Remote Procedure Call(远程过程调用),说得通俗一点就是:调用远程计算机上的服务,就像调用本地服务一样。

    RPC 可基于 HTTP 或 TCP 协议,Web Service 就是基于 HTTP 协议的 RPC,它具有良好的跨平台性,但其性能却不如基于 TCP 协议的 RPC。会两方面会直接影响 RPC 的性能,一是传输方式,二是序列化。

    众所周知,TCP 是传输层协议,HTTP 是应用层协议,而传输层较应用层更加底层,在数据传输方面,越底层越快,因此,在一般情况下,TCP 一定比 HTTP 快。就序列化而言,Java 提供了默认的序列化方式,但在高并发的情况下,这种方式将会带来一些性能上的瓶颈,于是市面上出现了一系列优秀的序列化框架,比如:Protobuf、Kryo、Hessian、Jackson 等,它们可以取代 Java 默认的序列化,从而提供更高效的性能。

    为了支持高并发,传统的阻塞式 IO 显然不太合适,因此我们需要异步的 IO,即 NIO。Java 提供了 NIO 的解决方案,Java 7 也提供了更优秀的 NIO.2 支持,用 Java 实现 NIO 并不是遥不可及的事情,只是需要我们熟悉 NIO 的技术细节。

    我们需要将服务部署在分布式环境下的不同节点上,通过服务注册的方式,让客户端来自动发现当前可用的服务,并调用这些服务。这需要一种服务注册表(Service Registry)的组件,让它来注册分布式环境下所有的服务地址(包括:主机名与端口号)。

    应用、服务、服务注册表之间的关系见下图:

    系统架构

    每台 Server 上可发布多个 Service,这些 Service 共用一个 host 与 port,在分布式环境下会提供 Server 共同对外提供 Service。此外,为防止 Service Registry 出现单点故障,因此需要将其搭建为集群环境。

    本文将为您揭晓开发轻量级分布式 RPC 框架的具体过程,该框架基于 TCP 协议,提供了 NIO 特性,提供高效的序列化方式,同时也具备服务注册与发现的能力。

    根据以上技术需求,我们可使用如下技术选型:

  • Spring:它是最强大的依赖注入框架,也是业界的权威标准。
  • Netty:它使 NIO 编程更加容易,屏蔽了 Java 底层的 NIO 细节。
  • Protostuff:它基于 Protobuf 序列化框架,面向 POJO,无需编写 .proto 文件。
  • ZooKeeper:提供服务注册与发现功能,开发分布式系统的必备选择,同时它也具备天生的集群能力。
  •  
  • 服务端设计配置

  • 服务接口工程listen-rpc-service

  • 主要定义服务接口类和服务涉及到的实体类
  • listen-rpc-service工程目录结构图
  • listen-rpc-service工程pom.xml
  • <?xml version="1.0"?>
    <project
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
    	xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    	<modelVersion>4.0.0</modelVersion>
    	<parent>
    		<groupId>listen</groupId>
    		<artifactId>listen-parent</artifactId>
    		<version>0.0.1-SNAPSHOT</version>
    	</parent>
    	<artifactId>listen-rpc-service</artifactId>
    	<packaging>jar</packaging>
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    	</properties>
    	<dependencies>
    		<dependency>
    			<groupId>listen</groupId>
    			<artifactId>listen-rpc-common</artifactId>
    			<version>0.0.1-SNAPSHOT</version>
    		</dependency>
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>fastjson</artifactId>
    			<version>1.1.41</version>
    		</dependency>
    	</dependencies>
    </project>
    

     

  • listen-rpc-service工程服务接口IHelloService
  • package com.listen.rpc.service;
    
    import java.util.List;
    
    import com.listen.rpc.entity.User;
    
    public interface IHelloService {
    
    	public String hello(String name);
    	
    	public User getUser(String name);
    	
    	public List<User> getUsers(int size);
    	
    	public User updateUser(User user);
    	
    }
    
  • listen-rpc-service工程User类
  • package com.listen.rpc.entity;
    
    import java.util.Date;
    
    import com.alibaba.fastjson.JSON;
    
    public class User {
    
    	private String name;
    	private Date birthday;
    	private boolean sex;
    	public User(String name, Date birthday, boolean sex){
    		this.name = name;
    		this.birthday = birthday;
    		this.sex = sex;
    	}
    	public String getName() {
    		return name;
    	}
    	public void setName(String name) {
    		this.name = name;
    	}
    	public Date getBirthday() {
    		return birthday;
    	}
    	public void setBirthday(Date birthday) {
    		this.birthday = birthday;
    	}
    	public boolean isSex() {
    		return sex;
    	}
    	public void setSex(boolean sex) {
    		this.sex = sex;
    	}
    	public String toString(){
    		return JSON.toJSONString(this);
    	}
    }
    
  • 工程依赖引用的listen-rpc-common工程

  • 主要放置服务端和客户端共用的组件,而且这些组件可以被其他服务包共用,所以要抽取出来
  • listen-rpc-common工程目录结构图
  • listen-rpc-common工程pom.xml文件
  • <?xml version="1.0"?>
    <project
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
    	xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    	<modelVersion>4.0.0</modelVersion>
    	<parent>
    		<groupId>listen</groupId>
    		<artifactId>listen-parent</artifactId>
    		<version>0.0.1-SNAPSHOT</version>
    	</parent>
    	<artifactId>listen-rpc-common</artifactId>
    	<packaging>jar</packaging>
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    	</properties>
    	<dependencies>
    		<!-- Netty -->
    		<dependency>
    			<groupId>io.netty</groupId>
    			<artifactId>netty-all</artifactId>
    			<version>4.0.24.Final</version>
    		</dependency>
    		<!-- Protostuff -->
    		<dependency>
    			<groupId>com.dyuproject.protostuff</groupId>
    			<artifactId>protostuff-core</artifactId>
    			<version>1.0.8</version>
    		</dependency>
    		<dependency>
    			<groupId>com.dyuproject.protostuff</groupId>
    			<artifactId>protostuff-runtime</artifactId>
    			<version>1.0.8</version>
    		</dependency>
    		<!-- Objenesis -->
    		<dependency>
    			<groupId>org.objenesis</groupId>
    			<artifactId>objenesis</artifactId>
    			<version>2.1</version>
    		</dependency>
    		<!-- Spring -->
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context</artifactId>
    			<version>3.2.12.RELEASE</version>
    		</dependency>
    	</dependencies>
    </project>
    

     

  • listen-rpc-common工程RpcService注解

  • package com.listen.rpc.annotation;
    
    import java.lang.annotation.ElementType;
    import java.lang.annotation.Retention;
    import java.lang.annotation.RetentionPolicy;
    import java.lang.annotation.Target;
    
    import org.springframework.stereotype.Component;
    
    @Target({ElementType.TYPE})
    @Retention(RetentionPolicy.RUNTIME)
    @Component // 表明可被 Spring 扫描
    public @interface RpcService {
    
    	Class<?> value();
    	
    }
    

    listen-rpc-common工程Constant接口

  • package com.listen.rpc.common;
    
    public interface Constant {
    
    	int ZK_SESSION_TIMEOUT = 5000;
    
    	//在创建数据节点前,先用zkCli.sh客户端连接上服务端,查看目前存在的数据节点,
    	//把下面的/zookeeper/quota改为你自己的,/zookeeper/quota是我自己Zookeeper的节点
        String ZK_REGISTRY_PATH = "/zookeeper/quota";
        String ZK_DATA_PATH = ZK_REGISTRY_PATH + "/data";
    }
    

    listen-rpc-common工程RpcDecoder解码类

  • package com.listen.rpc.common;
    
    import java.util.List;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ByteToMessageDecoder;
    
    public class RpcDecoder extends ByteToMessageDecoder{
    
    	private Class<?> genericClass;
    
        public RpcDecoder(Class<?> genericClass) {
            this.genericClass = genericClass;
        }
    
        @Override
        public final 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 obj = SerializationUtil.deserialize(data, genericClass);
            out.add(obj);
        }
    }
    

    listen-rpc-common工程RpcEncoder解码类

  • package com.listen.rpc.common;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    
    public class RpcEncoder extends MessageToByteEncoder {
    
    	private Class<?> genericClass;
    
        public RpcEncoder(Class<?> genericClass) {
            this.genericClass = genericClass;
        }
    
        @Override
        public void encode(ChannelHandlerContext ctx, Object in, ByteBuf out) throws Exception {
            if (genericClass.isInstance(in)) {
                byte[] data = SerializationUtil.serialize(in);
                out.writeInt(data.length);
                out.writeBytes(data);
            }
        }
    }
    

    listen-rpc-common工程RpcRequest请求类

  • package com.listen.rpc.common;
    
    public class RpcRequest {
    
    	private String requestId;
        private String className;
        private String methodName;
        private Class<?>[] parameterTypes;
        private Object[] parameters;
    	public String getRequestId() {
    		return requestId;
    	}
    	public void setRequestId(String requestId) {
    		this.requestId = requestId;
    	}
    	public String getClassName() {
    		return className;
    	}
    	public void setClassName(String className) {
    		this.className = className;
    	}
    	public String getMethodName() {
    		return methodName;
    	}
    	public void setMethodName(String methodName) {
    		this.methodName = methodName;
    	}
    	public Class<?>[] getParameterTypes() {
    		return parameterTypes;
    	}
    	public void setParameterTypes(Class<?>[] parameterTypes) {
    		this.parameterTypes = parameterTypes;
    	}
    	public Object[] getParameters() {
    		return parameters;
    	}
    	public void setParameters(Object[] parameters) {
    		this.parameters = parameters;
    	}
        
    }
    

    listen-rpc-common工程RpcResponse响应类

  • package com.listen.rpc.common;
    
    public class RpcResponse {
    
    	private String requestId;
        private Throwable error;
        private Object result;
    	public String getRequestId() {
    		return requestId;
    	}
    	public void setRequestId(String requestId) {
    		this.requestId = requestId;
    	}
    	public Throwable getError() {
    		return error;
    	}
    	public void setError(Throwable error) {
    		this.error = error;
    	}
    	public Object getResult() {
    		return result;
    	}
    	public void setResult(Object result) {
    		this.result = result;
    	}
        
    }
    

    listen-rpc-common工程SerializationUtil序列化反序列化类

  • 服务接口实现工程listen-rpc-service-impl

  • 主要放置listen-rpc-service接口的实现类,这个工程会依赖公共RPC服务工程listen-rpc-server,listen-rpc-server主要是连接Zookeeper实现服务的注册与发现
  • listen-rpc-server工程的目录结构图

  • listen-rpc-server工程pom.xml
  • <?xml version="1.0"?>
    <project
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
    	xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    	<modelVersion>4.0.0</modelVersion>
    	<parent>
    		<groupId>listen</groupId>
    		<artifactId>listen-parent</artifactId>
    		<version>0.0.1-SNAPSHOT</version>
    	</parent>
    	<artifactId>listen-rpc-server</artifactId>
    	<packaging>jar</packaging>
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    	</properties>
    	<dependencies>
    		<dependency>
    			<groupId>listen</groupId>
    			<artifactId>listen-rpc-common</artifactId>
    			<version>0.0.1-SNAPSHOT</version>
    		</dependency>
    
    		<!-- ZooKeeper -->
    		<dependency>
    			<groupId>org.apache.zookeeper</groupId>
    			<artifactId>zookeeper</artifactId>
    			<version>3.4.6</version>
    		</dependency>
    
    		<!-- Apache Commons Collections -->
    		<dependency>
    			<groupId>org.apache.commons</groupId>
    			<artifactId>commons-collections4</artifactId>
    			<version>4.0</version>
    		</dependency>
    	</dependencies>
    	
    	<build>
    		<plugins>
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-source-plugin</artifactId>
    				<version>2.2.1</version>
    				<executions>
    					<execution>
    						<id>attach-sources</id>
    						<goals>
    							<goal>jar</goal>
    						</goals>
    					</execution>
    				</executions>
    			</plugin>
    			<plugin>
    				<groupId>org.apache.tomcat.maven</groupId>
    				<artifactId>tomcat7-maven-plugin</artifactId>
    				<version>2.2</version>
    				<configuration>
    					<port>8000</port>
    					<path>/</path>
    				</configuration>
    			</plugin>
    		</plugins>
    	</build>
    </project>
    

     

  • listen-rpc-server工程RpcServer服务启动类

  • package com.listen.rpc.server;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import org.apache.commons.collections4.MapUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.BeansException;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.ApplicationContextAware;
    
    import com.listen.rpc.annotation.RpcService;
    import com.listen.rpc.common.RpcDecoder;
    import com.listen.rpc.common.RpcEncoder;
    import com.listen.rpc.common.RpcRequest;
    import com.listen.rpc.common.RpcResponse;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    public class RpcServer implements ApplicationContextAware, InitializingBean{
    
    	private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);
    
        private String serverAddress;
        private ServiceRegistry serviceRegistry;
    
        private Map<String, Object> handlerMap = new HashMap<String, Object>(); // 存放接口名与服务对象之间的映射关系
    
        public RpcServer(String serverAddress) {
            this.serverAddress = serverAddress;
        }
    
        public RpcServer(String serverAddress, ServiceRegistry serviceRegistry) {
            this.serverAddress = serverAddress;
            this.serviceRegistry = serviceRegistry;
        }
    
        public void setApplicationContext(ApplicationContext ctx) throws BeansException {
            Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(RpcService.class); // 获取所有带有 RpcService 注解的 Spring Bean
            if (MapUtils.isNotEmpty(serviceBeanMap)) {
                for (Object serviceBean : serviceBeanMap.values()) {
                    String interfaceName = serviceBean.getClass().getAnnotation(RpcService.class).value().getName();
                    handlerMap.put(interfaceName, serviceBean);
                }
            }
        }
    
        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) 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[] array = serverAddress.split(":");
                String host = array[0];
                int port = Integer.parseInt(array[1]);
    
                ChannelFuture future = bootstrap.bind(host, port).sync();
                LOGGER.debug("server started on port {}", port);
    
                if (serviceRegistry != null) {
                    serviceRegistry.register(serverAddress); // 注册服务地址
                }
    
                future.channel().closeFuture().sync();
            } finally {
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
        }
    }
    

    listen-rpc-server工程ServiceRegistry服务注册类

  • package com.listen.rpc.server;
    
    import java.io.IOException;
    import java.util.concurrent.CountDownLatch;
    
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooDefs;
    import org.apache.zookeeper.ZooKeeper;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.listen.rpc.common.Constant;
    
    public class ServiceRegistry {
    
    	private static final Logger LOGGER = LoggerFactory.getLogger(ServiceRegistry.class);
    
        private CountDownLatch latch = new CountDownLatch(1);
    
        private String registryAddress;
    
        public ServiceRegistry(String registryAddress) {
            this.registryAddress = registryAddress;
        }
    
        public void register(String data) {
            if (data != null) {
                ZooKeeper zk = connectServer();
                if (zk != null) {
                    createNode(zk, data);
                }
            }
        }
    
        private ZooKeeper connectServer() {
            ZooKeeper zk = null;
            try {
                zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {
                    public void process(WatchedEvent event) {
                        if (event.getState() == Event.KeeperState.SyncConnected) {
                            latch.countDown();
                        }
                    }
                });
                latch.await();
            } catch (IOException e) {
                LOGGER.error("", e);
            } catch (InterruptedException e){
            	LOGGER.error("", e);
            }
            return zk;
        }
    
        private void createNode(ZooKeeper zk, String data) {
            try {
                byte[] bytes = data.getBytes();
                String path = zk.create(Constant.ZK_DATA_PATH, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
                LOGGER.debug("create zookeeper node ({} => {})", path, data);
            } catch (KeeperException e) {
            	e.printStackTrace();
                LOGGER.error("", e);
            } catch (InterruptedException e){
            	LOGGER.error("", e);
            }
        }
    }
    

    listen-rpc-server工程RpcHandler请求统一处理类

  • listen-rpc-service-impl工程目录结构图

  • listen-rpc-service-impl服务实现工程pom.xml
  • <?xml version="1.0"?>
    <project
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
    	xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    	<modelVersion>4.0.0</modelVersion>
    	<parent>
    		<groupId>listen</groupId>
    		<artifactId>listen-parent</artifactId>
    		<version>0.0.1-SNAPSHOT</version>
    	</parent>
    	<artifactId>listen-rpc-service-impl</artifactId>
    	<packaging>war</packaging>
    	<dependencies>
    		<dependency>
    			<groupId>listen</groupId>
    			<artifactId>listen-rpc-server</artifactId>
    			<version>0.0.1-SNAPSHOT</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-webmvc</artifactId>
    			<version>3.2.12.RELEASE</version>
    		</dependency>
    		<dependency>
    			<groupId>javax.servlet</groupId>
    			<artifactId>servlet-api</artifactId>
    			<version>2.5</version>
    			<scope>provided</scope>
    		</dependency>
    		<dependency>
    			<groupId>listen</groupId>
    			<artifactId>listen-rpc-service</artifactId>
    			<version>0.0.1-SNAPSHOT</version>
    		</dependency>
    	</dependencies>
    	<build>
    		<plugins>
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-source-plugin</artifactId>
    				<version>2.2.1</version>
    				<executions>
    					<execution>
    						<id>attach-sources</id>
    						<goals>
    							<goal>jar</goal>
    						</goals>
    					</execution>
    				</executions>
    			</plugin>
    			<plugin>
    				<groupId>org.apache.tomcat.maven</groupId>
    				<artifactId>tomcat7-maven-plugin</artifactId>
    				<version>2.2</version>
    				<configuration>
    					<port>8000</port>
    					<path>/</path>
    				</configuration>
    			</plugin>
    		</plugins>
    	</build>
    </project>
    

     

  • listen-rpc-service-impl工程HelloServiceImpl服务实现类

  • package com.listen.rpc.service.impl;
    
    import java.util.ArrayList;
    import java.util.Calendar;
    import java.util.Date;
    import java.util.List;
    
    import com.listen.rpc.annotation.RpcService;
    import com.listen.rpc.entity.User;
    import com.listen.rpc.service.IHelloService;
    
    @RpcService(IHelloService.class)// 指定远程接口  使用 RpcService注解定义在服务接口的实现类上,需要对该实现类指定远程接口,因为实现类可能会实现多个接口,一定要告诉框架哪个才是远程接口。
    public class HelloServiceImpl implements IHelloService {
    
    	public String hello(String name) {
    		String result = "hello" + name;
    		System.out.println(result);
    		return result;
    	}
    
    	@Override
    	public User getUser(String name) {
    		User user = new User(name, new Date(), true);
    		return user;
    	}
    
    	@Override
    	public List<User> getUsers(int size) {
    		List<User> list = new ArrayList<User>();
    		User user = null;
    		String name = "foo";
    		Date birthday = new Date();
    		Calendar cal = Calendar.getInstance();
    		cal.setTime(birthday);
    		for(int i = 0; i < size; i++){
    			cal.add(Calendar.DAY_OF_MONTH, 1);
    			user = new User(name, cal.getTime(), i%2==0 ? true : false);
    			list.add(user);
    		}
    		return list;
    	}
    
    	@Override
    	public User updateUser(User user) {
    		user.setName(user.getName() + "-update");
    		return user;
    	}
    
    }
    

    listen-rpc-service-impl工程配置文件config.properties

  • # ZooKeeper 服务器
    registry.address=127.0.0.1:2181
    
    # RPC 服务器
    server.address=127.0.0.1:8000
    #以上配置表明:连接本地的 ZooKeeper 服务器,并在 8000 端口上发布 RPC 服务。

    listen-rpc-service-impl工程spring.xml配置文件

  • <?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop"
        xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p" xmlns:util="http://www.springframework.org/schema/util" xmlns:jdbc="http://www.springframework.org/schema/jdbc"
        xmlns:cache="http://www.springframework.org/schema/cache"
        xsi:schemaLocation="
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context.xsd
        http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/tx
        http://www.springframework.org/schema/tx/spring-tx.xsd
        http://www.springframework.org/schema/jdbc
        http://www.springframework.org/schema/jdbc/spring-jdbc-3.1.xsd
        http://www.springframework.org/schema/cache
        http://www.springframework.org/schema/cache/spring-cache-3.1.xsd
        http://www.springframework.org/schema/aop
        http://www.springframework.org/schema/aop/spring-aop.xsd
        http://www.springframework.org/schema/util
        http://www.springframework.org/schema/util/spring-util.xsd">
    
        <!-- 自动扫描web包 ,将带有注解的类 纳入spring容器管理 -->
        <context:component-scan base-package="com.listen.rpc"></context:component-scan>
        <context:property-placeholder location="classpath:config.properties"/>
        
        <!-- 配置服务注册组件 -->
        <bean id="serviceRegistry" class="com.listen.rpc.server.ServiceRegistry">
            <constructor-arg name="registryAddress" value="${registry.address}"/>
        </bean>
    
        <!-- 配置 RPC 服务器 -->
        <bean id="rpcServer" class="com.listen.rpc.server.RpcServer">
            <constructor-arg name="serverAddress" value="${server.address}"/>
            <constructor-arg name="serviceRegistry" ref="serviceRegistry"/>
        </bean>
    
    </beans>
    

     

至此,服务端代码编写完毕,接下来开始编写客户端代码,客户端我打算用springMVC实现服务的测试

  • 客户端工程listen-rpc-client

  • 主要是springMVC实现服务端服务接口的调用测试
  • listen-rpc-client工程目录结构图
  •  
  • listen-rpc-client工程的pom.xml文件
  • <?xml version="1.0"?>
    <project
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
    	xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    	<modelVersion>4.0.0</modelVersion>
    	<parent>
    		<groupId>listen</groupId>
    		<artifactId>listen-parent</artifactId>
    		<version>0.0.1-SNAPSHOT</version>
    	</parent>
    	<artifactId>listen-rpc-client</artifactId>
    	<packaging>war</packaging>
    	<dependencies>
    		<dependency>
    			<groupId>listen</groupId>
    			<artifactId>listen-rpc-common</artifactId>
    			<version>0.0.1-SNAPSHOT</version>
    		</dependency>
    		<dependency>
    			<groupId>listen</groupId>
    			<artifactId>listen-rpc-service</artifactId>
    			<version>0.0.1-SNAPSHOT</version>
    		</dependency>
    		<dependency>
    			<groupId>javax.servlet</groupId>
    			<artifactId>javax.servlet-api</artifactId>
    			<version>3.1.0</version>
    			<scope>provided</scope>
    		</dependency>
    		<!-- Spring -->
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context</artifactId>
    			<version>3.2.12.RELEASE</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-webmvc</artifactId>
    			<version>3.2.12.RELEASE</version>
    		</dependency>
    
    		<!-- ZooKeeper -->
    		<dependency>
    			<groupId>org.apache.zookeeper</groupId>
    			<artifactId>zookeeper</artifactId>
    			<version>3.4.6</version>
    		</dependency>
    
    		<!-- CGLib -->
    		<dependency>
    			<groupId>cglib</groupId>
    			<artifactId>cglib</artifactId>
    			<version>3.1</version>
    		</dependency>
    	</dependencies>
    	<build>
    		<finalName>listen-rpc-client</finalName>
    		<plugins>
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-source-plugin</artifactId>
    				<version>2.2.1</version>
    				<executions>
    					<execution>
    						<id>attach-sources</id>
    						<goals>
    							<goal>jar</goal>
    						</goals>
    					</execution>
    				</executions>
    			</plugin>
    			<plugin>
    				<groupId>org.apache.tomcat.maven</groupId>
    				<artifactId>tomcat7-maven-plugin</artifactId>
    				<version>2.2</version>
    				<configuration>
    					<port>8080</port>
    					<path>/</path>
    				</configuration>
    			</plugin>
    		</plugins>
    	</build>
    </project>
    

     

  • listen-rpc-client工程HelloController类
  • package com.listen.rpc.controller;
    
    import java.util.Date;
    import java.util.List;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    
    import com.listen.rpc.client.RpcProxy;
    import com.listen.rpc.entity.User;
    import com.listen.rpc.service.IHelloService;
    
    @Controller
    public class HelloController {
    
    	@Autowired
    	private RpcProxy rpcProxy;
    	@RequestMapping("/hello")
    	public void hello(String name){
    		IHelloService service = rpcProxy.create(IHelloService.class);
    		String result = service.hello(name);
    		System.out.println(result);
    	}
    	
    	@RequestMapping("/getUser")
    	public void getUser(String name){
    		IHelloService service = rpcProxy.create(IHelloService.class);
    		System.out.println(service.getUser(name).toString());
    	}
    	
    	@RequestMapping("/getUsers")
    	public void getUsers(int size){
    		IHelloService service = rpcProxy.create(IHelloService.class);
    		List<User> list = service.getUsers(size);
    		for(User user : list){
    			System.out.println(user.toString());
    		}
    	}
    	
    	@RequestMapping("/updateUser")
    	public void updateUser(String name){
    		User user = new User(name, new Date(), true);
    		IHelloService service = rpcProxy.create(IHelloService.class);
    		user = service.updateUser(user);
    		System.out.println(user.toString());
    	}
    }
    

    listen-rpc-client工程代理类RpcProxy

  • package com.listen.rpc.client;
    
    import java.lang.reflect.Method;
    import java.util.UUID;
    
    import com.listen.rpc.common.RpcRequest;
    import com.listen.rpc.common.RpcResponse;
    
    import net.sf.cglib.proxy.InvocationHandler;
    import net.sf.cglib.proxy.Proxy;
    
    public class RpcProxy {
    
    	private String serverAddress;
        private ServiceDiscovery serviceDiscovery;
    
        public RpcProxy(String serverAddress) {
            this.serverAddress = serverAddress;
        }
    
        public RpcProxy(ServiceDiscovery serviceDiscovery) {
            this.serviceDiscovery = serviceDiscovery;
        }
    
        @SuppressWarnings("unchecked")
        public <T> T create(Class<?> interfaceClass) {
            return (T) Proxy.newProxyInstance(
                interfaceClass.getClassLoader(),
                new Class<?>[]{interfaceClass},
                new InvocationHandler() {
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        RpcRequest request = new RpcRequest(); // 创建并初始化 RPC 请求
                        request.setRequestId(UUID.randomUUID().toString());
                        request.setClassName(method.getDeclaringClass().getName());
                        request.setMethodName(method.getName());
                        request.setParameterTypes(method.getParameterTypes());
                        request.setParameters(args);
    
                        if (serviceDiscovery != null) {
                            serverAddress = serviceDiscovery.discover(); // 发现服务
                        }
    
                        String[] array = serverAddress.split(":");
                        String host = array[0];
                        int port = Integer.parseInt(array[1]);
    
                        RpcClient client = new RpcClient(host, port); // 初始化 RPC 客户端
                        RpcResponse response = client.send(request); // 通过 RPC 客户端发送 RPC 请求并获取 RPC 响应
    
                        if (response.getError() != null) {
                            throw response.getError();
                        } else {
                            return response.getResult();
                        }
                    }
                }
            );
        }
    }
    

    listen-rpc-client工程客户端类RpcClient

  • package com.listen.rpc.client;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.listen.rpc.common.RpcDecoder;
    import com.listen.rpc.common.RpcEncoder;
    import com.listen.rpc.common.RpcRequest;
    import com.listen.rpc.common.RpcResponse;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {
    
    	private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);
    
        private String host;
        private int port;
    
        private RpcResponse response;
    
        private final Object obj = new Object();
    
        public RpcClient(String host, int port) {
            this.host = host;
            this.port = port;
        }
    
        @Override
        public void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
            this.response = response;
    
            synchronized (obj) {
                obj.notifyAll(); // 收到响应,唤醒线程
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            LOGGER.error("client caught exception", cause);
            ctx.close();
        }
    
        public RpcResponse send(RpcRequest request) throws Exception {
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline()
                                .addLast(new RpcEncoder(RpcRequest.class)) // 将 RPC 请求进行编码(为了发送请求)
                                .addLast(new RpcDecoder(RpcResponse.class)) // 将 RPC 响应进行解码(为了处理响应)
                                .addLast(RpcClient.this); // 使用 RpcClient 发送 RPC 请求
                        }
                    })
                    .option(ChannelOption.SO_KEEPALIVE, true);
    
                ChannelFuture future = bootstrap.connect(host, port).sync();
                future.channel().writeAndFlush(request).sync();
    
                synchronized (obj) {
                    obj.wait(); // 未收到响应,使线程等待
                }
    
                if (response != null) {
                    future.channel().closeFuture().sync();
                }
                return response;
            } finally {
                group.shutdownGracefully();
            }
        }
    
    }
    

    listen-rpc-client工程服务发现类ServiceDiscovery

  • package com.listen.rpc.client;
    
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.CountDownLatch;
    
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.listen.rpc.common.Constant;
    
    import io.netty.util.internal.ThreadLocalRandom;
    
    public class ServiceDiscovery {
    
    	private static final Logger LOGGER = LoggerFactory.getLogger(ServiceDiscovery.class);
    
        private CountDownLatch latch = new CountDownLatch(1);
    
        private volatile List<String> dataList = new ArrayList<String>();
    
        private String registryAddress;
    
        public ServiceDiscovery(String registryAddress) {
            this.registryAddress = registryAddress;
    
            ZooKeeper zk = connectServer();
            if (zk != null) {
                watchNode(zk);
            }
        }
    
        public String discover() {
            String data = null;
            int size = dataList.size();
            if (size > 0) {
                if (size == 1) {
                    data = dataList.get(0);
                    LOGGER.debug("using only data: {}", data);
                } else {
                    data = dataList.get(ThreadLocalRandom.current().nextInt(size));
                    LOGGER.debug("using random data: {}", data);
                }
            }
            return data;
        }
    
        private ZooKeeper connectServer() {
            ZooKeeper zk = null;
            try {
                zk = new ZooKeeper(registryAddress, Constant.ZK_SESSION_TIMEOUT, new Watcher() {
                    public void process(WatchedEvent event) {
                        if (event.getState() == Event.KeeperState.SyncConnected) {
                            latch.countDown();
                        }
                    }
                });
                latch.await();
            } catch (IOException e) {
                LOGGER.error("", e);
            } catch (InterruptedException e){
            	LOGGER.error("", e);
            }
            return zk;
        }
    
        private void watchNode(final ZooKeeper zk) {
            try {
                List<String> nodeList = zk.getChildren(Constant.ZK_REGISTRY_PATH, new Watcher() {
                    public void process(WatchedEvent event) {
                        if (event.getType() == Event.EventType.NodeChildrenChanged) {
                            watchNode(zk);
                        }
                    }
                });
                List<String> dataList = new ArrayList<String>();
                for (String node : nodeList) {
                    byte[] bytes = zk.getData(Constant.ZK_REGISTRY_PATH + "/" + node, false, null);
                    dataList.add(new String(bytes));
                }
                LOGGER.debug("node data: {}", dataList);
                this.dataList = dataList;
            } catch (KeeperException e) {
            	e.printStackTrace();
                LOGGER.error("", e);
            } catch (InterruptedException e){
            	LOGGER.error("", e);
            }
        }
    }
    

    listen-rpc-client工程配置文件config.properties

  • # ZooKeeper 服务器
    registry.address=127.0.0.1:2181
    #以上配置表明:连接本地的 ZooKeeper 服务器,并在 8000 端口上发布 RPC 服务。

    listen-rpc-client工程spring.xml配置文件

  • <?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
        xmlns:aop="http://www.springframework.org/schema/aop" xmlns:p="http://www.springframework.org/schema/p"
        xsi:schemaLocation="http://www.springframework.org/schema/beans
               http://www.springframework.org/schema/beans/spring-beans-3.2.xsd
               http://www.springframework.org/schema/context
               http://www.springframework.org/schema/context/spring-context-3.2.xsd
                http://www.springframework.org/schema/aop
               http://www.springframework.org/schema/aop/spring-aop-3.2.xsd">
        <!-- 启用spring mvc 注解 -->
        <context:annotation-config />
        <!-- 完成请求和注解POJO的映射 -->
        <bean class="org.springframework.web.servlet.mvc.annotation.AnnotationMethodHandlerAdapter" />
        <!-- 对转向页面的路径解析。prefix:前缀, suffix:后缀 -->
        <bean class="org.springframework.web.servlet.view.InternalResourceViewResolver" p:prefix="/WEB-INF/jsp/" p:suffix=".jsp" />
        <context:component-scan base-package="com.listen.rpc" />
        <context:property-placeholder location="classpath:config.properties"/>
    
        <!-- 配置服务发现组件 -->
        <bean id="serviceDiscovery" class="com.listen.rpc.client.ServiceDiscovery">
            <constructor-arg name="registryAddress" value="${registry.address}"/>
        </bean>
    
        <!-- 配置 RPC 代理 -->
        <bean id="rpcProxy" class="com.listen.rpc.client.RpcProxy">
            <constructor-arg name="serviceDiscovery" ref="serviceDiscovery"/>
        </bean>
        
    </beans>
    

     

至此,客户端代码编写完毕,接下来就是测试了

测试服务接口

  • 启动服务端 tomcat7:run;勾选Skip test
  •  
  • 启动客户端 tomcat7:run;勾选Skip test
  • 访问地址测试
  • http://localhost:8080/hello?name=aaa 返回helloaaa
  • http://localhost:8080/getUser?name=aaa 返回{"birthday":1464258865975,"name":"aaa","sex":true}
  • http://localhost:8080/getUsers?size=5 返回{"birthday":1464345304049,"name":"foo","sex":true}
    {"birthday":1464431704049,"name":"foo","sex":false}
    {"birthday":1464518104049,"name":"foo","sex":true}
    {"birthday":1464604504049,"name":"foo","sex":false}
    {"birthday":1464690904049,"name":"foo","sex":true}
  • http://localhost:8080/updateUser?name=kaka 返回{"birthday":1464258949048,"name":"kaka-update","sex":true}

至此说明服务可以正常发布及调用。

需要注意的点为Zookeeper数据地址的获取,可参考 Windows下Zookeeper的简单配置

 

工程源码地址:

https://git.oschina.net/git.listen/listen-parent

https://git.oschina.net/git.listen/listen-rpc-common

https://git.oschina.net/git.listen/listen-rpc-server

https://git.oschina.net/git.listen/listen-rpc-service

https://git.oschina.net/git.listen/listen-rpc-service-impl

https://git.oschina.net/git.listen/listen-rpc-client

© 著作权归作者所有

L
粉丝 13
博文 28
码字总数 53225
作品 0
宝山
程序员
私信 提问
加载中

评论(12)

f
fjsh123
一言难尽,代码中的坑颇多了,粘包拆包问题在大请求量的时候就出现问题了,你这里是每次请求重建链接所以看不出来,哎
f
fjsh123
最近几天看了下,做demo还好,但如果到企业应用级别还差的有点远了
f
fjsh123
你好,看了下部分源码,写的不错,赞一个,想在你基础上做下修改调整后重新发布可以嘛?
j
jayevan

引用来自“chenjunhui00”的评论

少了序列化,反序列化,handler类,这几个怎么补全?

引用来自“Listen_ing”的评论

源码已经上传到码云,请看博客后面的码云地址
为什么不用github😓
zhjh256
zhjh256
欢迎使用spider RPC,non-trivial系统的最佳选择http://www.cnblogs.com/zhjh256/category/911531.html
L
Listen_ing 博主

引用来自“贺伟锋”的评论

你好,我下载了你的"Spring+Netty+Protostuff+ZooKeeper" 源码,不知道你的服务端是怎么启动的,ChannelFuture future = bootstrap.bind(host, port).sync();这行代码是阻塞式的,tomcat是不会走的。请问你是怎么把这个服务端启动起来的?
只需要启动服务实现工程和测试的客户端工程就可以,服务注册和发现是用的Zookeeper
贺伟锋
贺伟锋
你好,我下载了你的"Spring+Netty+Protostuff+ZooKeeper" 源码,不知道你的服务端是怎么启动的,ChannelFuture future = bootstrap.bind(host, port).sync();这行代码是阻塞式的,tomcat是不会走的。请问你是怎么把这个服务端启动起来的?
L
Listen_ing 博主

引用来自“IAcoupman”的评论

能共享下源码吗 谢谢79
源码已经上传到码云,请看博客后面的码云地址
L
Listen_ing 博主

引用来自“chenjunhui00”的评论

少了序列化,反序列化,handler类,这几个怎么补全?
源码已经上传到码云,请看博客后面的码云地址
chenjunhui00
chenjunhui00
少了序列化,反序列化,handler类,这几个怎么补全?
XXL-RPC v1.2.2,分布式服务框架,内置原生轻量级注册中心

Release Notes 1、默认通讯方案切换为 Netty,可选方案依赖均调整为 provided 类型;提供强制依赖最小精简选型组合 "netty + hessian + 无注册中心(推荐采用:XXL-RPC原生注册中心)"; 2、X...

许雪里
2018/11/27
1K
9
轻量级rpc框架--bbossgroups RPC

bbossgroups RPC 是基于bbossaop的轻量级rpc框架,感兴趣的朋友可以用一用。bbossgroups提供的RPC框架是bboss aop子项目中一个子模块,具有以下特点: 1.支持多种通讯协议jms,jgroups,min...

bboss
2010/04/07
2.2K
0
基于 Swoole 的轻量级高性能框架 swoolefy 1.0.6 发布

swoolefy是基于swoole实现的轻量级高性能框架,框架支持http,websocket,udp服务器,以及基于tcp实现可扩展的rpc服务,同时支持composer包方式安装部署项目。基于实用,swoolefy抽象Event事...

bingcool
2018/06/11
1K
5
XXL-REGISTRY v1.0.0 发布,轻量级分布式服务注册中心

Release Notes 1、轻量级:基于DB与磁盘文件,只需要提供一个DB实例即可,无第三方依赖; 2、实时性:借助内部广播机制,新服务上线、下线,可以在1s内推送给客户端; 3、数据同步:注册中心...

许雪里
2018/12/03
915
5
bbossgroups RPC 基于aop的轻量级rpc框架

bbossgroups RPC 基于aop的轻量级rpc框架 收藏 bbossgroups RPC 是基于bbossaop的轻量级rpc框架,感兴趣的朋友可以用一用。bbossgroups提供的RPC框架是bboss aop子项目中一个子模块,具有以下...

yin_bp
2010/04/06
30
0

没有更多内容

加载失败,请刷新页面

加载更多

自定义ApiBoot Logging链路以及单元ID生成策略

ApiBoot Logging会为每一个请求都对应创建链路编号(TraceID)以及单元编号(SpanID),用于归类每一次请求日志,通过一个链路下日志单元的Parent SpanID可以进行上下级关系的梳理。 前文回顾...

恒宇少年
26分钟前
11
0
浅谈 Application 和 activity

对于 在 Application初始化一些变量,为什么不可以放在activity 或者其他的组件里呢? 这里就根据个人的理解来讲述一下,欢迎补充指正。 首先 activity 是以栈的形式出现,一个app应用会有多...

MrLins
26分钟前
11
0
Allegro的脚本文件内容里都有哪些

小伙伴们在使用Allegro的时候是否经常用到脚本文件夹呢?scr的用法其实可真不简单。。。 首先脚本文件的运行模式就存在很多种,比如不提示错误信息,不弹出确认对画框(这样很有利于我们执行...

demyar
27分钟前
21
0
微信升级外链管理规范,「砍一刀帮我加速」要被禁止了

原创: 蒋鸿昌 首发:「知晓程序」公众号 - 最好的微信新商业媒体 几天前,知名互联网评论人阑夕模仿皮尤研究中心(Pew Research Center)在美国做的互联网通识调查问卷,做了一份中文版问卷...

知晓云
28分钟前
15
0
CentOS 7接投影仪

我将一台安装着CentOS 7图形界面的惠普笔记本电脑当桌面使用。最近,想要连接投影仪时却遇到了问题。笔记本有一个HDMI接口。我买了一个HDMI---->VGA的转接线,连上笔记本电脑后,屏幕一直在闪...

大别阿郎
31分钟前
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部