一个多线程例子，通过启动三个线程 等三个线程都跑完后打印结果
需求描述：开启一个多线程跑3条门店的数据，每个门店都新起一个线程去跑数据，当所有线程跑完后主线程打印结果，记录报错的门店编码
1.线程池配置类
import com.alibaba.fastjson.JSONObject;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;/*** 对优惠卷创建设置对线程池默认配置** @author woniu* @date 2024-01-16*/
@Configuration
public class CouponPolicyThreadPoolConfig {/*** CPU核心数(不一定准确，在单核多线程的CPU情况下可能会存在数据的偏差)*/private final static Integer PROCESSINGCOUNT = Runtime.getRuntime().availableProcessors();/*** 该线程池，专门用于优惠政策的数据从持久层加载到进程内存中的工作过程。用于提高加载效率** @return*/@Bean("couponPolicyCreateExecutor")@ConditionalOnMissingBean(name = "couponPolicyCreateExecutor")public ThreadPoolExecutor getCouponPolicyCreateExecutor() {return new ThreadPoolExecutor(PROCESSINGCOUNT, PROCESSINGCOUNT, 1, TimeUnit.HOURS, new LinkedBlockingQueue<>(), new CouponPolicyCreateExecutorThreadFactory());}/*** 获取CouponPolicyCreateTask bean（多例）* @return*/@Bean@Scope("prototype")public CouponPolicyCreateTask getCouponPolicyCreateTask(String tenantCode, JSONObject salePolicyJson) {return new CouponPolicyCreateTask(tenantCode,salePolicyJson);}/*** 自定义实现优惠卷创建的执行线程工厂*/private class CouponPolicyCreateExecutorThreadFactory implements ThreadFactory {private final AtomicInteger count = new AtomicInteger(1);@Overridepublic Thread newThread(Runnable r) {return new Thread(r, "coupon-create-task-" + count.getAndIncrement());}}}
2.任务类
import com.alibaba.fastjson.JSONObject;
import com.woniu.crm.user.User;
import com.woniu.woniuuser.mapper.UserMapper;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;
import java.util.concurrent.Callable;/*** 优惠卷创建任务类** @author woniu* @date 2024-01-16*/
public class CouponPolicyCreateTask implements Callable<Integer> {/*** 日志*/private static final Logger LOGGER = LoggerFactory.getLogger(CouponPolicyCreateTask.class);@Resourceprivate UserMapper userMapper;/*** 租户编码*/private String tenantCode;/*** 促销政策json*/private JSONObject salePolicyJson;public CouponPolicyCreateTask(String tenantCode, JSONObject salePolicyJson) {this.tenantCode = tenantCode;this.salePolicyJson=salePolicyJson;Validate.notNull(salePolicyJson, "CouponPolicyCreateTask: tenantCode blank !!");Validate.notBlank(tenantCode, "CouponPolicyCreateTask: salePolicyJson null !!");}@Override@Transactionalpublic Integer call() throws Exception {try {LOGGER.info("租户编码"+tenantCode);User user=new User();user.setUsername(tenantCode);userMapper.insert(user);salePolicyJson.put("2","3");if ("1".equals(tenantCode)) {Thread.sleep(1000);throw new RuntimeException("1出现报错");}else{Thread.sleep(2000);}} catch (RuntimeException e) {throw new RuntimeException("出现报错编码为:" + tenantCode);}return 1;}}
3.主线程调用类
/*** 测试多线程* @param* @return*/
@Override
@Transactional(rollbackFor = Exception.class)
public void multiThread(){/*** 需求描述开启一个多线程跑3条门店的数据，每个门店都新起一个线程去跑数据，当所有线程跑完后主线程打印结果，记录报错的门店编码*///1.新建三个门店编码List<String> tenantCodes=new ArrayList<>();for(int i=0;i<3;i++){tenantCodes.add(String.valueOf(i));}log.info("所有任务开始");Set<Future<Integer>> loadingSet = Sets.newLinkedHashSet();String json="{\"salePolicyCode\":\"123\"}";JSONObject jsonObject = JSON.parseObject(json);for(String tenant:tenantCodes){String s = jsonObject.toJSONString();//注意这里必须转成字符串然后去转换 或者将对象重新拷贝，不然所有对象中用的都是同一个jsonObjectJSONObject newJsonObject = JSON.parseObject(s);//通过getBean方式获取taskCouponPolicyCreateTask task = this.applicationContext.getBean(CouponPolicyCreateTask.class, tenant, newJsonObject);//submit方式提交任务Future<Integer> submit = couponPolicyCreateExecutor.submit(task);loadingSet.add(submit);}List<String> failTenantCodes = new ArrayList<>();for(Future<Integer> result:loadingSet){try {//当执行到这里时，get方法会进行阻塞主线程，当执行result任务的线程执行完成后get方法才会释放，程序继往下执行result.get();}catch (ExecutionException | InterruptedException e) {log.error("主线程报错");//通过报错信息截取报错的租户编码String message = e.getMessage();int index = message.indexOf("出现报错租户编码为:");String substring = message.substring(index + 10);failTenantCodes.add(substring);continue;}}log.info("所有任务结束");log.info("出现报错租户编码集合"+JSON.toJSONString(failTenantCodes));}
)
