OpenStack4j 多线程编程session问题（一）
背景
项目中使用了 OpenStack4j github链接 来调用 OpenStack 接口，最近负责实现一个“自动化构建上百个虚拟节点”功能，觉得单线程模式下一个个创虚拟机太慢了，考虑使用多线程的方式创建虚拟机，创建过程中遇到了如下异常：
# org.openstack4j.api.exceptions.OS4JException:
# Unable to retrieve current session.
# Please verify thread has a current session available.
从字面上，是说我当前的线程没有对应的会话（session），那么这个会话到底是怎么回事呢？如何用多线程实现对 OpenStack 的并发请求呢？下面就来说一说
错误还原
使用 OpenStack4j 创建虚拟机的代码很简单，分两步走，第一步，建立一个 client，第二步，用 client 获取 compute 服务并创建虚拟机，代码示范如下：
第一步，建立一个 client：
OSClientV3 os = OSFactory.builderV3()
.endpoint("http://<fqdn>:5000/v3")
.credentials("admin", "secret", Identifier.byId("user domain id"))
.scopeToProject(Identifier.byId("project id"))
.authenticate());
第二步，获取 compute服务并创建虚拟机，这里的server就是创建的虚拟机：
ServerCreate serverCreate = Builders.server()
.name(name)
.addSecurityGroup(securityGroup)
.flavor(flavorId)
.image(imageId)
.networks(networks)
.keypairName(keypairName)
.build();
Server server = os.compute().servers().boot(serverCreate);
那么用多线程创建虚拟机的代码结构就大致如下，当然，只是简单的示范代码：
OSClientV3 os = OSFactory.builderV3()
.endpoint("http://<fqdn>:5000/v3")
.credentials("admin", "secret", Identifier.byId("user domain id"))
.scopeToProject(Identifier.byId("project id"))
.authenticate());
//用线程池管理线程
ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
for(int i = 0; i < n; i++) {
executorService.submit(new createMission(os));
}
executorService.shutdown();
//创建一个任务类，继承线程
class CreateMission extends Thread {
OSClientV3 os;
public CreateMission(OSClientV3 os) {
this.os = os;
}
@Override
public void run() {
//创建serverCreate对象，这里省略代码
Server server = os.compute().servers().boot(serverCreate);
}
}
如果这么写的话，就报错啦，遇到上文所述的 session 异常：
# org.openstack4j.api.exceptions.OS4JException:
# Unable to retrieve current session.
# Please verify thread has a current session available.
为了解决这个错误，我们就得知道 session 是什么，为什么在多线程下会出错
Session
下面结合 OpenStack4j 源码来分析一下这个 session
说简单也简单，从 client 的创建说起：
OSClientV3 os = OSFactory.builderV3()
.endpoint("http://<fqdn>:5000/v3")
.credentials("admin", "secret", Identifier.byId("user domain id"))
.scopeToProject(Identifier.byId("project id"))
.authenticate());
如上所示，client 是用 OSFactory 的 builderV3 实现的，返回了一个ClientV3 对象：
package org.openstack4j.openstack;
public abstract class OSFactory<T extends OSFactory<T>> {
……
public static V3 builderV3() {
return new ClientV3();
}
}
在创建 client 的一系列调用中， clientV3 对象最后调用了一个方法 authenticate()：
package org.openstack4j.openstack.client;
public abstract class OSClientBuilder {
……
public static class ClientV3 extends OSClientBuilder {
……
public OSClientV3 authenticate() throws AuthenticationException {
if (this.tokenId != null && this.tokenId.length() > 0) {
return this.scope != null ? (OSClientV3)OSAuthenticator.invoke(new KeystoneAuth(this.tokenId, this.scope), this.endpoint, this.perspective, this.config, this.provider) : (OSClientV3)OSAuthenticator.invoke(new KeystoneAuth(this.tokenId), this.endpoint, this.perspective, this.config, this.provider);
} else {
return this.user != null && this.user.length() > 0 ? (OSClientV3)OSAuthenticator.invoke(new KeystoneAuth(this.user, this.password, this.domain, this.scope), this.endpoint, this.perspective, this.config, this.provider) : (OSClientV3)OSAuthenticator.invoke(new KeystoneAuth(this.scope, Type.TOKENLESS), this.endpoint, this.perspective, this.config, this.provider);
}
}
}
}
要注意 return 语句里调用了 OSAuthenticator.invoke 方法，这里面的实现如下：
package org.openstack4j.openstack.internal;
public class OSAuthenticator {
……
public static OSClient invoke(KeystoneAuth auth, String endpoint, Facing perspective, Config config,
CloudProvider provider) {
SessionInfo info = new SessionInfo(endpoint, perspective, false, provider);
return authenticateV3(auth, info, config);
}
private static OSClientV3 authenticateV3(KeystoneAuth auth, SessionInfo info, Config config) {
……
if (!info.reLinkToExistingSession) {
OSClientSessionV3 v3 = OSClientSessionV3.createSession(token, info.perspective, info.provider, config);
v3.reqId = reqId;
return v3;
}
OSClientSessionV3 current = (OSClientSessionV3) OSClientSessionV3.getCurrent();
current.token = token;
current.reqId = reqId;
return current;
}
}
注意看最后，这个方法返回了一个 OSClientSessionV3 的对象！所以说我们用的 client，其实就可以视为一个 session
以上介绍了那么多看似繁琐的调用，就是为了让我们形成这样一个理解：
在 OpenStack4j 里，创建的 client 可以被理解成创建了一次会话（session），以后的调用，都是基于此次会话的
最后看一下session类：
/**
* A client which has been identified. Any calls spawned from this session will
* automatically utilize the original authentication that was successfully
* validated and authorized
*
* @author Jeremy Unruh
*/
public abstract class OSClientSession {
private static final ThreadLocal<OSClientSession> sessions = new ThreadLocal<OSClientSession>();
……
public static class OSClientSessionV3 extends OSClientSession<OSClientSessionV3, OSClientV3> {
……
}
}
注意看一开始定义的 ThreadLocal<OSClientSession> sessions
谜底揭晓了，OSClientSession 用 ThreadLocal<OSClientSession> 为每个线程维护了一个 session(也就是client)
每次使用client调用某服务的时候，最后会调用到这里：
package org.openstack4j.openstack.internal;
public class BaseOpenStackService {
……
@SuppressWarnings("rawtypes")
private <R> Invocation<R> builder(Class<R> returnType, String path, HttpMethod method) {
OSClientSession ses = OSClientSession.getCurrent();
if (ses == null) {
throw new OS4JException(
"Unable to retrieve current session. Please verify thread has a current session available.");
}
RequestBuilder<R> req = HttpRequest.builder(returnType).endpointTokenProvider(ses).config(ses.getConfig())
.method(method).path(path);
Map headers = ses.getHeaders();
if (headers != null && headers.size() > 0){
return new Invocation<R>(req, serviceType, endpointFunc).headers(headers);
}else{
return new Invocation<R>(req, serviceType, endpointFunc);
}
}
}
上图里，着重看这段代码：
OSClientSession ses = OSClientSession.getCurrent();
if (ses == null) {
throw new OS4JException(
"Unable to retrieve current session. Please verify thread has a current session available.");
}
回顾一下之前多线程创建虚拟机的代码，我们在主线程里创建了 client (也可理解为会话)，在另外的线程里试图用这个会话调用 OpenStack 的服务，但这个会话是用 ThreadLocal 维护的，只属于主线程，另外的线程当然取不到这个会话，于是就报错啦
于是要使用多线程并发调用 OpenStack 的话，需要在每个线程里都创建一个会话，再进行调用，代码框架大概如下：
//用线程池管理线程
ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
for(int i = 0; i < n; i++) {
executorService.submit(new createMission());
}
executorService.shutdown();
//创建一个任务类，继承线程
class CreateMission extends Thread {
@Override
public void run() {
OSClientV3 os = OSFactory.builderV3()
.endpoint("http://<fqdn>:5000/v3")
.credentials("admin", "secret", Identifier.byId("user domain id"))
.scopeToProject(Identifier.byId("project id"))
.authenticate());
//创建serverCreate对象，这里省略代码
Server server = os.compute().servers().boot(serverCreate);
}
}
但是！让我们假设一个场景，如果我要创建500个虚拟机，我用一个线程池，维护了若干个线程。每次，一个线程执行一次创建虚拟机的任务，就要创建一次会话，那么创建500个虚拟机，就需要创建500个会话！
这样对OpenStack服务器的压力是不是太大了？如果要创建1000个，1万个虚拟机呢？
既然我们用了线程池，复用了线程，我们是不是应该想办法把会话也复用一下？
这时候，就想到了利用 ThreadLocal，至于怎么用，下一篇再说了。
