Spark RPC

DedicateMessageLoop只服务于单个RPC endpoint

继承

RpcEndpointAddress

+ rpcAddress: RpcAddress+ name: String+ toString

InboxMessage代表消息的抽象

<<RpcHandler>>

OneWayMessage

sealed <<MessageLoop>>

- active : LinkedBlockingQueue[Inbox]# receiveLoopRunnable : Runnable

- receiveLoop() : Unit

InboxMessage是Inbox中的存储的对象，而Inbox的实现有这么多

Dispatcher

0..*

<<RpcEndpointRef>>

- maxRetries- retryWaitMs- defaultAskTimeout

NettyRpcEnvFactory

+ create(config:RpcEnvConfig):RpcEnv

1

NettyRpcEnv

TransportServer

- context:TransportContext- conf:TransportConf- appRpcHandler:RpcHandler- bootstraps:List[TransportServerBootstrap]- bootstrap:ServerBootstrap- channelFuture:ChannelFuture- port:int- pooledAllocator:PooledByteBufAllocator

RemoteProcessConnected

<<RpcEnv>>

- conf:SparkConf

RpcEnv通过startServer启动一个服务. 启动服务意味着什么？

SharedMessageLoop可用于多个RPC endpoint

RemoteProcessConnectionError

TransportClientFactory

Inbox

- endpoint: RpcEndpoint- endpointName: String# message : LinkedList[InboxMessage]

+ process(Dispatcher): Unit+ post(InboxMessage): Unit+stop(): Unit

 NettyRpcHandler

- dispatcher:Dispatcher- nettyEnv:NettyRpcEnv- streamManager:StreamManager

RpcAddress

+ host: String+ port: Int

+ hostPort: String+ toSparkURL: String

TransportClient

- channel: Channel- handler: TransportResponseHandler- clientId: String

OnStart$

OnStop$

RpcEnvConfig

+ conf: SparkConf+ name: StringbindAddress: StringadvertiseAddress: Stringport: IntsecurityManager: SecurityManagernumUsableCores: IntclientMode: Boolean

<<RpcEnvFactory>>

RemoteProcessDisconnected

RpcMessage

TransportContext

- chunkFetchWorkers:EventLoopGroup - conf:TransportConf- rpcHandler:RpcHandler- ENCODER:MessageEncoder - DECODER:MessageDecoder

<<RpcEndpoint>>

+ rpcEnv:RpcEnv

TransportConf

- conf:ConfigProvider- module:String