JAVA中如何实现一个阻塞方法

作者: 王小杰at2019 | 来源:发表于2019-01-28 00:42 被阅读32次

[TOC]

需求

线程A等待变量 var1 ,当变量 var1 不为空的时候立即输出 var1 的值并赋值为 null, 赋值的操作则有线程B完成

实现1 使用 sleep

这种实现方式有明显的缺点,如果不使用sleep是消耗大量的系统资源,在使用sleep的时候会造成消息提示不及时,时间越短系统开销就越大,时间越长延迟就越高


/**
 * 等待对方输入消息,并显示
 */
@Slf4j
public class CharExample {

    volatile static String message;

    public static void main(String[] args) {

        ExecutorService executorService = Executors.newCachedThreadPool();
        // 收线程
        executorService.submit(() -> {
            while (true) {
                log.info("接收就绪");
                if (StringUtils.isEmpty(message)) {
                    Thread.sleep(1000);
                    continue;
                }
                log.info("我收到了:{}", message);
                message = "";
            }
        });
        executorService.submit(() -> {
            while (true) {
                Scanner scanner = new Scanner(System.in);
                System.out.println("请输入要发送的消息:");
                message = scanner.next();
            }
        });
        log.info("start");

    }
}

13:10:43.773 [main] INFO cn.wyj.learn.park.CharExample - start
13:10:43.773 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
113:10:44.780 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
2113:10:45.781 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
2
请输入要发送的消息:
21
请输入要发送的消息:
213:10:46.781 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
13:10:46.781 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 我收到了:21
13:10:46.783 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
13:10:47.784 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪

13:10:48.784 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
22
请输入要发送的消息:
2
请输入要发送的消息:
22
请输入要发送的消息:
22
请输入要发送的消息:
213:10:49.784 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
13:10:49.784 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 我收到了:22
13:10:49.784 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
13:10:50.785 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
13:10:51.785 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
13:10:52.786 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪
13:10:53.786 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample - 接收就绪

实现方式2 wait/notify

这种方式实现,不是使用轮训提减少了系统资源的开销,但是代码维护了较大

在Java5里是用wait/notify/notifyAll来同步的。wait/notify机制有个很蛋疼的地方是，比如线程B要用notify通知线程A，那么线程B要确保线程A已经在wait调用上等待了，否则线程A可能永远都在等待。编程的时候就会很蛋疼。另外，是调用notify，还是notifyAll？notify只会唤醒一个线程，如果错误地有两个线程在同一个对象上wait等待，那么又悲剧了。为了安全起见，貌似只能调用notifyAll了。


/**
 * 等待对方输入消息,并显示
 */
@Slf4j
public class CharExample2 {

    volatile static String message;
    static Object lock = new Object();


