温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

多线程执行的过程

发布时间:2021-10-11 10:49:56 来源:亿速云 阅读:93 作者:iii 栏目:大数据

这篇文章主要讲解了“多线程执行的过程”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“多线程执行的过程”吧!

1、主线程等待子线程执行完成后再执行——Thread.join()

Vector<Thread> vector = new Vector<>(10);
		for (int i = 0; i < 10; i++) {
			MyThread myThread = new MyThread("id" + i, "name" + i, roles);
			Thread t = new Thread(myThread);
			vector.add(t);
			t.start();
		}
		for (Thread thread : vector) {
			try {
				thread.join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

2、Java多线程--让主线程等待子线程执行完毕(CountDownLatch类)

Runable接口实现类

import com.winning.jcfw.core.util.SpringUtil;
import com.winning.jcfw.empi.dao.DaGrJbxxDao;
import com.winning.jcfw.empi.entity.PatientEntity;
import com.winning.jcfw.empi.service.RegisterEmpiService;
import com.winning.jcfw.empi.service.impl.RegisterEmpiSerImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @ClassName DaGrJbxxEmpiRunnable
 * @Description 公卫表 DA_GR_JBXX 线程执行单条生成EMPI
 * @Author WJX
 * @Date 2019/10/25 17:45
 **/
@Slf4j
public class DaGrJbxxEmpiRunnable implements Runnable{

    private RegisterEmpiService registerEmpiService = SpringUtil.getBean(RegisterEmpiSerImpl.class);

    private DaGrJbxxDao daGrJbxxDao = SpringUtil.getBean(DaGrJbxxDao.class);

    private CountDownLatch dLatch;

    /**
     * 机构代码集合
     */
    private List<String> jgList;

    public DaGrJbxxEmpiRunnable(List<String> threadJgLsit,CountDownLatch downLatch){
        super();
        jgList = threadJgLsit;
        dLatch = downLatch;
    }

    @Override
    public void run() {

        for(int i=0;i<jgList.size();i++){
            String yljgdm = jgList.get(i);
            int dataCount = daGrJbxxDao.getCountByYljgdm(yljgdm);
            log.info("医疗机构代码:"+ yljgdm + "****** 待生成记录:" + dataCount);
            for(int j=0;j<dataCount;j++){
                PatientEntity patientEntity = daGrJbxxDao.selectOne(yljgdm);
                if(null == patientEntity){
                    break;
                }
                String empi = registerEmpiService.buildingEmpi(patientEntity);
                if(StringUtils.isBlank(empi)){
                    //更新公卫表 标识失败
                    daGrJbxxDao.updateFail(patientEntity);
                }else{
                    //更新公卫表 标识成功
                    daGrJbxxDao.updateSuc(patientEntity);
                }
            }
        }
        //当子线程执行完成后,计数器减一;
        dLatch.countDown();
    }
}

调用接口实现类

import com.winning.jcfw.empi.dao.DaGrJbxxDao;
import com.winning.jcfw.empi.service.DaGrJbxxService;
import com.winning.jcfw.empi.thread.DaGrJbxxEmpiRunnable;
import com.winning.jcfw.empi.util.ConvertListUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @ClassName DaGrJbxxSerImpl
 * @Description TODO
 * @Author WJX
 * @Date 2019/10/28 14:53
 **/
@Service
@Slf4j
public class DaGrJbxxSerImpl implements DaGrJbxxService {

    @Autowired
    private DaGrJbxxDao daGrJbxxDao;

    @Override
    public void handleData(int perListCount) {
        int totalCount = daGrJbxxDao.getCount();
        log.info("公卫表 DA_GR_JBXX 表待生成记录:" + totalCount + "条数据");
        //机构集合
        List<String> jgList = daGrJbxxDao.getJgCount();
        //线程机构集合
        List<List<String>> threadjgList = ConvertListUtils.getjgLsit(jgList,perListCount);
        int threadCount = threadjgList.size();
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        //创建计数器对象,此处构造器传入的int类型实参,必须与下文需要的创建的子线程个数相同。
        CountDownLatch latch = new CountDownLatch(threadCount);
        for(int k=0;k<threadCount;k++){
            List<String> yldmList = threadjgList.get(k);
            Runnable daGrJbxxEmpiRunnable = new DaGrJbxxEmpiRunnable(yldmList,latch);
            Thread daGrJbxxEmpiThread = new Thread(daGrJbxxEmpiRunnable);
            executor.execute(daGrJbxxEmpiThread);
        }

        try {
            latch.await();//保证之前的所有的线程都执行完成,才会走下面的
        } catch (InterruptedException e) {
            log.info("线程执行异常",e);
        }
    }
}

CountDownLatch源码解析:

/**
     * Constructs a {@code CountDownLatch} initialized with the given count.
     *
     * @param count the number of times {@link #countDown} must be invoked
     *        before threads can pass through {@link #await}
     * @throws IllegalArgumentException if {@code count} is negative
     */
    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }

await方法

/**
     * Causes the current thread to wait until the latch has counted down to
     * zero, unless the thread is {@linkplain Thread#interrupt interrupted}.
     *
     * <p>If the current count is zero then this method returns immediately.
     *
     * <p>If the current count is greater than zero then the current
     * thread becomes disabled for thread scheduling purposes and lies
     * dormant until one of two things happen:
     * <ul>
     * <li>The count reaches zero due to invocations of the
     * {@link #countDown} method; or
     * <li>Some other thread {@linkplain Thread#interrupt interrupts}
     * the current thread.
     * </ul>
     *
     * <p>If the current thread:
     * <ul>
     * <li>has its interrupted status set on entry to this method; or
     * <li>is {@linkplain Thread#interrupt interrupted} while waiting,
     * </ul>
     * then {@link InterruptedException} is thrown and the current thread's
     * interrupted status is cleared.
     *
     * @throws InterruptedException if the current thread is interrupted
     *         while waiting
     */
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

sync.acquireSharedInterruptibly(1)

/**
     * Acquires in shared mode, aborting if interrupted.  Implemented
     * by first checking interrupt status, then invoking at least once
     * {@link #tryAcquireShared}, returning on success.  Otherwise the
     * thread is queued, possibly repeatedly blocking and unblocking,
     * invoking {@link #tryAcquireShared} until success or the thread
     * is interrupted.
     * @param arg the acquire argument.
     * This value is conveyed to {@link #tryAcquireShared} but is
     * otherwise uninterpreted and can represent anything
     * you like.
     * @throws InterruptedException if the current thread is interrupted
     */
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }


