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