    public static void main(String[] args) throws InterruptedException {

        ExecutorService executorService = Executors.newCachedThreadPool();
        // 收线程
        executorService.submit(() -> {
            while (true) {
                log.info("接收就绪:");
                synchronized (lock) {
                    try {
                        if (StringUtils.isEmpty(message)) {

                            lock.wait();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    log.info("我收到了:{}", message);
                    message = "";
                    lock.notifyAll();
                }
            }
        });
        executorService.submit(() -> {
            while (true) {
                synchronized (lock) {
                    if (StringUtils.isNotEmpty(message)) {
                        lock.wait();
                    }
                    Scanner scanner = new Scanner(System.in);
                    System.out.println("请输入要发送的消息:");
                    message = scanner.next();
                    lock.notifyAll();
                }
            }
        });
    }
}

日志输出结果

3:23:55.724 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 接收就绪:
请输入要发送的消息:
123
13:23:57.302 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 我收到了:123
13:23:57.304 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 接收就绪:
请输入要发送的消息:
123
13:23:58.069 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 我收到了:123
13:23:58.069 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 接收就绪:
请输入要发送的消息:
123
13:23:58.581 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 我收到了:123
13:23:58.581 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 接收就绪:
请输入要发送的消息:
123
13:23:59.059 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 我收到了:123
13:23:59.059 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 接收就绪:
请输入要发送的消息:
123
13:23:59.621 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 我收到了:123
13:23:59.621 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 接收就绪:
请输入要发送的消息:
123
13:24:00.066 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 我收到了:123
13:24:00.066 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample2 - 接收就绪:
请输入要发送的消息:
123

实现方式3 park/unpark 实现 unsafe.park,unpark natvie 方法

unpark函数为线程提供“许可(permit)”，线程调用park函数则等待“许可”。这个有点像信号量，但是这个“许可”是不能叠加的，“许可”是一次性的。

比如线程B连续调用了三次unpark函数，当线程A调用park函数就使用掉这个“许可”，如果线程A再次调用park，则进入等待状态。

注意，unpark函数可以先于park调用。比如线程B调用unpark函数，给线程A发了一个“许可”，那么当线程A调用park时，它发现已经有“许可”了，那么它会马上再继续运行。

LockSupport.park()和LockSupport.unpark()实现线程的阻塞和唤醒的。LockSupport很类似于二元信号量(只有1个许可证可供使用)

park/unpark 模型真正解耦了线程之间的同步，线程之间不再需要一个Object或者其它变量来存储状态，不再需要关心对方的状态。

不完美的地方还是需要维护一个线程,有就是在使用多个接收线程的时候没有办法处理

@Slf4j
public class CharExample3 {

    volatile static String message;

    static Thread t1;

    public static void main(String[] args) throws InterruptedException {

        ExecutorService executorService = Executors.newCachedThreadPool();
        // 收线程
        executorService.submit(() -> {
            while (true) {
                t1 = Thread.currentThread();
                log.info("接收就绪:");
                if (StringUtils.isEmpty(message)) {
                    LockSupport.park();
                    log.info("我收到了:{}", message);
                    message = "";
                }
            }
        });
        executorService.submit(() -> {
            while (true) {
                Scanner scanner = new Scanner(System.in);
                System.out.println("请输入要发送的消息:");
                message = scanner.next();
                if (StringUtils.isNotEmpty(message)) {
                    LockSupport.unpark(t1);
                }
            }
        });
    }
}

请输入要发送的消息:
13:52:21.400 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample3 - 接收就绪:
123
请输入要发送的消息:
13:52:23.083 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample3 - 我收到了:123
13:52:23.086 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample3 - 接收就绪:
123
请输入要发送的消息:
13:52:24.186 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample3 - 我收到了:123
13:52:24.186 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample3 - 接收就绪:
1123
请输入要发送的消息:
13:52:25.440 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample3 - 我收到了:1123
13:52:25.440 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample3 - 接收就绪:
123
请输入要发送的消息:
13:52:26.199 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample3 - 我收到了:123
13:52:26.199 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample3 - 接收就绪:

实现方式4 semaphore 信号量

基于信号量的实现非常简单 ,并且可以实现多个消费者


/**
 * 等待对方输入消息,并显示
 */
@Slf4j
public class CharExample4 {

    volatile static String message;


    public static void main(String[] args) throws InterruptedException {

        Semaphore semaphore = new Semaphore(0);
        ExecutorService executorService = Executors.newCachedThreadPool();
        // 收线程
        for (int i = 0; i < 3; i++) {
            executorService.submit(() -> {
                while (true) {
                    //获取许可
                    semaphore.acquire();
                    log.info("接收就绪:");
                    if (StringUtils.isEmpty(message)) {
                        LockSupport.park();
                        log.info("我收到了:{}", message);
                        message = "";
                    }
                }
            });
        }
        executorService.submit(() -> {
            while (true) {
                Scanner scanner = new Scanner(System.in);
                System.out.println("请输入要发送的消息:");
                message = scanner.next();
                if (StringUtils.isNotEmpty(message)) {
                    //增加一个许可
                    semaphore.release();
                }
            }
        });
    }
}


请输入要发送的消息:
12
请输入要发送的消息:
13:58:25.492 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample4 - 接收就绪:
12
请输入要发送的消息:
13:58:25.887 [pool-1-thread-2] INFO cn.wyj.learn.park.CharExample4 - 接收就绪:
12
请输入要发送的消息:
13:58:26.208 [pool-1-thread-3] INFO cn.wyj.learn.park.CharExample4 - 接收就绪:
21
请输入要发送的消息:
13:58:26.488 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample4 - 接收就绪:
21
请输入要发送的消息:
13:58:26.718 [pool-1-thread-2] INFO cn.wyj.learn.park.CharExample4 - 接收就绪:
21
请输入要发送的消息:
13:58:26.956 [pool-1-thread-3] INFO cn.wyj.learn.park.CharExample4 - 接收就绪:
21
请输入要发送的消息:
13:58:27.139 [pool-1-thread-1] INFO cn.wyj.learn.park.CharExample4 - 接收就绪:

21
请输入要发送的消息:
13:58:27.512 [pool-1-thread-2] INFO cn.wyj.learn.park.CharExample4 - 接收就绪:
21

JAVA中如何实现一个阻塞方法

需求

实现1 使用 sleep

实现方式2 wait/notify

实现方式3 park/unpark 实现 unsafe.park,unpark natvie 方法

实现方式4 semaphore 信号量

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读