项目地址：https://github.com/JunManYuanLong/fun-svrBUG`都在 [测开笔记](https://mp.weixin.qq.com/mp/appmsgalbum?action=getalbum&album_id=1384854258558025729&__biz=MzU4MTE2NDEyMQ==#wechat_redirect) 里面了，有兴趣可以一读。，我觉得出去测试框架部分的内容以外，有两个地方值得借鉴。开发过程中遇到的问题和写过的`
号外：这个仓库里面都是一些开源测试框架和测试平台，大家有 GitHub 账号的请不要吝啬星星。
多线程处理用例参数和执行用例场景下，线程池的引入。这个首先解决了多用例运行的耗时太多的问题，其次也解决了每次处理任务新建线程对于性能的消耗。
具体的方案就是新建一个全局的线程池，然后把所有多线程任务包装成一个线程对象，通过将任务丢到线程池中，然后通过CountDownLatch这个类实现等待执行结束，然后进行下一步操作。具体可参考：- CountDownLatch 类在性能测试中应用。
核心代码如下：
package com.okay.family.common.threadpool;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 自定义线程池,用例批量运行用例,非并发测试线程池
*/
public class OkayThreadPool {
private static ThreadPoolExecutor executor = createPool();
public static void addSyncWork(Runnable runnable) {
executor.execute(runnable);
}
private static ThreadPoolExecutor createPool() {
return new ThreadPoolExecutor(16, 100, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1000));
}
}
package com.okay.family.common.threadpool;
import com.okay.family.common.basedata.OkayConstant;
import com.okay.family.common.bean.testcase.CaseRunRecord;
import com.okay.family.common.bean.testcase.request.CaseDataBean;
import com.okay.family.common.enums.CaseAvailableStatus;
import com.okay.family.common.enums.RunResult;
import com.okay.family.utils.RunCaseUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
public class CaseRunThread implements Runnable {
private static Logger logger = LoggerFactory.getLogger(CaseRunThread.class);
int envId;
CaseDataBean bean;
CaseRunRecord record;
CountDownLatch countDownLatch;
public CaseRunRecord getRecord() {
return record;
}
private CaseRunThread() {
}
public CaseRunThread(CaseDataBean bean, CountDownLatch countDownLatch, int runId, int envId) {
this.bean = bean;
this.envId = envId;
this.countDownLatch = countDownLatch;
this.record = new CaseRunRecord();
record.setRunId(runId);
record.setUid(bean.getUid());
record.setParams(bean.getParams());
record.setCaseId(bean.getId());
record.setMark(OkayConstant.RUN_MARK.getAndIncrement());
bean.getHeaders().put(OkayConstant.MARK_HEADER, record.getMark());
record.setHeaders(bean.getHeaders());
}
@Override
public void run() {
try {
if (bean.getAvailable() == RunResult.USER_ERROR.getCode()) {
record.fail(RunResult.USER_ERROR, bean);
} else if (bean.getEnvId() != envId || bean.getAvailable() != CaseAvailableStatus.OK.getCode()) {
record.fail(RunResult.UNRUN, bean);
} else {
RunCaseUtil.run(bean, record);
}
} catch (Exception e) {
logger.warn("用例运行出错,ID:" + bean.getId(), e);
record.fail(RunResult.UNRUN, bean);
} finally {
countDownLatch.countDown();
}
}
}
其中包括线程同步锁和分布式锁。之所以采用两个，主要是因为竞争中拿不到锁的时候，不会像业务开发那样直接丢出来拿锁失败的业务，而是需要等待其他线程安全对用户的验证之后，再取出最新的用户凭证。这里面涉及到的东西比较复杂，中间因为逻辑问题我也写了好几个 BUG。
这里涉及的一些多线程编程的内容，还有在多用例执行的过程中我用到ConcurrentHashMap作为缓存，第一是为了减少对数据库的读写。第二是为了防止用例中大量引用错误的用户导致执行时间变长。
核心代码如下：
/**
* 获取用户登录凭据,map缓存
*
* @param id
* @param map
* @return
*/
@Override
@Transactional(isolation = Isolation.REPEATABLE_READ)
public String getCertificate(int id, ConcurrentHashMap<Integer, String> map) {
if (map.containsKey(id)) return map.get(id);
Object o = UserLock.get(id);
synchronized (o) {
if (map.containsKey(id)) return map.get(id);
logger.warn("非缓存读取用户数据{}", id);
TestUserCheckBean user = testUserMapper.findUser(id);
if (user == null) UserStatusException.fail("用户不存在,ID:" + id);
String create_time = user.getCreate_time();
long create = Time.getTimestamp(create_time);
long now = Time.getTimeStamp();
if (now - create < OkayConstant.CERTIFICATE_TIMEOUT && user.getStatus() == UserState.OK.getCode()) {
map.put(id, user.getCertificate());
return user.getCertificate();
}
boolean b = UserUtil.checkUserLoginStatus(user);
logger.info("环境:{},用户:{},身份:{},登录状态验证:{}", user.getEnvId(), user.getId(), user.getRoleId(), b);
if (!b) {
updateUserStatus(user);
if (user.getStatus() != UserState.OK.getCode()) {
map.put(id, OkayConstant.EMPTY);
UserStatusException.fail("用户不可用,ID:" + id);
}
} else {
testUserMapper.updateUserStatus(user);
}
map.put(id, user.getCertificate());
return user.getCertificate();
}
}
/**
* 更新用户登录状态,全局锁+分布式锁
*
* @param bean
* @return
*/
@Override
@Transactional(isolation = Isolation.REPEATABLE_READ)
public int updateUserStatus(TestUserCheckBean bean) {
int userLock = NodeLock.getUserLock(bean.getId());
int lock = commonService.lock(userLock);
if (lock == 0) {
logger.info("分布式锁竞争失败,ID:{}", bean.getId());
int i = 0;
while (true) {
SourceCode.sleep(OkayConstant.WAIT_INTERVAL);
TestUserCheckBean user = testUserMapper.findUser(bean.getId());
String create_time = user.getCreate_time();
long create = Time.getTimestamp(create_time);
long now = Time.getTimeStamp();
if (now - create < OkayConstant.CERTIFICATE_TIMEOUT && user.getStatus() == UserState.OK.getCode()) {
bean.copyFrom(user);
return testUserMapper.updateUserStatus(bean);
}
if (i++ > OkayConstant.WAIT_MAX_TIME) {
UserStatusException.fail("获取分布式锁超时,无法更新用户凭据:id:" + bean.getId());
}
}
} else {
logger.info("分布式锁竞争成功,ID:{}", bean.getId());
try {
TestUserCheckBean user = testUserMapper.findUser(bean.getId());
String create_time = user.getCreate_time();
long create = Time.getTimestamp(create_time);
long now = Time.getTimeStamp();
if (bean.same(user) && StringUtils.isNotBlank(user.getCertificate())) {
if (now - create < OkayConstant.CERTIFICATE_TIMEOUT && user.getStatus() == UserState.OK.getCode()) {
bean.copyFrom(user);
return testUserMapper.updateUserStatus(bean);
}
if (UserUtil.checkUserLoginStatus(user)) bean.copyFrom(user);
}
UserUtil.updateUserStatus(bean);
return testUserMapper.updateUserStatus(bean);
} catch (Exception e) {
logger.error("用户验证失败!ID:{}", bean.getId(), e);
bean.setStatus(UserState.CANNOT.getCode());
return testUserMapper.updateUserStatus(bean);
} finally {
commonService.unlock(userLock);
}
}
}