5.2实现服务端注册多个服务
5.1中我们注册完HelloService的实现类，服务器就自行启动了。针对上述问题，将服务的注册和服务器启动分离，使得服务端可以提供多个服务。
5.2.1服务注册表
我们需要一个容器，这个容器很简单，就是保存一些本地服务的信息，并且在获得一个服务名字的时候能够返回这个服务的信息。创建一个 ServiceRegistry 接口：
public interface ServiceRegistry { <T> void register(T service); Object getService(String serviceName); }
一目了然，一个register注册服务信息，一个getService获取服务信息。
我们新建一个默认的注册表类 DefaultServiceRegistry 来实现这个接口，提供服务注册服务，如下：
public class DefaultServiceRegistry implements ServiceRegistry { private static final Logger logger = LoggerFactory.getLogger(DefaultServiceRegistry.class); private final Map<String, Object> serviceMap = new ConcurrentHashMap<>(); private final Set<String> registeredService = ConcurrentHashMap.newKeySet(); @Override public synchronized <T> void register(T service) { String serviceName = service.getClass().getCanonicalName(); if(registeredService.contains(serviceName)) return; registeredService.add(serviceName); Class<?>[] interfaces = service.getClass().getInterfaces(); if(interfaces.length == 0) { throw new RpcException(RpcError.SERVICE_NOT_IMPLEMENT_ANY_INTERFACE); } for(Class<?> i : interfaces) { serviceMap.put(i.getCanonicalName(), service); } logger.info("向接口: {} 注册服务: {}", interfaces, serviceName); } @Override public synchronized Object getService(String serviceName) { Object service = serviceMap.get(serviceName); if(service == null) { throw new RpcException(RpcError.SERVICE_NOT_FOUND); } return service; } }
我们将服务名与提供服务的对象的对应关系保存在一个 ConcurrentHashMap 中，并且使用一个 Set 来保存当前有哪些对象已经被注册。在注册服务时，默认采用这个对象实现的接口的完整类名作为服务名，例如某个对象 A 实现了接口 X 和 Y，那么将 A 注册进去后，会有两个服务名 X 和 Y 对应于 A 对象。这种处理方式也就说明了某个接口只能有一个对象提供服务。
获得服务的对象就更简单了，直接去 Map 里查找就行了。
5.2.2其他处理
为了降低耦合度，我们不会把 ServiceRegistry 和某一个 RpcServer 绑定在一起，而是在创建 RpcServer 对象时，传入一个 ServiceRegistry 作为这个服务的注册表。
那么 RpcServer 这个类现在就变成了这样：
public class RpcServer { private static final Logger logger = LoggerFactory.getLogger(RpcServer.class); private static final int CORE_POOL_SIZE = 5; private static final int MAXIMUM_POOL_SIZE = 50; private static final int KEEP_ALIVE_TIME = 60; private static final int BLOCKING_QUEUE_CAPACITY = 100; private final ExecutorService threadPool; private RequestHandler requestHandler = new RequestHandler(); private final ServiceRegistry serviceRegistry; public RpcServer(ServiceRegistry serviceRegistry) { this.serviceRegistry = serviceRegistry; BlockingQueue<Runnable> workingQueue = new ArrayBlockingQueue<>(BLOCKING_QUEUE_CAPACITY); ThreadFactory threadFactory = Executors.defaultThreadFactory(); threadPool = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, workingQueue, threadFactory); } public void start(int port) { try (ServerSocket serverSocket = new ServerSocket(port)) { logger.info("服务器启动……"); Socket socket; while((socket = serverSocket.accept()) != null) { logger.info("消费者连接: {}:{}", socket.getInetAddress(), socket.getPort()); threadPool.execute(new RequestHandlerThread(socket, requestHandler, serviceRegistry)); } threadPool.shutdown(); } catch (IOException e) { logger.error("服务器启动时有错误发生:", e); } } }
在创建 RpcServer 时需要传入一个已经注册好服务的 ServiceRegistry，而原来的 register 方法也被改成了 start 方法，因为服务的注册已经不由 RpcServer 处理了，它只需要启动就行了。
而在每一个请求处理线程（RequestHandlerThread）中也就需要传入 ServiceRegistry 了，这里把处理线程和处理逻辑分成了两个类：RequestHandlerThread 只是一个线程，从ServiceRegistry 获取到提供服务的对象后，就会把 RpcRequest 和服务对象直接交给 RequestHandler 去处理，反射等过程被放到了 RequestHandler 里。
（1）处理线程类（工作线程 ）：RequesthandlerThread.java：处理线程，接收对象等
public class RequestHandlerThread implements Runnable { private static final Logger logger = LoggerFactory.getLogger(RequestHandlerThread.class); private Socket socket; private RequestHandler requestHandler; private ServiceRegistry serviceRegistry; public RequestHandlerThread(Socket socket, RequestHandler requestHandler, ServiceRegistry serviceRegistry) { this.socket = socket; this.requestHandler = requestHandler; this.serviceRegistry = serviceRegistry; } @Override public void run() { try (ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream()); ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream())) { RpcRequest rpcRequest = (RpcRequest) objectInputStream.readObject(); String interfaceName = rpcRequest.getInterfaceName(); Object service = serviceRegistry.getService(interfaceName); Object result = requestHandler.handle(rpcRequest, service); objectOutputStream.writeObject(RpcResponse.success(result)); objectOutputStream.flush(); } catch (IOException | ClassNotFoundException e) { logger.error("调用或发送时有错误发生：", e); } } }
（2）处理逻辑类：RequestHandler.java：通过反射进行方法调用
public class RequestHandler { private static final Logger logger = LoggerFactory.getLogger(RequestHandler.class); public Object handle(RpcRequest rpcRequest, Object service) { Object result = null; try { result = invokeTargetMethod(rpcRequest, service); logger.info("服务:{} 成功调用方法:{}", rpcRequest.getInterfaceName(), rpcRequest.getMethodName()); } catch (IllegalAccessException | InvocationTargetException e) { logger.error("调用或发送时有错误发生：", e); } return result; } private Object invokeTargetMethod(RpcRequest rpcRequest, Object service) throws IllegalAccessException, InvocationTargetException { Method method; try { method = service.getClass().getMethod(rpcRequest.getMethodName(), rpcRequest.getParamTypes()); } catch (NoSuchMethodException e) { return RpcResponse.fail(ResponseCode.METHOD_NOT_FOUND); } return method.invoke(service, rpcRequest.getParameters()); } }
在这种情况下，客户端完全不需要做任何改动。
ps：JDK创建线程池如果没有指定线程工厂则会使用了默认的线程工厂（DefaultThreadFactory）
static class DefaultThreadFactory implements ThreadFactory { private static final AtomicInteger poolNumber = new AtomicInteger(1); private final ThreadGroup group; private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix; DefaultThreadFactory() { // 声明安全管理器 SecurityManager s = System.getSecurityManager(); // 得到线程组 group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); // 线程名前缀,例如 "pool-1-thread-" namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } /** * 用于创建一个线程 */ public Thread newThread(Runnable r) { Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0); // 设置线程t为前台线程 if (t.isDaemon()) t.setDaemon(false); // 设置线程t的优先级为5 if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY); return t; } }
5.2.3测试
我比较懒，还是搞一个服务的，就是测试下兼容性而已（理论上没问题）。
服务端的测试：
public class TestServer { public static void main(String[] args) { HelloService helloService = new HelloServiceImpl(); // 创建服务注册的实现类注册服务，RpcServer注入服务，然后启动服务 ServiceRegistry serviceRegistry = new DefaultServiceRegistry(); serviceRegistry.register(helloService); RpcServer rpcServer = new RpcServer(serviceRegistry); rpcServer.start(9000); } }
客户端不需要变动。
执行后应当获得和上次相同的结果。
