接着上一篇往下讲，DubboBootstrap.start() 方法里面，initialize之后，就开始暴露服务了。
exportServices()
private void exportServices() {
configManager.getServices().forEach(sc -> {
// TODO, compatible with ServiceConfig.export()
ServiceConfig serviceConfig = (ServiceConfig) sc;
serviceConfig.setBootstrap(this);
if (exportAsync) {
ExecutorService executor = executorRepository.getServiceExporterExecutor();
Future<?> future = executor.submit(() -> {
sc.export();
exportedServices.add(sc);
});
asyncExportingFutures.add(future);
} else {
sc.export();
exportedServices.add(sc);
}
});
}
遍历刚刚在定义的服务，这里有同步暴露和异步线程池暴露服务。
export
public synchronized void export() {
if (bootstrap == null) {
bootstrap = DubboBootstrap.getInstance();
bootstrap.initialize();
}
// 检查配置
checkAndUpdateSubConfigs();
//初始化一个服务的元数据信息
serviceMetadata.setVersion(getVersion());
serviceMetadata.setGroup(getGroup());
serviceMetadata.setDefaultGroup(getGroup());
serviceMetadata.setServiceType(getInterfaceClass());
serviceMetadata.setServiceInterfaceName(getInterface());
serviceMetadata.setTarget(getRef());
// 可以配置是否暴露服务
if (!shouldExport()) {
return;
}
// 延迟暴露服务
if (shouldDelay()) {
DELAY_EXPORT_EXECUTOR.schedule(this::doExport, getDelay(), TimeUnit.MILLISECONDS);
} else {
// 暴露服务方法
doExport();
}
exported();
}
可以发现，demo里面classic的启动方式就是直接调用这个方法
checkAndUpdateSubConfigs
private void checkAndUpdateSubConfigs() {
// 如果providerConfig不为空并且protocols或configCenter为空，用provider的配置
completeCompoundConfigs();
// 检查providerConfig是否为空，是的话用默认配置
checkDefault();
// 同上检查Protocol
checkProtocol();
// list默认是空的，可以实现自己的扩展
List<ConfigInitializer> configInitializers = ExtensionLoader.getExtensionLoader(ConfigInitializer.class)
.getActivateExtension(URL.valueOf("configInitializer://"), (String[]) null);
configInitializers.forEach(e -> e.initServiceConfig(this));
// 如果协议不是inJvm，检查address的合法性
if (!isOnlyInJvm()) {
checkRegistry();
}
// 上一章讲过，把环境变量的一些值注入到这个类对应的一些字段里面
this.refresh();
if (StringUtils.isEmpty(interfaceName)) {
throw new IllegalStateException("<dubbo:service interface=\"\" /> interface not allow null!");
}
// 泛化调用
if (ref instanceof GenericService) {
interfaceClass = GenericService.class;
if (StringUtils.isEmpty(generic)) {
generic = Boolean.TRUE.toString();
}
} else {
try {
interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
.getContextClassLoader());
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e.getMessage(), e);
}
// 校验类是否存在是否接口，方法配置是否合法（方法是否存在等
checkInterfaceAndMethods(interfaceClass, getMethods());
// 校验实现类是否存在，是否是接口的子类
checkRef();
generic = Boolean.FALSE.toString();
}
// 已废弃
if (local != null) {
if ("true".equals(local)) {
local = interfaceName + "Local";
}
Class<?> localClass;
try {
localClass = ClassUtils.forNameWithThreadContextClassLoader(local);
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e.getMessage(), e);
}
if (!interfaceClass.isAssignableFrom(localClass)) {
throw new IllegalStateException("The local implementation class " + localClass.getName() + " not implement interface " + interfaceName);
}
}
// 本地存根
if (stub != null) {
if ("true".equals(stub)) {
stub = interfaceName + "Stub";
}
Class<?> stubClass;
try {
stubClass = ClassUtils.forNameWithThreadContextClassLoader(stub);
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e.getMessage(), e);
}
if (!interfaceClass.isAssignableFrom(stubClass)) {
throw new IllegalStateException("The stub implementation class " + stubClass.getName() + " not implement interface " + interfaceName);
}
}
// 校验是否为空等
checkStubAndLocal(interfaceClass);
ConfigValidationUtils.checkMock(interfaceClass, this);
ConfigValidationUtils.validateServiceConfig(this);
// 可以实现自己的一些后置处理器
postProcessConfig();
}
doExport -> doExportUrls
private void doExportUrls() {
// 存service的类
ServiceRepository repository = ApplicationModel.getServiceRepository();
// 接口方法的详细信息
ServiceDescriptor serviceDescriptor = repository.registerService(getInterfaceClass());
// 存provider
repository.registerProvider(
getUniqueServiceName(),
ref,
serviceDescriptor,
this,
serviceMetadata
);
// 获取到注册中心的URL
List<URL> registryURLs = ConfigValidationUtils.loadRegistries(this, true);
// 遍历协议
for (ProtocolConfig protocolConfig : protocols) {
// 接口路径：org.apache.dubbo.demo.DemoService
String pathKey = URL.buildKey(getContextPath(protocolConfig)
.map(p -> p + "/" + path)
.orElse(path), group, version);
// In case user specified path, register service one more time to map it to path.
repository.registerService(pathKey, interfaceClass);
// TODO, uncomment this line once service key is unified
serviceMetadata.setServiceKey(pathKey);
// 根据协议暴露服务
doExportUrlsFor1Protocol(protocolConfig, registryURLs);
}
}
doExportUrlsFor1Protocol方法内容很多，最重要的是下面几行代码
// 忽略一些配置相关的代码
Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(EXPORT_KEY, url.toFullString()));
DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
Exporter<?> exporter = PROTOCOL.export(wrapperInvoker);
exporters.add(exporter);
1、PROXY_FACTORY.getInvoker：把实现类包装成invoker
默认是JavassistProxyFactory
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
// 为目标类创建 Wrapper
final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
// 创建匿名 Invoker 类对象，并实现 doInvoke 方法。
return new AbstractProxyInvoker<T>(proxy, type, url) {
@Override
protected Object doInvoke(T proxy, String methodName,
Class<?>[] parameterTypes,
Object[] arguments) throws Throwable {
// 调用 Wrapper 的 invokeMethod 方法，invokeMethod 最终会调用目标方法
return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
}
};
}
生成的Wrapper类具有通用的invokeMethod方法，从这个方法可以看出，不需要依赖服务提供者的api也可以实现接口调用，实际上泛化调用就是这样做的。
用arthas看一下动态生成的Wrapper类，代码很简单。
public class Wrapper1 extends Wrapper implements ClassGenerator.DC {
public static String[] pns;
public static Map pts;
public static String[] mns;
public static String[] dmns;
public static Class[] mts0;
public static Class[] mts1;
public String[] getDeclaredMethodNames() {
return dmns;
}
public Class getPropertyType(String string) {
return (Class)pts.get(string);
}
public String[] getMethodNames() {
return mns;
}
public Object invokeMethod(Object object, String string, Class[] arrclass, Object[] arrobject) throws InvocationTargetException {
DemoServiceImpl demoServiceImpl;
try {
demoServiceImpl = (DemoServiceImpl)object;
}
catch (Throwable throwable) {
throw new IllegalArgumentException(throwable);
}
try {
if ("sayHello".equals(string) && arrclass.length == 1) {
return demoServiceImpl.sayHello((String)arrobject[0]);
}
if ("sayHelloAsync".equals(string) && arrclass.length == 1) {
return demoServiceImpl.sayHelloAsync((String)arrobject[0]);
}
}
catch (Throwable throwable) {
throw new InvocationTargetException(throwable);
}
throw new NoSuchMethodException(new StringBuffer().append("Not found method \"").append(string).append("\" in class org.apache.dubbo.demo.provider.DemoServiceImpl.").toString());
}
public String[] getPropertyNames() {
return pns;
}
public boolean hasProperty(String string) {
return pts.containsKey(string);
}
public Object getPropertyValue(Object object, String string) {
try {
DemoServiceImpl demoServiceImpl = (DemoServiceImpl)object;
}
catch (Throwable throwable) {
throw new IllegalArgumentException(throwable);
}
throw new NoSuchPropertyException(new StringBuffer().append("Not found property \"").append(string).append("\" field or getter method in class org.apache.dubbo.demo.provider.DemoServiceImpl.").toString());
}
public void setPropertyValue(Object object, String string, Object object2) {
try {
DemoServiceImpl demoServiceImpl = (DemoServiceImpl)object;
}
catch (Throwable throwable) {
throw new IllegalArgumentException(throwable);
}
throw new NoSuchPropertyException(new StringBuffer().append("Not found property \"").append(string).append("\" field or setter method in class org.apache.dubbo.demo.provider.DemoServiceImpl.").toString());
}
}
再看返回的AbstractProxyInvoker类，核心是invoke方法。第一步就是wrapper的invokeMethod方法调用具体实现类，第二包装返回对象，因为有可能是异步调用，用CompletableFuture包装了result。
public Result invoke(Invocation invocation) throws RpcException {
try {
// wrapper的invokeMethod方法
Object value = doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments());
CompletableFuture<Object> future = wrapWithFuture(value);
CompletableFuture<AppResponse> appResponseFuture = future.handle((obj, t) -> {
AppResponse result = new AppResponse();
if (t != null) {
if (t instanceof CompletionException) {
result.setException(t.getCause());
} else {
result.setException(t);
}
} else {
result.setValue(obj);
}
return result;
});
return new AsyncRpcResult(appResponseFuture, invocation);
} catch (InvocationTargetException e) {
if (RpcContext.getContext().isAsyncStarted() && !RpcContext.getContext().stopAsync()) {
logger.error("Provider async started, but got an exception from the original method, cannot write the exception back to consumer because an async result may have returned the new thread.", e);
}
return AsyncRpcResult.newDefaultAsyncResult(null, e.getTargetException(), invocation);
} catch (Throwable e) {
throw new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
2、PROTOCOL.export(wrapperInvoker): 开启服务，注册到注册中心
默认是org.apache.dubbo.registry.client.RegistryProtocol#export方法
主要关注doLocalExport和register两块代码
// 省略其他代码
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker, providerUrl);
// url to registry
final Registry registry = getRegistry(originInvoker);
final URL registeredProviderUrl = getUrlToRegistry(providerUrl, registryUrl);
// decide if we need to delay publish
boolean register = providerUrl.getParameter(REGISTER_KEY, true);
if (register) {
register(registryUrl, registeredProviderUrl);
}
// 省略其他代码
1、doLocalExport，protocol 默认是dubbo协议
private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker, URL providerUrl) {
String key = getCacheKey(originInvoker);
return (ExporterChangeableWrapper<T>) bounds.computeIfAbsent(key, s -> {
Invoker<?> invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl);
return new ExporterChangeableWrapper<>((Exporter<T>) protocol.export(invokerDelegate), originInvoker);
});
}
一路导航到org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#export
关键方法openServer -> createServer , 这里就是启动一个netty的服务端
private ProtocolServer createServer(URL url) {
// 省略代码
ExchangeServer server;
try {
server = Exchangers.bind(url, requestHandler);
} catch (RemotingException e) {
throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
}
str = url.getParameter(CLIENT_KEY);
if (str != null && str.length() > 0) {
Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
if (!supportedTypes.contains(str)) {
throw new RpcException("Unsupported client type: " + str);
}
}
return new DubboProtocolServer(server);
}
重点看Exchangers.bind(url, requestHandler) 怎么创建服务，这里传的requestHandler是DubboProtocol 里面的匿名内部类，消费者请求进来最后都是在requestHandler.reply 处理的
public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handler == null) {
throw new IllegalArgumentException("handler == null");
}
url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
return getExchanger(url).bind(url, handler);
}
getExchanger(url) 默认是org.apache.dubbo.remoting.exchange.support.header.HeaderExchanger。
可以看到Handler对象一层套一层，处理的时候也就是从最外层到最里面的过程，类似责任链。重点关注HeaderExchangeHandler，核心逻辑都在里面。
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}
先往下看Transporters.bind方法，返回一个RemotingServer，即远程服务。
新版本getTransporter返回对象默认是netty4。
public static RemotingServer bind(URL url, ChannelHandler... handlers) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handlers == null || handlers.length == 0) {
throw new IllegalArgumentException("handlers == null");
}
ChannelHandler handler;
if (handlers.length == 1) {
handler = handlers[0];
} else {
handler = new ChannelHandlerDispatcher(handlers);
}
return getTransporter().bind(url, handler);
}
往下org.apache.dubbo.remoting.transport.netty4.NettyTransporter的bind方法
public RemotingServer bind(URL url, ChannelHandler handler) throws RemotingException {
return new NettyServer(url, handler);
}
NettyServer 的构造方法，调用父类的构造方法，这里又包装了一层handle，MultiMessageHandler->HeartbeatHandler->AllChannelHandler->handler
事件分发器默认是AllChannelHandler，后续再看这段逻辑
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
super(ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME), ChannelHandlers.wrap(handler, url));
}
...
// org.apache.dubbo.remoting.transport.dispatcher.ChannelHandlers的warp方法
protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
.getAdaptiveExtension().dispatch(handler, url)));
}
父类构造方法关键逻辑doOpen() 交给子类实现
public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
super(url, handler);
localAddress = getUrl().toInetSocketAddress();
String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost());
int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort());
if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
bindIp = ANYHOST_VALUE;
}
bindAddress = new InetSocketAddress(bindIp, bindPort);
this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS);
this.idleTimeout = url.getParameter(IDLE_TIMEOUT_KEY, DEFAULT_IDLE_TIMEOUT);
try {
doOpen();
if (logger.isInfoEnabled()) {
logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
}
} catch (Throwable t) {
throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName()
+ " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
}
executor = executorRepository.createExecutorIfAbsent(url);
}
这里就是标准的netty服务端启动逻辑，pipeline里面加了四个处理器。
- decoder 解码
- encoder 编码
- server-idle-handler 存活检测
- handler 业务处理
@Override
protected void doOpen() throws Throwable {
bootstrap = new ServerBootstrap();
bossGroup = NettyEventLoopFactory.eventLoopGroup(1, "NettyServerBoss");
workerGroup = NettyEventLoopFactory.eventLoopGroup(
getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
"NettyServerWorker");
final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this);
channels = nettyServerHandler.getChannels();
bootstrap.group(bossGroup, workerGroup)
.channel(NettyEventLoopFactory.serverSocketChannelClass())
.option(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
.childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// FIXME: should we use getTimeout()?
int idleTimeout = UrlUtils.getIdleTimeout(getUrl());
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
if (getUrl().getParameter(SSL_ENABLED_KEY, false)) {
ch.pipeline().addLast("negotiation",
SslHandlerInitializer.sslServerHandler(getUrl(), nettyServerHandler));
}
ch.pipeline()
.addLast("decoder", adapter.getDecoder())
.addLast("encoder", adapter.getEncoder())
.addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS))
.addLast("handler", nettyServerHandler);
}
});
// bind
ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
channelFuture.syncUninterruptibly();
channel = channelFuture.channel();
}
至此，doLocalExport方法结束。总结就是开启一个netty服务，层层包装了handle处理业务逻辑。梳理一下handle调用链，后续服务调用用得上。
- org.apache.dubbo.remoting.transport.netty4.NettyServerHandler
- org.apache.dubbo.remoting.transport.netty4.NettyServer
- org.apache.dubbo.remoting.transport.MultiMessageHandler
- org.apache.dubbo.remoting.exchange.support.header.HeartbeatHandler
- org.apache.dubbo.remoting.transport.dispatcher.all.AllChannelHandler
- org.apache.dubbo.remoting.transport.DecodeHandler
- org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeHandler
- DubboProtocol里面的匿名类 requestHandler
回到RegistryProtocol的export方法，本地服务开启之后，要注册服务到注册中心
2、register
register 只有两行，registryFactory 默认是ZookeeperRegistryFactory，不过有包装类RegistryFactoryWrapper
private void register(URL registryUrl, URL registeredProviderUrl) {
Registry registry = registryFactory.getRegistry(registryUrl);
registry.register(registeredProviderUrl);
}
进来就是RegistryFactoryWrapper的getRegistry，加了一层监听对象，
public Registry getRegistry(URL url) {
return new ListenerRegistryWrapper(registryFactory.getRegistry(url),
Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(RegistryServiceListener.class)
.getActivateExtension(url, "registry.listeners")));
}
registry.register 实际就是下面这段代码，多了一个监听步骤，默认是没有监听器的
public void register(URL url) {
try {
registry.register(url);
} finally {
if (CollectionUtils.isNotEmpty(listeners)) {
RuntimeException exception = null;
for (RegistryServiceListener listener : listeners) {
if (listener != null) {
try {
listener.onRegister(url);
} catch (RuntimeException t) {
logger.error(t.getMessage(), t);
exception = t;
}
}
}
if (exception != null) {
throw exception;
}
}
}
}
继续registry.register(url)，就是ZookeeperRegistr的方法了。
ZookeeperRegistr 构造方法创建了zk的客户端连接zkClient，register方法在父类FailbackRegistry里面，然后调子类的doRegister。然后用在zk上面创建临时节点的过程，往下不分析了。
public void doRegister(URL url) {
try {
zkClient.create(toUrlPath(url), url.getParameter(DYNAMIC_KEY, true));
} catch (Throwable e) {
throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
启动成功，完结撒花~
服务提供者启动简单总结：
- 初始化配置
- 遍历每个serviceConfig
- 暴露服务：
- 本地开启一个netty 服务端，开始监听请求（openserver只会启动一次
- 把服务信息注册到zk上面
