终于测试完成了，也上线了，虽然有些曲折，一期目标基本达成。
多线程
多线程处理用例参数和执行用例场景下，线程池的引入。这个首先解决了多用例运行的耗时太多的问题，其次也解决了每次处理任务新建线程对于性能的消耗。
具体的方案就是新建一个全局的线程池，然后把所有多线程任务包装成一个线程对象，通过将任务丢到线程池中，然后通过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); } } } |
本文内容不用于商业目的，如涉及知识产权问题，请权利人联系51Testing小编(021-64471599-8017)，我们将立即处理
