多线程同步控制使用示例

用户5166330

发布于 2019-04-16 15:08:14

8030

发布于 2019-04-16 15:08:14

背景

在公司做某个项目的时候，所有数据从第三方通过调用接口的方式获取到。但是第三方返回的数据有新老数据之分，所谓新老数据之分，是指某一条数据，你对它进行了修改，数据库会存两条，旧的一条估计是作为历史版本数据。但是，第三方返给我的数据是没有一个字段去标识这个新旧数据的。所以，这就需要去做个过滤。为了简便，在获取到所有数据之后，把新数据单独存表，后面的操作以这个表为主表，就可以避免很多麻烦。可能很多人看完这个第一感觉会把这个问题归于去重。后面我否定了，我认为去重，是指数据存在多条一模一样的，去掉重复的，随便取一条，或者我认为这个比去重稍稍复杂一点。(条条大路通罗马，不一定非得最优解嘛，视图就不说了)

正题

第三方提供的接口很多，为了提高效率，采用多线程的方式去拉去数据。那么问题来了，一次同时跑多少个线程?越多越好吗?答案肯定是否定的。至于具体多少个，网上给出了答案。我个人没有测试，有兴趣你们可以试试。拿来主义之网上说-最好起cpu核心数量x2个线程或者 cpu核心数量x2+2个线程。重新扯回正题：如何实现先把所有数据拉取完了之后，在单独起一个线程去做数据收集？这就需要对线程做一些控制了。网上给出了几种方案，我选择了使用CountDownLatch线程辅助同步类。

示例代码

这是主测试代码，功能就是创建一个固定大小为5的线程池。用线程池去跑11线程。这里面主要是要注意设置需要等待线程的数量。CountDownLatch countDownLatch = new CountDownLatch(10)；这个表示线程记数为10个。

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

public class Mian {

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        List<Runnable> taskList = new ArrayList<Runnable>();
        CountDownLatch countDownLatch = new CountDownLatch(10);//这个是重点
        taskList.add(new Special(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.add(new Common(countDownLatch));
        taskList.forEach(t -> {
            executorService.execute(t);
        });
        executorService.shutdown();
    }
}

下面是普通的线程代码，这个里面需要注意的就是countDownLatch.countDown()这句话，把线程记数值减一。这步操作也是加锁的，因为记数对于所有线程来说都是共享的，多线程操作共享变量，你懂得不加锁会怎样。

package thread;

import java.util.concurrent.CountDownLatch;

public class Common implements Runnable {

    private CountDownLatch countDownLatch;

    public Common(CountDownLatch countDownLatch) {
        this.setCountDownLatch(countDownLatch);
    }

    @Override
    public void run() {
        synchronized (countDownLatch) {
            System.out.println("进入普通线程了");
            countDownLatch.countDown();
            System.out.println("目前的线程记数值为：" + countDownLatch.getCount());
        }
    }

    public CountDownLatch getCountDownLatch() {
        return countDownLatch;
    }

    public void setCountDownLatch(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }

}

下面是特定的线程，它需要等待我前面设置的线程跑完之后在运行。这里面主要的代码就是countDownLatch.await()；这句话的就是去判断线程记数值是否为0。若是0就会继续往下走，否则就会阻塞。

package thread;

import java.util.concurrent.CountDownLatch;

public class Special implements Runnable {

    private CountDownLatch countDownLatch;

    public Special(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        System.out.println("进入特殊线程");
        System.out.println("特殊线程，此时线程记数值为：" + countDownLatch.getCount());
        try {
            countDownLatch.await();//这是重点
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("特殊线程，此时线程记数值为：" + countDownLatch.getCount());
        System.out.println("特殊线程，结束特殊线程");
    }

    public CountDownLatch getCountDownLatch() {
        return countDownLatch;
    }

    public void setCountDownLatch(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }

}