Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feat/threadpool ext #1383

Merged
merged 23 commits into from
Jan 9, 2024
Merged
Changes from 14 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion all/pom.xml
Original file line number Diff line number Diff line change
@@ -56,7 +56,7 @@
<maven.compiler.target>1.8</maven.compiler.target>
<project.build.sourceEncoding>utf-8</project.build.sourceEncoding>
<slf4j.version>1.7.21</slf4j.version>
<sofa.common.tools.version>1.3.2</sofa.common.tools.version>
<sofa.common.tools.version>1.3.15</sofa.common.tools.version>
<javassist.version>3.29.2-GA</javassist.version>
<netty.version>4.1.44.Final</netty.version>
<hessian.version>3.5.0</hessian.version>
2 changes: 1 addition & 1 deletion bom/pom.xml
Original file line number Diff line number Diff line change
@@ -53,7 +53,7 @@
<junit.version>4.13.1</junit.version>
<!-- alipay libs -->
<bolt.version>1.6.6</bolt.version>
<sofa.common.tools.version>1.3.2</sofa.common.tools.version>
<sofa.common.tools.version>1.3.15</sofa.common.tools.version>
<tracer.version>3.0.8</tracer.version>
<lookout.version>1.4.1</lookout.version>
<!-- Build args -->
Original file line number Diff line number Diff line change
@@ -554,13 +554,16 @@ public ClientTransport getAvailableClientTransport(ProviderInfo providerInfo) {
transport = uninitializedConnections.get(providerInfo);
if (transport != null) {
// 未初始化则初始化
synchronized (this) {
providerLock.lock();
try {
transport = uninitializedConnections.get(providerInfo);
if (transport != null) {
initClientTransport(consumerConfig.getInterfaceId(), providerInfo, transport);
uninitializedConnections.remove(providerInfo);
}
return getAvailableClientTransport(providerInfo);
} finally {
providerLock.unlock();
}
}

Original file line number Diff line number Diff line change
@@ -21,6 +21,8 @@

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
* 规则id的配置形式:a,b,!c,d
@@ -49,18 +51,21 @@ public abstract class BeanIdMatchFilter extends Filter {
private List<String> excludeId;

private volatile boolean formatComplete;
private final Object formatLock = new Object();
protected final Lock lock = new ReentrantLock();

@Override
public boolean needToLoad(FilterInvoker invoker) {
AbstractInterfaceConfig config = invoker.config;
String invokerId = config.getId();
if (!formatComplete) {
synchronized (formatLock) {
lock.lock();
Lo1nt marked this conversation as resolved.
Show resolved Hide resolved
try {
if (!formatComplete) {
formatId(idRule);
formatComplete = true;
}
} finally {
lock.unlock();
}
}

Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alipay.sofa.rpc.common.threadpool;

import com.alipay.sofa.rpc.config.ServerConfig;
import com.alipay.sofa.rpc.ext.Extensible;

import java.util.concurrent.Executor;

/**
*
* @author junyuan
* @version SofaExecutorFactory.java, v 0.1 2023年12月13日 20:02 junyuan Exp $
*/
@Extensible
public interface SofaExecutorFactory {
Executor createExecutor();

Executor createExecutor(String namePrefix);

Executor createExecutor(String namePrefix, ServerConfig serverConfig);
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alipay.sofa.rpc.common.threadpool;

/**
*
* @author junyuan
* @version ThreadPoolConstant.java, v 0.1 2023年12月12日 14:01 junyuan Exp $
*/
public class ThreadPoolConstant {
public static final String DefaultThreadNamePrefix = "SOFA-RPC-DEFAULT";
Lo1nt marked this conversation as resolved.
Show resolved Hide resolved
public static final String BizThreadNamePrefix = "SEV-BOLT-BIZ-";

public static final String ConfigKeyThreadPrefix = "threadNamePrefix";
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alipay.sofa.rpc.common.threadpool.extension;

import com.alipay.sofa.rpc.common.struct.NamedThreadFactory;
import com.alipay.sofa.rpc.common.threadpool.SofaExecutorFactory;
import com.alipay.sofa.rpc.common.threadpool.ThreadPoolConstant;
import com.alipay.sofa.rpc.config.ServerConfig;
import com.alipay.sofa.rpc.ext.Extension;
import com.alipay.sofa.rpc.server.BusinessPool;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
*
* @author junyuan
* @version DefaultThreadPoolFactory.java, v 0.1 2023年12月11日 20:55 junyuan Exp $
*/
@Extension(value = "cached")
public class CachedThreadPoolFactory implements SofaExecutorFactory {
protected ServerConfig defaultConfig = new ServerConfig();

@Override
public Executor createExecutor() {
return createExecutor(ThreadPoolConstant.DefaultThreadNamePrefix);
}

@Override
public Executor createExecutor(String namePrefix) {
return createExecutor(namePrefix, defaultConfig);
}

@Override
public Executor createExecutor(String namePrefix, ServerConfig serverConfig) {
ThreadPoolExecutor executor = BusinessPool.initPool(serverConfig);
executor.setThreadFactory(new NamedThreadFactory(namePrefix, serverConfig.isDaemon()));
return executor;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alipay.sofa.rpc.common.threadpool.extension;

import com.alipay.sofa.rpc.common.threadpool.SofaExecutorFactory;
import com.alipay.sofa.rpc.common.threadpool.ThreadPoolConstant;
import com.alipay.sofa.common.thread.virtual.SofaVirtualThreadFactory;
import com.alipay.sofa.rpc.config.ServerConfig;
import com.alipay.sofa.rpc.ext.Extension;

import java.util.concurrent.Executor;

/**
*
* @author junyuan
* @version VirtualThreadPoolFactory.java, v 0.1 2023年12月12日 11:11 junyuan Exp $
*/
@Extension(value = "virtual")
public class VirtualThreadPoolFactory implements SofaExecutorFactory {

@Override
public Executor createExecutor() {
return SofaVirtualThreadFactory.ofExecutorService(ThreadPoolConstant.DefaultThreadNamePrefix);
}

@Override
public Executor createExecutor(String namePrefix) {
return SofaVirtualThreadFactory.ofExecutorService(namePrefix);
}

@Override
public Executor createExecutor(String namePrefix, ServerConfig serverConfig) {
// virtual thread does not support any configs now
return SofaVirtualThreadFactory.ofExecutorService(namePrefix);
}
}
Original file line number Diff line number Diff line change
@@ -25,6 +25,8 @@

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
* @author bystander
@@ -37,6 +39,11 @@ public class DynamicConfigManagerFactory {
*/
private final static ConcurrentMap<String, DynamicConfigManager> ALL_DYNAMICS = new ConcurrentHashMap<String, DynamicConfigManager>();

/**
* 类锁
*/
private final static Lock classLock = new ReentrantLock();

/**
* slf4j Logger for this class
*/
@@ -49,12 +56,13 @@ public class DynamicConfigManagerFactory {
* @param alias 别名
* @return DynamicManager 实现
*/
public static synchronized DynamicConfigManager getDynamicManager(String appName, String alias) {
public static DynamicConfigManager getDynamicManager(String appName, String alias) {
Lo1nt marked this conversation as resolved.
Show resolved Hide resolved
if (ALL_DYNAMICS.size() > 3) { // 超过3次 是不是配错了?
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("Size of dynamic manager is greater than 3, Please check it!");
}
}
classLock.lock();
try {
// 注意:RegistryConfig重写了equals方法,如果多个RegistryConfig属性一样,则认为是一个对象
DynamicConfigManager registry = ALL_DYNAMICS.get(alias);
@@ -73,6 +81,8 @@ public static synchronized DynamicConfigManager getDynamicManager(String appName
} catch (Throwable e) {
throw new SofaRpcRuntimeException(LogCodes.getLog(LogCodes.ERROR_LOAD_EXT, "DynamicConfigManager", alias),
e);
} finally {
classLock.unlock();
}
}

Original file line number Diff line number Diff line change
@@ -19,6 +19,7 @@
import com.alipay.sofa.rpc.common.struct.NamedThreadFactory;
import com.alipay.sofa.rpc.common.utils.ThreadPoolUtils;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -87,20 +88,31 @@ public UserThreadPool(String uniqueThreadPoolName) {
/**
* 线程池
*/
@Deprecated
transient volatile ThreadPoolExecutor executor;
transient volatile ExecutorService executorService;

/**
* 初始化线程池
*/
public void init() {
executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MILLISECONDS,
executorService = buildExecutorService();
if (executorService instanceof ThreadPoolExecutor) {
executor = (ThreadPoolExecutor) executorService;
}
}

protected ExecutorService buildExecutorService() {
ThreadPoolExecutor executorService = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime,
TimeUnit.MILLISECONDS,
ThreadPoolUtils.buildQueue(queueSize), new NamedThreadFactory(threadPoolName));
if (allowCoreThreadTimeOut) {
executor.allowCoreThreadTimeOut(true);
}
if (prestartAllCoreThreads) {
executor.prestartAllCoreThreads();
}
return executorService;
}

/**
@@ -257,14 +269,20 @@ public UserThreadPool setKeepAliveTime(int keepAliveTime) {
*
* @return the executor
*/
@Deprecated
public ThreadPoolExecutor getExecutor() {
if (executor == null) {
getExecutorService();
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这里是初始化executorService,最终返回的是executor

return executor;
}

public ExecutorService getExecutorService() {
if (executorService == null) {
synchronized (this) {
if (executor == null) {
if (executorService == null) {
init();
}
}
}
return executor;
return executorService;
}
}
Loading