RPC
Remote Procedure Call：远程过程调用，通讯协议，允许一台计算机调用另一台计算机的程序
Client：服务消费者
Server：服务提供者
调用流程：
1、Client以本地调用方法调用服务
2、ClientStub将请求方法名、参数封装起来
3、ClientStub通过编码后发送
4、ServerStub接收到后通过编码发送给Server调用API
5、Server返回请求结果给ServerStub进行编码发送
6、ClientStub接收到请求结果进行编码后发送给Client
HelloService
/**
* 接口，服务提供方与服务消费方都需要实现
*/
public interface HelloService {
String hello(String msg);
}
HelloServerImpl
public class HelloServerImpl implements HelloService {
/**
* 当消费方调用该方法时，返回一个结果
*/
private static int count = 0;
@Override
public String hello(String msg) {
System.out.println("收到客户端消息="+msg);
//根据msg返回不同结果
if (msg != null) {
return "你好客户端，我收到了你发来的消息[" + msg + "] 第"+(++count)+"次调用";
} else {
return "你好客户端，我收到了你发来的消息";
}
}
}
NettyServer
/**
* 编写一个方法，完成对NettyServer初始化与启动
*/
public class NettyServer {
/**
* 不把核心方法对外暴露，外包括一层，不能启动方法传递不同参数
*/
public static void startServer(String hostname, int port) {
startServer0(hostname, port);
}
private static void startServer0(String hostname, int port) {
NioEventLoopGroup BossGroup = new NioEventLoopGroup(1);
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(BossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new StringEncoder());
pipeline.addLast(new StringDecoder());
pipeline.addLast(new NettyServerHandler());//自定义业务处理器
}
});
ChannelFuture channelFuture = serverBootstrap.bind(hostname, port).sync();
System.out.println("服务提供方开始提供服务...");
channelFuture.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
BossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
NettyServerHandler
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//获取客户端发送的消息，并调用我们的服务
System.out.println("msg=" + msg);
/**
* 客户端在调用服务器的api时，我们需要定义一个协议
* 如：
* 要求每次发消息时，必须以某个字符串开头“Hello#Cyrus#”
*/
if (msg.toString().startsWith(ClientBootStrap.providerName)) {
String result = new HelloServerImpl().hello(msg.toString().substring(msg.toString().lastIndexOf("#") + 1));
ctx.writeAndFlush(result);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
NettyClient
public class NettyClient {
//创建一个线程池，大小为本机cpu核数
private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
private static NettyClientHandler client; //自定义Handler
private static int count; //计数器
//创建一个方法使用代理模式，获取一个代理对象
public Object getBean(final Class<?> serviceClass, final String providerName) {
return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
new Class<?>[]{serviceClass},((proxy, method, args) -> {
//{}中的代码，客户端每次调用hello方法都会进入该代码中
System.out.println("(proxy, method, args)被进入...."+(++count));
if (client == null) {
initClient();
}
//设置要发给服务器端的消息 args[0]:传递的参数
client.setPara(providerName + args[0]);
/**
* 将client（自定义Handler）交给线程池去执行
* 线程执行client获取客户端返回的结果
*/
return executor.submit(client).get();
}));
}
//初始化客户端
private static void initClient() {
client = new NettyClientHandler();
NioEventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
try {
bootstrap.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new StringEncoder());
pipeline.addLast(new StringDecoder());
pipeline.addLast(client);
}
});
bootstrap.connect("127.0.0.1", 9999).sync();
} catch (Exception e) {
e.printStackTrace();
}
}
}
NettyClientHandler
public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {
private ChannelHandlerContext context;//上下文
private String result;//返回的结果
private String para;//客户端调用方法是，传入的参数
//与服务器建立连接后，就会被调用，该方法是第一个调用的(1)/一旦建立连接只创建一次
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("channelActive被调用");
context = ctx; //在其他方法中需要使用ctx
}
//收到服务器的数据后，调用方法(4)
@Override
public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("channelRead被调用");
result = msg.toString();
//随机唤醒等待的线程
notify();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
//被代理对象调用，真正发送数据给服务器，->wait等待被唤醒,唤醒后返回结果(3)(5)
@Override
public synchronized Object call() throws Exception {
System.out.println("call_1被调用");
context.writeAndFlush(para);
//进行wait
wait(); //等待channelRead()方法获取到服务器响应的结果后唤醒
System.out.println("call_2被调用");
return result; //服务方返回的结果
}
//(2)
void setPara(String para) {
System.out.println("setPara");
this.para = para;
}
}
ServerBootStrap
/**
* ServerBootStrap会启动一个服务提供者（NettyServer）
*/
public class ServerBootStrap {
public static void main(String[] args) {
//代码代理
NettyServer.startServer("127.0.0.1", 9999);
}
}
ClientBootStrap
public class ClientBootStrap {
public static final String providerName = "Hello#Cyrus#";
public static void main(String[] args) throws InterruptedException {
//创建一个消费者
NettyClient customer = new NettyClient();
//创建代理对象
HelloService service = (HelloService) customer.getBean(HelloService.class, providerName);
//通过代理对象调用服务提供者的方法（服务API）
for (; ; ) {
Thread.sleep(5 * 1000);
String result = service.hello("你好 dubbo...");
System.out.println("调用的结果 result="+result);
}
}
}