HelloworldServer.scala
object HelloworldServer {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
val securityManager = new SecurityManager(conf)
val rpcEnv: RpcEnv = RpcEnv.create("hello-service", "localhost", 52345, conf, securityManager)
val helloEndpoint: RpcEndpoint = new HelloEndpoint(rpcEnv)
rpcEnv.setupEndpoint("hello-service", helloEndpoint)
rpcEnv.awaitTermination()
}
}
class HelloEndpoint(override val rpcEnv: RpcEnv) extends RpcEndpoint {
override def onStart(): Unit = {
println("start hello endpoint")
}
override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
case SayHi(msg) => {
//println(s"receive $msg")
context.reply(s"Hi, $msg")
}
case SayBye(msg) => {
//println(s"receive $msg")
context.reply(s"Bye, $msg")
}
}
override def onStop(): Unit = {
println("stop hello endpoint")
}
}
HelloworldClient.scala
object HelloworldClient {
def main(args: Array[String]): Unit = {
syncCall()
}
def syncCall() = {
// 初始化RpcEnv环境
val conf = new SparkConf
// 这里的rpc环境主机需要指定本机,端口号可以任意指定
val rpcEnv = RpcEnv.create("hello-client", "localhost", 52346, conf, new SecurityManager(conf))
// 根据Server端IP + Port获取后端服务的引用,得到的是RpcEndpointRef类型对象
val endpointRef = rpcEnv.setupEndpointRef(RpcAddress("localhost", 52345), "hello-service")
// 1、客户端异步请求
// 客户端通过RpcEndpointRef#ask方法异步访问服务端,服务端通过RpcEndpoint#receiveAndReply方法获取到该请求后处理
val future = endpointRef.ask[String](SayBye("I am zhangsan"))
// 客户端请求成功/失败时的处理方法
future.onComplete {
case scala.util.Success(value) => println(s"Got the result = $value")
case scala.util.Failure(e) => println(s"Got error: $e")
}
// 客户端等待超时时间
Await.result(future, Duration("5s"))
// 2、客户端同步请求
val resp = endpointRef.askSync[String](SayHi("hehe"))
print(resp)
}
}