网站首页优化公司,苏州公司建站,电商运营培训班,网站开发 质保金公司需要使用java技术栈接入一套自定义的通讯协议#xff0c;所以参考下dubbo的实现原理。
consumer
主要使用ThreadlessExecutor实现全consumer的全双工通讯。consumer创建本次请求的requestId用于将response和request匹配。 然后分以下几步完成一次请求发送并接收结果所以参考下dubbo的实现原理。
consumer
主要使用ThreadlessExecutor实现全consumer的全双工通讯。consumer创建本次请求的requestId用于将response和request匹配。 然后分以下几步完成一次请求发送并接收结果 槽发送消息前将用于接收结果的executor放到一个map中存储 发送消息调用netty发送 挂起线程等待response 异步接收结果netty线程接收到结果后唤醒之前挂起的线程
槽
创造一个ThreadlessExecutor并将其存储到DefaultFuture.FUTURES中。之后消息返回后会从DefaultFuture.FUTURES取出ThrealessExecutor并设置结果。
构建ThreadlessExecutor
init:56, ThreadlessExecutor (org.apache.dubbo.common.threadpool)
getCallbackExecutor:190, AbstractInvoker (org.apache.dubbo.rpc.protocol)
doInvoke:103, DubboInvoker (org.apache.dubbo.rpc.protocol.dubbo)
invoke:163, AbstractInvoker (org.apache.dubbo.rpc.protocol)
invoke:52, AsyncToSyncInvoker (org.apache.dubbo.rpc.protocol)
invoke:89, MonitorFilter (org.apache.dubbo.monitor.support)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:51, FutureFilter (org.apache.dubbo.rpc.protocol.dubbo.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:69, ConsumerContextFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:78, ListenerInvokerWrapper (org.apache.dubbo.rpc.listener)
invoke:56, InvokerWrapper (org.apache.dubbo.rpc.protocol)
doInvoke:82, FailoverClusterInvoker (org.apache.dubbo.rpc.cluster.support)
invoke:259, AbstractClusterInvoker (org.apache.dubbo.rpc.cluster.support)
intercept:47, ClusterInterceptor (org.apache.dubbo.rpc.cluster.interceptor)
invoke:92, AbstractCluster$InterceptorInvokerNode (org.apache.dubbo.rpc.cluster.support.wrapper)
invoke:82, MockClusterInvoker (org.apache.dubbo.rpc.cluster.support.wrapper)
invoke:74, InvokerInvocationHandler (org.apache.dubbo.rpc.proxy)
sayHello:-1, proxy0 (org.apache.dubbo.common.bytecode)
main:43, Consumer (org.apache.dubbo.samples.mock)存储到DefaultFuture.FUTURES
init:85, DefaultFuture (org.apache.dubbo.remoting.exchange.support)
newFuture:108, DefaultFuture (org.apache.dubbo.remoting.exchange.support)
request:133, HeaderExchangeChannel (org.apache.dubbo.remoting.exchange.support.header)
request:95, HeaderExchangeClient (org.apache.dubbo.remoting.exchange.support.header)
request:91, ReferenceCountExchangeClient (org.apache.dubbo.rpc.protocol.dubbo)
doInvoke:105, DubboInvoker (org.apache.dubbo.rpc.protocol.dubbo)
invoke:163, AbstractInvoker (org.apache.dubbo.rpc.protocol)
invoke:52, AsyncToSyncInvoker (org.apache.dubbo.rpc.protocol)
invoke:89, MonitorFilter (org.apache.dubbo.monitor.support)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:51, FutureFilter (org.apache.dubbo.rpc.protocol.dubbo.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:69, ConsumerContextFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:78, ListenerInvokerWrapper (org.apache.dubbo.rpc.listener)
invoke:56, InvokerWrapper (org.apache.dubbo.rpc.protocol)
doInvoke:82, FailoverClusterInvoker (org.apache.dubbo.rpc.cluster.support)
invoke:259, AbstractClusterInvoker (org.apache.dubbo.rpc.cluster.support)
intercept:47, ClusterInterceptor (org.apache.dubbo.rpc.cluster.interceptor)
invoke:92, AbstractCluster$InterceptorInvokerNode (org.apache.dubbo.rpc.cluster.support.wrapper)
invoke:82, MockClusterInvoker (org.apache.dubbo.rpc.cluster.support.wrapper)
invoke:74, InvokerInvocationHandler (org.apache.dubbo.rpc.proxy)
sayHello:-1, proxy0 (org.apache.dubbo.common.bytecode)
main:43, Consumer (org.apache.dubbo.samples.mock)发送消息
业务线程会将消息发送到NioEventLoop的任务队列中
offerTask:330, SingleThreadEventExecutor (io.netty.util.concurrent)
addTask:321, SingleThreadEventExecutor (io.netty.util.concurrent)
execute:765, SingleThreadEventExecutor (io.netty.util.concurrent)
safeExecute:1007, AbstractChannelHandlerContext (io.netty.channel)
write:825, AbstractChannelHandlerContext (io.netty.channel)
writeAndFlush:794, AbstractChannelHandlerContext (io.netty.channel)
writeAndFlush:831, AbstractChannelHandlerContext (io.netty.channel)
writeAndFlush:1071, DefaultChannelPipeline (io.netty.channel)
writeAndFlush:300, AbstractChannel (io.netty.channel)
send:162, NettyChannel (org.apache.dubbo.remoting.transport.netty4)
send:178, AbstractClient (org.apache.dubbo.remoting.transport)
send:53, AbstractPeer (org.apache.dubbo.remoting.transport)
request:135, HeaderExchangeChannel (org.apache.dubbo.remoting.exchange.support.header)
request:95, HeaderExchangeClient (org.apache.dubbo.remoting.exchange.support.header)
request:91, ReferenceCountExchangeClient (org.apache.dubbo.rpc.protocol.dubbo)
doInvoke:105, DubboInvoker (org.apache.dubbo.rpc.protocol.dubbo)
invoke:163, AbstractInvoker (org.apache.dubbo.rpc.protocol)
invoke:52, AsyncToSyncInvoker (org.apache.dubbo.rpc.protocol)
invoke:89, MonitorFilter (org.apache.dubbo.monitor.support)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:51, FutureFilter (org.apache.dubbo.rpc.protocol.dubbo.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:69, ConsumerContextFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:78, ListenerInvokerWrapper (org.apache.dubbo.rpc.listener)
invoke:56, InvokerWrapper (org.apache.dubbo.rpc.protocol)
doInvoke:82, FailoverClusterInvoker (org.apache.dubbo.rpc.cluster.support)
invoke:259, AbstractClusterInvoker (org.apache.dubbo.rpc.cluster.support)
intercept:47, ClusterInterceptor (org.apache.dubbo.rpc.cluster.interceptor)
invoke:92, AbstractCluster$InterceptorInvokerNode (org.apache.dubbo.rpc.cluster.support.wrapper)
invoke:82, MockClusterInvoker (org.apache.dubbo.rpc.cluster.support.wrapper)
invoke:74, InvokerInvocationHandler (org.apache.dubbo.rpc.proxy)
sayHello:-1, proxy0 (org.apache.dubbo.common.bytecode)
main:43, Consumer (org.apache.dubbo.samples.mock)挂起线程
waitAndDrain:89, ThreadlessExecutor (org.apache.dubbo.common.threadpool)
get:179, AsyncRpcResult (org.apache.dubbo.rpc)
invoke:61, AsyncToSyncInvoker (org.apache.dubbo.rpc.protocol)
invoke:89, MonitorFilter (org.apache.dubbo.monitor.support)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:51, FutureFilter (org.apache.dubbo.rpc.protocol.dubbo.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:69, ConsumerContextFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:78, ListenerInvokerWrapper (org.apache.dubbo.rpc.listener)
invoke:56, InvokerWrapper (org.apache.dubbo.rpc.protocol)
doInvoke:82, FailoverClusterInvoker (org.apache.dubbo.rpc.cluster.support)
invoke:259, AbstractClusterInvoker (org.apache.dubbo.rpc.cluster.support)
intercept:47, ClusterInterceptor (org.apache.dubbo.rpc.cluster.interceptor)
invoke:92, AbstractCluster$InterceptorInvokerNode (org.apache.dubbo.rpc.cluster.support.wrapper)
invoke:82, MockClusterInvoker (org.apache.dubbo.rpc.cluster.support.wrapper)
invoke:74, InvokerInvocationHandler (org.apache.dubbo.rpc.proxy)
sayHello:-1, proxy0 (org.apache.dubbo.common.bytecode)
main:43, Consumer (org.apache.dubbo.samples.mock)异步接收结果
netty线程中接收结果并唤醒之前阻塞的业务线程
execute:138, ThreadlessExecutor (org.apache.dubbo.common.threadpool)
received:62, AllChannelHandler (org.apache.dubbo.remoting.transport.dispatcher.all)
received:90, HeartbeatHandler (org.apache.dubbo.remoting.exchange.support.header)
received:43, MultiMessageHandler (org.apache.dubbo.remoting.transport)
received:147, AbstractPeer (org.apache.dubbo.remoting.transport)
channelRead:83, NettyClientHandler (org.apache.dubbo.remoting.transport.netty4)
invokeChannelRead:362, AbstractChannelHandlerContext (io.netty.channel)
invokeChannelRead:348, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:340, AbstractChannelHandlerContext (io.netty.channel)
channelRead:286, IdleStateHandler (io.netty.handler.timeout)
invokeChannelRead:362, AbstractChannelHandlerContext (io.netty.channel)
invokeChannelRead:348, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:340, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:310, ByteToMessageDecoder (io.netty.handler.codec)
channelRead:284, ByteToMessageDecoder (io.netty.handler.codec)
invokeChannelRead:362, AbstractChannelHandlerContext (io.netty.channel)
invokeChannelRead:348, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:340, AbstractChannelHandlerContext (io.netty.channel)
channelRead:1434, DefaultChannelPipeline$HeadContext (io.netty.channel)
invokeChannelRead:362, AbstractChannelHandlerContext (io.netty.channel)
invokeChannelRead:348, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:965, DefaultChannelPipeline (io.netty.channel)
read:163, AbstractNioByteChannel$NioByteUnsafe (io.netty.channel.nio)
processSelectedKey:647, NioEventLoop (io.netty.channel.nio)
processSelectedKeysOptimized:582, NioEventLoop (io.netty.channel.nio)
processSelectedKeys:499, NioEventLoop (io.netty.channel.nio)
run:461, NioEventLoop (io.netty.channel.nio)
run:884, SingleThreadEventExecutor$5 (io.netty.util.concurrent)
run:30, FastThreadLocalRunnable (io.netty.util.concurrent)
run:748, Thread (java.lang)provider
接收请求netty线程接收请求 切换线程将请求转给业务线程 返回结果业务线程处理之后调用channel.write返回结果
接收请求
received:62, AllChannelHandler (org.apache.dubbo.remoting.transport.dispatcher.all)
received:90, HeartbeatHandler (org.apache.dubbo.remoting.exchange.support.header)
received:43, MultiMessageHandler (org.apache.dubbo.remoting.transport)
received:147, AbstractPeer (org.apache.dubbo.remoting.transport)
channelRead:98, NettyServerHandler (org.apache.dubbo.remoting.transport.netty4)
invokeChannelRead:362, AbstractChannelHandlerContext (io.netty.channel)
invokeChannelRead:348, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:340, AbstractChannelHandlerContext (io.netty.channel)
channelRead:286, IdleStateHandler (io.netty.handler.timeout)
invokeChannelRead:362, AbstractChannelHandlerContext (io.netty.channel)
invokeChannelRead:348, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:340, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:310, ByteToMessageDecoder (io.netty.handler.codec)
channelRead:284, ByteToMessageDecoder (io.netty.handler.codec)
invokeChannelRead:362, AbstractChannelHandlerContext (io.netty.channel)
invokeChannelRead:348, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:340, AbstractChannelHandlerContext (io.netty.channel)
channelRead:1434, DefaultChannelPipeline$HeadContext (io.netty.channel)
invokeChannelRead:362, AbstractChannelHandlerContext (io.netty.channel)
invokeChannelRead:348, AbstractChannelHandlerContext (io.netty.channel)
fireChannelRead:965, DefaultChannelPipeline (io.netty.channel)
read:163, AbstractNioByteChannel$NioByteUnsafe (io.netty.channel.nio)
processSelectedKey:647, NioEventLoop (io.netty.channel.nio)
processSelectedKeysOptimized:582, NioEventLoop (io.netty.channel.nio)
processSelectedKeys:499, NioEventLoop (io.netty.channel.nio)
run:461, NioEventLoop (io.netty.channel.nio)
run:884, SingleThreadEventExecutor$5 (io.netty.util.concurrent)
run:30, FastThreadLocalRunnable (io.netty.util.concurrent)
run:748, Thread (java.lang)切换线程
sayHello:31, DemoServiceImpl (org.apache.dubbo.samples.mock.impl)
invokeMethod:-1, Wrapper1 (org.apache.dubbo.common.bytecode)
doInvoke:47, JavassistProxyFactory$1 (org.apache.dubbo.rpc.proxy.javassist)
invoke:84, AbstractProxyInvoker (org.apache.dubbo.rpc.proxy)
invoke:56, DelegateProviderMetaDataInvoker (org.apache.dubbo.config.invoker)
invoke:56, InvokerWrapper (org.apache.dubbo.rpc.protocol)
invoke:52, ExceptionFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:89, MonitorFilter (org.apache.dubbo.monitor.support)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:46, TimeoutFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:77, TraceFilter (org.apache.dubbo.rpc.protocol.dubbo.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:129, ContextFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:152, GenericFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:38, ClassLoaderFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
invoke:41, EchoFilter (org.apache.dubbo.rpc.filter)
invoke:81, ProtocolFilterWrapper$1 (org.apache.dubbo.rpc.protocol)
reply:145, DubboProtocol$1 (org.apache.dubbo.rpc.protocol.dubbo)
handleRequest:100, HeaderExchangeHandler (org.apache.dubbo.remoting.exchange.support.header)
received:175, HeaderExchangeHandler (org.apache.dubbo.remoting.exchange.support.header)
received:51, DecodeHandler (org.apache.dubbo.remoting.transport)
run:57, ChannelEventRunnable (org.apache.dubbo.remoting.transport.dispatcher)
runWorker:1149, ThreadPoolExecutor (java.util.concurrent)
run:624, ThreadPoolExecutor$Worker (java.util.concurrent)
run:748, Thread (java.lang)返回结果 offerTask:327, SingleThreadEventExecutor (io.netty.util.concurrent)
addTask:321, SingleThreadEventExecutor (io.netty.util.concurrent)
execute:765, SingleThreadEventExecutor (io.netty.util.concurrent)
safeExecute:1007, AbstractChannelHandlerContext (io.netty.channel)
write:825, AbstractChannelHandlerContext (io.netty.channel)
writeAndFlush:794, AbstractChannelHandlerContext (io.netty.channel)
writeAndFlush:831, AbstractChannelHandlerContext (io.netty.channel)
writeAndFlush:1071, DefaultChannelPipeline (io.netty.channel)
writeAndFlush:300, AbstractChannel (io.netty.channel)
send:162, NettyChannel (org.apache.dubbo.remoting.transport.netty4)
send:98, HeaderExchangeChannel (org.apache.dubbo.remoting.exchange.support.header)
send:87, HeaderExchangeChannel (org.apache.dubbo.remoting.exchange.support.header)
lambda$handleRequest$0:110, HeaderExchangeHandler (org.apache.dubbo.remoting.exchange.support.header)
accept:-1, 195186633 (org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeHandler$$Lambda$148)
uniWhenComplete:774, CompletableFuture (java.util.concurrent)
uniWhenCompleteStage:792, CompletableFuture (java.util.concurrent)
whenComplete:2153, CompletableFuture (java.util.concurrent)
whenComplete:110, CompletableFuture (java.util.concurrent)
handleRequest:101, HeaderExchangeHandler (org.apache.dubbo.remoting.exchange.support.header)
received:175, HeaderExchangeHandler (org.apache.dubbo.remoting.exchange.support.header)
received:51, DecodeHandler (org.apache.dubbo.remoting.transport)
run:57, ChannelEventRunnable (org.apache.dubbo.remoting.transport.dispatcher)
runWorker:1149, ThreadPoolExecutor (java.util.concurrent)
run:624, ThreadPoolExecutor$Worker (java.util.concurrent)
run:748, Thread (java.lang)细节
getPreferredExecutorService
不论consumer还是provider都是通过AllChannelHandler#received接收远程信息。区别在于consumer使用的是ThreadlessExecutor而provider使用的是ThreadPoolExecutor。
AllChannelHandler#received中调用的getPreferredExecutorService方法会根据接收的消息是否为response来决定使用哪个线程。
ThreadlessExecutorThreadPoolExecutor用途consumer接收provider接收初始化时机发送请求前接收请求存储位置DefaultFuture.FUTURESDefaultExecutorRepository.data存储keyrequestId端口业务线程业务线程IO线程业务线程 ThreadlessExecutor#execute由IO线程调用ThreadlessExecutor#waitAndDrain由业务线程调用 ThreadPoolExecutor存储位置相关细节可以看DefaultExecutorRepository#createExecutorIfAbsent requestId是标识每次请求的唯一id每构造一次Request实例都会由Request#newId生成一个新的 ThreadlessExecutor
ThreadlessExecutor中并没有启动新的线程其主要作用是将异步的远程访问转为同步。所以在调用堆栈中可以看到AsyncToSyncInvoker。 业务线程调用waitAndDrain时若远程provider尚未返回数据queue.take()会使线程挂起。
netty线程接收到provider的返回值之后会调用ThreadlessExecutor#excute将结果添加到queue中。于是之前因queue.take()挂起的线程可以继续运行了。
AllChannelHandler
调试的过程中发现AllChannelHandler#received方法比较关键consumer和provider的接收功能都会在此方法中从io线程转换到业务线程
于是想研究下AllChannelHandler主要从以下几个方面研究 实例化一般情况下一个实例要么在需要的时候被实例化要么在需要之前实例化并存到一个地方。所以想知道AllChannelHandler的实例属于哪一种如果是后者那么它存到哪里了 调用基于上面的问题如果是后一种情况如何从存的地方拿出来并使用的 维度每个provider都有一个AllChannelHandler实例还是每个端口一个实例
实例化 AllChannelHandler作为nettyServerHandler的一个属性存储 截图为providerconsumer原理与provider大体一致 调用 远程消息到来后会进入nettyServerHandler#channelRead方法中此方法会调用AllChannelHandler#received方法 截图为providerconsumer原理与provider大体一致 维度
AllChannelHandler的创建维度是ip:port若多个provider在同一个ip:port开启也只会创建一个AllChannelHandler换句话说就只创建一个netty服务端。多个consumer从同一个ip:port消费不同的服务也只会创建一个。
对于providerDubboProtocol#openServer中会根据ip:port缓存netty服务端避免重复创建。而AllChannelHandler的创建又与netty服务端的创建相对应所以AllChannelHandler的维度也是ip:port
对于consumerDubboProtocol#getSharedClient中会根据ip:port缓存netty客户端