/**
     * Acquires in shared interruptible mode.
     * @param arg the acquire argument
     */
    private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

主要是parkAndCheckInterrupt()是如何将线程阻塞的

await方法使当前线程等待直到count值为0,或者当前线程被打断!如果当前的count值为0,那么await方法直接返回,当前线程不会阻塞!如果当前的count值大于0,那么当前线程阻塞(线程调度机制无法给当前线程分配CPU时间片),直到以下两种情况任意一种发生为止:
count值通过countDown方法的调用达到0 或者 其他线程打断了当前线程

/**
     * Convenience method to park and then check if interrupted
     *
     * @return {@code true} if interrupted
     */
    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

unparkSuccessor()方法唤醒线程

/**
 * Wakes up node's successor, if one exists.
 *
 * @param node the node
 */
private void unparkSuccessor(Node node) {
    /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
     * Thread to unpark is held in successor, which is normally
     * just the next node.  But if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     */
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

最后我们来看一段最简单的使用park与unpark方法阻塞唤醒线程代码:

public static void main(String[] args) {

		Thread t = new Thread(() -> {
			System.out.println("阻塞线程1");
			LockSupport.park();
			System.out.println("线程1执行完啦");
		});

		t.start();

		try {
			Thread.sleep(2000);
			System.out.println("唤醒线程1");
			LockSupport.unpark(t);
			Thread.sleep(5000);
			System.out.println("主线程结束");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

阻塞线程1
唤醒线程1
线程1执行完啦
主线程结束

感谢各位的阅读,以上就是“多线程执行的过程”的内容了,经过本文的学习后,相信大家对多线程执行的过程这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是亿速云,小编将为大家推送更多相关知识点的文章,欢迎关注!

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI