Java利用Sping框架编写RPC远程过程调用服务的教程

public interface HelloService {

  String hello(String name);
}

@RpcService(HelloService.class) // 指定远程接口
public class HelloServiceImpl implements HelloService {

  @Override
  public String hello(String name) {
    return "Hello! " + name;
  }
}

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component // 表明可被 Spring 扫描
public @interface RpcService {

  Class<?> value();
}

# ZooKeeper 服务器
registry.address=127.0.0.1:2181

# RPC 服务器
server.address=127.0.0.1:8000

public class RpcBootstrap {

  public static void main(String[] args) {
    new ClassPathXmlApplicationContext("spring.xml");
  }
}

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() {
        @Override
        public void process(WatchedEvent event) {
          if (event.getState() == Event.KeeperState.SyncConnected) {
            latch.countDown();
          }
        }
      });
      latch.await();
    } catch (IOException | 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 | InterruptedException e) {
      LOGGER.error("", e);
    }
  }
}

public interface Constant {

  int ZK_SESSION_TIMEOUT = 5000;

  String ZK_REGISTRY_PATH = "/registry";
  String ZK_DATA_PATH = ZK_REGISTRY_PATH + "/data";
}

public class RpcServer implements ApplicationContextAware, InitializingBean {

  private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);

  private String serverAddress;
  private ServiceRegistry serviceRegistry;

  private Map handlerMap = new HashMap<>(); // 存放接口名与服务对象之间的映射关系

  public RpcServer(String serverAddress) {
    this.serverAddress = serverAddress;
  }

  public RpcServer(String serverAddress, ServiceRegistry serviceRegistry) {
    this.serverAddress = serverAddress;
    this.serviceRegistry = serviceRegistry;
  }

  @Override
  public void setApplicationContext(ApplicationContext ctx) throws BeansException {
    Map 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);
      }
    }
  }

  @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() {
          @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[] 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();
    }
  }
}

public class RpcRequest {

  private String requestId;
  private String className;
  private String methodName;
  private Class<?>[] parameterTypes;
  private Object[] parameters;

  // getter/setter...
}

public class RpcResponse {

  private String requestId;
  private Throwable error;
  private Object result;

  // getter/setter...
}

public class RpcDecoder extends ByteToMessageDecoder {

  private Class<?> genericClass;

  public RpcDecoder(Class<?> genericClass) {
    this.genericClass = genericClass;
  }

  @Override
  public void decode(ChannelHandlerContext ctx, ByteBuf in, List