高并发编程-线程生产者消费者的综合示例

需求

需求：假设有10个线程，最多同时运行5个要求：不使用线程池，使用synchronized-wait¬ifyAll机制

实现

详见注释

代码语言：javascript复制

package com.artisan.test;

import java.time.LocalTime;
import java.util.*;

/**
 * 需求： 假设有10个线程，最多同时运行5个
 * 要求： 不使用线程池，使用synchronized-wait¬ifyAll机制
 */
public class ExerciseDemo {

    // 锁 Monitor
    private static final LinkedList<Control>  CONTROLLIST = new LinkedList();
    // 同时运行的最大线程数
    private static final int MAX_THREADS = 5;

    /**
     * 创建线程
     * @param threadName 线程名称
     * @return
     */
    public static Thread createWorkThread(String threadName){
        return new Thread(() ->{
            // 加锁
            synchronized (CONTROLLIST){
                Optional.of(Thread.currentThread().getName()   " GOT LOCK ,BEGIN..."   LocalTime.now().withNano(0)).ifPresent(System.out::println);
                // 使用while
                // 当集合中运行的线程数量大于5时，wait,放弃锁，不执行
                while (CONTROLLIST.size() >= MAX_THREADS){
                    try {
                        Optional.of(Thread.currentThread().getName()   " WAIT").ifPresent(System.out::println);
                        CONTROLLIST.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }

                // 加入到LinkedList最后
                CONTROLLIST.addLast(new Control());
            }

            //模拟每个线程的业务，假设需要10秒才能结束
            Optional.of(Thread.currentThread().getName()   " working..."   LocalTime.now().withNano(0)).ifPresent(System.out::println);
            try {
                Thread.sleep(10_000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 输出业务完成
            Optional.of(Thread.currentThread().getName()   " END..."   LocalTime.now().withNano(0)).ifPresent(System.out::println);
            // 加锁
            synchronized (CONTROLLIST){
                // 移除最上面的线程
                CONTROLLIST.removeFirst();
                // 唤醒其他所有等待的线程
                CONTROLLIST.notifyAll();
            }

        },threadName);
    }

    /**
     * 主流程
     * @param args
     */
    public static void main(String[] args) {
        List<Thread> workers  = new ArrayList();
        Arrays.asList("T1","T2","T3","T4","T5","T6","T7","T8","T9","T10")
                .stream()
                .map(ExerciseDemo::createWorkThread)
                .forEach(t->{
                    // 启动线程
                    t.start();
                    // 加入到集合列表，待后续一起join
                    workers.add(t);
                });

        // 比那里保存线程的集合，10个线程 join
        workers.stream().forEach(t -> {
                    try {
                        t.join();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
        );

        // 全部完成后，输出结束标识
        Optional.of("DONE").ifPresent(System.out::println);
    }


    /**
     * 没啥实质做用，仅仅是个控制标识
     */
    static class  Control{

    }

}

运行日志

代码语言：javascript复制

"E:Program FilesJavajdk1.8.0_161binjava" "-javaagent:E:Program FilesJetBrainsIntelliJ IDEA 2017.2.4libidea_rt.jar=60076:E:Program FilesJetBrainsIntelliJ IDEA 2017.2.4bin" -Dfile.encoding=UTF-8 -classpath "E:Program FilesJavajdk1.8.0_161jrelibcharsets.jar;E:Program FilesJavajdk1.8.0_161jrelibdeploy.jar;E:Program FilesJavajdk1.8.0_161jrelibextaccess-bridge-64.jar;E:Program FilesJavajdk1.8.0_161jrelibextcldrdata.jar;E:Program FilesJavajdk1.8.0_161jrelibextdnsns.jar;E:Program FilesJavajdk1.8.0_161jrelibextjaccess.jar;E:Program FilesJavajdk1.8.0_161jrelibextjfxrt.jar;E:Program FilesJavajdk1.8.0_161jrelibextlocaledata.jar;E:Program FilesJavajdk1.8.0_161jrelibextnashorn.jar;E:Program FilesJavajdk1.8.0_161jrelibextsunec.jar;E:Program FilesJavajdk1.8.0_161jrelibextsunjce_provider.jar;E:Program FilesJavajdk1.8.0_161jrelibextsunmscapi.jar;E:Program FilesJavajdk1.8.0_161jrelibextsunpkcs11.jar;E:Program FilesJavajdk1.8.0_161jrelibextzipfs.jar;E:Program FilesJavajdk1.8.0_161jrelibjavaws.jar;E:Program FilesJavajdk1.8.0_161jrelibjce.jar;E:Program FilesJavajdk1.8.0_161jrelibjfr.jar;E:Program FilesJavajdk1.8.0_161jrelibjfxswt.jar;E:Program FilesJavajdk1.8.0_161jrelibjsse.jar;E:Program FilesJavajdk1.8.0_161jrelibmanagement-agent.jar;E:Program FilesJavajdk1.8.0_161jrelibplugin.jar;E:Program FilesJavajdk1.8.0_161jrelibresources.jar;E:Program FilesJavajdk1.8.0_161jrelibrt.jar;D:IdeaProjectsmvctargetclasses" com.artisan.test.ExerciseDemo
T1 GOT LOCK ,BEGIN...00:03:41
T10 GOT LOCK ,BEGIN...00:03:41
T9 GOT LOCK ,BEGIN...00:03:41
T8 GOT LOCK ,BEGIN...00:03:41
T1 working...00:03:41
T7 GOT LOCK ,BEGIN...00:03:41
T8 working...00:03:41
T7 working...00:03:41
T10 working...00:03:41
T9 working...00:03:41
T6 GOT LOCK ,BEGIN...00:03:41
T6 WAIT
T5 GOT LOCK ,BEGIN...00:03:41
T5 WAIT
T4 GOT LOCK ,BEGIN...00:03:41
T4 WAIT
T3 GOT LOCK ,BEGIN...00:03:41
T3 WAIT
T2 GOT LOCK ,BEGIN...00:03:41
T2 WAIT
T10 END...00:03:51
T8 END...00:03:51
T2 working...00:03:51
T4 WAIT
T3 working...00:03:51
T5 WAIT
T6 WAIT
T1 END...00:03:51
T6 working...00:03:51
T5 WAIT
T4 WAIT
T9 END...00:03:51
T4 working...00:03:51
T5 WAIT
T7 END...00:03:51
T5 working...00:03:51
T2 END...00:04:01
T3 END...00:04:01
T6 END...00:04:01
T4 END...00:04:01
T5 END...00:04:01
DONE

Process finished with exit code 0

首先主线程中初始化10个线程，分别命名为T1 … T10，先把这10个线程临时存放到集合
遍历集合，分别join . 不能在上一步的地方join , 这样的话就只能一个线程一个线程的执行了（join会阻塞当前线程）
10个线程全部完成后，打印DONE
完成主要部分的编码后，就需要关注thread具体的业务逻辑了：定义一个锁 LinkedList ，当线程获取到锁，就将Control添加到Monitor中，如果大于规定的线程数，则wait
业务部分并行执行，当一个线程完成后，获取锁，从Monitor中移除一个Control，然后notifyAll所有正在等待的线程

符合需求 ,OK

join linkedlist monitor synchronized wait

0 人点赞