由于WS客户端服务和端口的初始化需要很长时间,因此我喜欢在启动时初始化它们并重用相同的端口实例。初始化看起来像这样:
private static RequestContext requestContext = null;
static
{
MyService service = new MyService();
MyPort myPort = service.getMyServicePort();
Map<String, Object> requestContextMap = ((BindingProvider) myPort).getRequestContext();
requestContextMap = ((BindingProvider)myPort).getRequestContext();
requestContextMap.put(BindingProvider.USERNAME_PROPERTY, uName);
requestContextMap.put(BindingProvider.PASSWORD_PROPERTY, pWord);
rc = new RequestContext();
rc.setApplication("test");
rc.setUserId("test");
}
我班上某个地方的电话:
myPort.someFunctionCall(requestContext, "someValue");
我的问题:这个调用是否是线程安全的?
根据CXF FAQ:
JAX-WS客户端代理是否安全?
官方JAX-WS回答:不可以。根据JAX-WS规范,客户端代理不是线程安全的。要编写可移植代码,您应将它们视为非线程安全并同步访问或使用实例池或类似代码。
CXF回答:对于许多用例,CXF代理是线程安全的。例外情况是:
- 使用
((BindingProvider)proxy).getRequestContext()
- 根据JAX-WS规范,请求上下文是PER INSTANCE。因此,那里设置的任何内容都会影响其他线程的请求。使用CXF,您可以:((BindingProvider)proxy).getRequestContext().put("thread.local.request.context","true");
以及对getRequestContext()的调用将使用线程本地请求上下文。这允许请求上下文是线程安全的。 (注意:响应上下文始终是CXF中的线程本地)- 管道上的设置 - 如果使用代码或配置直接操作管道(如设置TLS设置或类似设置),则这些设置不是线程安全的。管道是每个实例,因此将共享这些设置。此外,如果使用FailoverFeature和LoadBalanceFeatures,则会立即替换管道。因此,在设置线程上使用之前,管道上设置的设置可能会丢失。
- 会话支持 - 如果启用会话支持(请参阅jaxws规范),会话cookie将存储在管道中。因此,它将落入管道设置的上述规则,因此可以跨线程共享。
- WS-Security令牌 - 如果使用WS-SecureConversation或WS-Trust,则检索到的令牌将缓存在端点/代理中,以避免对STS进行额外(且昂贵)调用以获取令牌。因此,多个线程将共享令牌。如果每个线程具有不同的安全凭据或要求,则需要使用单独的代理实例。
对于管道问题,您可以安装一个使用本地或类似线程的新ConduitSelector。虽然这有点复杂。
对于大多数“简单”用例,您可以在多个线程上使用CXF代理。以上概述了其他人的解决方法。
一般来说,没有。
根据CXF FAQ http://cxf.apache.org/faq.html#FAQ-AreJAX-WSclientproxiesthreadsafe?
官方JAX-WS回答:不可以。根据JAX-WS规范,客户端代理不是线程安全的。要编写可移植代码,您应将它们视为非线程安全并同步访问或使用实例池或类似代码。
CXF回答:对于许多用例,CXF代理是线程安全的。
有关例外列表,请参阅常见问题解答。
正如您从上面的答案中看到的那样,JAX-WS客户端代理不是线程安全的,因此我只想分享我的实现,其他人将缓存客户端代理。我实际上遇到了同样的问题,并决定创建一个执行JAX-WS客户端代理缓存的spring bean。你可以看到更多细节http://programtalk.com/java/using-spring-and-scheduler-to-store/
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.logging.log4j.Logger;
import org.springframework.stereotype.Component;
/**
* This keeps the cache of MAX_CUNCURRENT_THREADS number of
* appConnections and tries to shares them equally amongst the threads. All the
* connections are created right at the start and if an error occurs then the
* cache is created again.
*
*/
/*
*
* Are JAX-WS client proxies thread safe? <br/> According to the JAX-WS spec,
* the client proxies are NOT thread safe. To write portable code, you should
* treat them as non-thread safe and synchronize access or use a pool of
* instances or similar.
*
*/
@Component
public class AppConnectionCache {
private static final Logger logger = org.apache.logging.log4j.LogManager.getLogger(AppConnectionCache.class);
private final Map<Integer, MyService> connectionCache = new ConcurrentHashMap<Integer, MyService>();
private int cachedConnectionId = 1;
private static final int MAX_CUNCURRENT_THREADS = 20;
private ScheduledExecutorService scheduler;
private boolean forceRecaching = true; // first time cache
@PostConstruct
public void init() {
logger.info("starting appConnectionCache");
logger.info("start caching connections"); ;;
BasicThreadFactory factory = new BasicThreadFactory.Builder()
.namingPattern("appconnectioncache-scheduler-thread-%d").build();
scheduler = Executors.newScheduledThreadPool(1, factory);
scheduler.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
initializeCache();
}
}, 0, 10, TimeUnit.MINUTES);
}
public void destroy() {
scheduler.shutdownNow();
}
private void initializeCache() {
if (!forceRecaching) {
return;
}
try {
loadCache();
forceRecaching = false; // this flag is used for initializing
logger.info("connections creation finished successfully!");
} catch (MyAppException e) {
logger.error("error while initializing the cache");
}
}
private void loadCache() throws MyAppException {
logger.info("create and cache appservice connections");
for (int i = 0; i < MAX_CUNCURRENT_THREADS; i++) {
tryConnect(i, true);
}
}
public MyPort getMyPort() throws MyAppException {
if (cachedConnectionId++ == MAX_CUNCURRENT_THREADS) {
cachedConnectionId = 1;
}
return tryConnect(cachedConnectionId, forceRecaching);
}
private MyPort tryConnect(int threadNum, boolean forceConnect) throws MyAppException {
boolean connect = true;
int tryNum = 0;
MyPort app = null;
while (connect && !Thread.currentThread().isInterrupted()) {
try {
app = doConnect(threadNum, forceConnect);
connect = false;
} catch (Exception e) {
tryNum = tryReconnect(tryNum, e);
}
}
return app;
}
private int tryReconnect(int tryNum, Exception e) throws MyAppException {
logger.warn(Thread.currentThread().getName() + " appservice service not available! : " + e);
// try 10 times, if
if (tryNum++ < 10) {
try {
logger.warn(Thread.currentThread().getName() + " wait 1 second");
Thread.sleep(1000);
} catch (InterruptedException f) {
// restore interrupt
Thread.currentThread().interrupt();
}
} else {
logger.warn(" appservice could not connect, number of times tried: " + (tryNum - 1));
this.forceRecaching = true;
throw new MyAppException(e);
}
logger.info(" try reconnect number: " + tryNum);
return tryNum;
}
private MyPort doConnect(int threadNum, boolean forceConnect) throws InterruptedException {
MyService service = connectionCache.get(threadNum);
if (service == null || forceConnect) {
logger.info("app service connects : " + (threadNum + 1) );
service = new MyService();
connectionCache.put(threadNum, service);
logger.info("connect done for " + (threadNum + 1));
}
return service.getAppPort();
}
}
对此的一般解决方案是在池中使用多个客户端对象,然后使用充当外观的代理。
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
class ServiceObjectPool<T> extends GenericObjectPool<T> {
public ServiceObjectPool(java.util.function.Supplier<T> factory) {
super(new BasePooledObjectFactory<T>() {
@Override
public T create() throws Exception {
return factory.get();
}
@Override
public PooledObject<T> wrap(T obj) {
return new DefaultPooledObject<>(obj);
}
});
}
public static class PooledServiceProxy<T> implements InvocationHandler {
private ServiceObjectPool<T> pool;
public PooledServiceProxy(ServiceObjectPool<T> pool) {
this.pool = pool;
}
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
T t = null;
try {
t = this.pool.borrowObject();
return method.invoke(t, args);
} finally {
if (t != null)
this.pool.returnObject(t);
}
}
}
@SuppressWarnings("unchecked")
public T getProxy(Class<? super T> interfaceType) {
PooledServiceProxy<T> handler = new PooledServiceProxy<>(this);
return (T) Proxy.newProxyInstance(interfaceType.getClassLoader(),
new Class<?>[]{interfaceType}, handler);
}
}
要使用代理:
ServiceObjectPool<SomeNonThreadSafeService> servicePool = new ServiceObjectPool<>(createSomeNonThreadSafeService);
nowSafeService = servicePool .getProxy(SomeNonThreadSafeService.class);