java Java的多线程简单例子

Posted 2021-05-08

package com.sunlearning.msb.concurrent;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class MigrationExcutors {
    private static MigrationExcutors instance = null;
    private static int MAX_THREAD_SIZE = 100;//线程池中数量

    /**
     * 调度任务线程池
     */
    private ExecutorService executorService;

    private MigrationExcutors() {
        executorService = Executors.newFixedThreadPool(MAX_THREAD_SIZE);
    }

    public static MigrationExcutors getInstance() {
        if (instance == null) {
            synchronized (MigrationExcutors.class) {
                if (instance == null)
                    return (instance = new MigrationExcutors());

                return instance;
            }
        }
        return instance;
    }

    /**
     * 提交任务
     *
     * @param callable
     * @return
     */
    public final Future submit(Callable callable) {
        return executorService.submit(callable);
    }

    public final void close() {
        executorService.shutdown();
    }

}

@Service
public class MigrationTask {
    private final static Logger logger = LoggerFactory.getLogger(MigrationTask.class);

    // 多线程迁移报名记录
    public void syncEnrollmentAsync() {
        logger.debug("开始报名记录的迁移--多线程");
        logger.debug("需要迁移报名记录的课程数量为 {}", courseIdMap.size());
        try {
            if (courseIdMap.size() > 0) {
                Semaphore sem = new Semaphore(env.getProperty("thread.max.size", Integer.class, 10));
                CountDownLatch cdl = new CountDownLatch(courseIdMap.size());
                Iterator<Long> it = courseIdMap.keySet().iterator();
                int i = 0;
                while (it.hasNext()) {
                    sem.acquire();
                    i++;
                    long wzCosId = it.next();
                    long courseId = courseIdMap.get(wzCosId);
                    MigrationExcutors.getInstance().submit(new SyncEnrollmentTask(sem, cdl, courseId, wzCosId));
                    logger.debug("启动第{}个线程进行同步报名记录", i);
                }
                cdl.await();
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        logger.debug("结束所有报名记录的迁移--多线程");
    }

    private class SyncEnrollmentTask implements Callable {

        private Semaphore sem;
        private CountDownLatch cdl;
        private long courseId;
        private long wzCosId;

        public SyncEnrollmentTask(Semaphore sem, CountDownLatch cdl, long courseId, long wzCosId) {
            this.sem = sem;
            this.cdl = cdl;
            this.courseId = courseId;
            this.wzCosId = wzCosId;
        }

        @Override
        public Object call() throws Exception {
            try {
                //TODO
            } catch (Exception e) {
                logger.error("迁移课程报名记录出错, courseId: {}, wzCozId: {}", courseId, wzCosId);
                logger.error(e.getMessage(), e);
            } finally {
                sem.release();
                cdl.countDown();
            }
            return null;
        }
    }
}