Java 中线程之间如何进行通信？

发表于 更新于 分类于 阅读次数： 本文字数： 6.5k 阅读时长 ≈ 6 分钟

嗨，你好呀，我是猿java

在 Java中，线程是执行的最小单元，那么线程之间是如何通信的呢？这篇文章我们一起来分析 5种常用的方式。

  1. 使用 wait()、notify() 和 notifyAll()
  2. 使用 BlockingQueue
  3. Exchanger
  4. 使用 Locks 和 Condition
  5. 使用 Semaphore

1. 使用 wait()、notify() 和 notifyAll()

Java的 Object 类提供了 wait()、notify() 和 notifyAll() 方法，这些方法可以用来实现线程之间的通信，这些方法必须在同步块或同步方法中调用。

  • **wait()**：使当前线程进入等待状态，直到其他线程调用 notify() 或 notifyAll()。
  • **notify()**：唤醒在该对象监视器上等待的单个线程。
  • **notifyAll()**：唤醒在该对象监视器上等待的所有线程。

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class SharedResource {
    private int data;
    private boolean hasData = false;

    public synchronized void produce(int value) throws InterruptedException {
        while (hasData) {
            wait();
        }
        this.data = value;
        hasData = true;
        notify();
    }

    public synchronized int consume() throws InterruptedException {
        while (!hasData) {
            wait();
        }
        hasData = false;
        notify();
        return data;
    }
}

public class ProducerConsumerExample {
    public static void main(String[] args) {
        SharedResource resource = new SharedResource();

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    resource.produce(i);
                    System.out.println("Produced: " + i);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    int data = resource.consume();
                    System.out.println("Consumed: " + data);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }
}

2. 使用 BlockingQueue

BlockingQueue 是Java中一个强大的接口，提供了线程安全的队列操作，并且可以在生产者-消费者模式中使用。BlockingQueue 不需要显式地使用同步机制，它内部已经处理好了线程同步问题。

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueExample {
    public static void main(String[] args) {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    queue.put(i);
                    System.out.println("Produced: " + i);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    int data = queue.take();
                    System.out.println("Consumed: " + data);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }
}

3. 使用 Locks 和 Condition

Java提供了 java.util.concurrent.locks 包，其中包含了 Lock 接口和 Condition 接口。Condition 提供了类似于 wait()、notify() 和 notifyAll() 的方法，但它们与 Lock 对象一起使用，提供了更灵活的线程通信机制。

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

class SharedResourceWithLock {
    private int data;
    private boolean hasData = false;
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    public void produce(int value) throws InterruptedException {
        lock.lock();
        try {
            while (hasData) {
                condition.await();
            }
            this.data = value;
            hasData = true;
            condition.signal();
        } finally {
            lock.unlock();
        }
    }

    public int consume() throws InterruptedException {
        lock.lock();
        try {
            while (!hasData) {
                condition.await();
            }
            hasData = false;
            condition.signal();
            return data;
        } finally {
            lock.unlock();
        }
    }
}

public class LockConditionExample {
    public static void main(String[] args) {
        SharedResourceWithLock resource = new SharedResourceWithLock();

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    resource.produce(i);
                    System.out.println("Produced: " + i);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    int data = resource.consume();
                    System.out.println("Consumed: " + data);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }
}

4. 使用 Exchanger

Exchanger 是一个用于线程间交换数据的同步点。两个线程可以在此同步点交换数据，Exchanger 的 exchange() 方法用于在两个线程之间交换数据。

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import java.util.concurrent.Exchanger;

public class ExchangerExample {
    public static void main(String[] args) {
        Exchanger<Integer> exchanger = new Exchanger<>();

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    System.out.println("Produced: " + i);
                    exchanger.exchange(i);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    int data = exchanger.exchange(null);
                    System.out.println("Consumed: " + data);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }
}

5. 使用 Semaphore

Semaphore 是一个计数信号量，通常用于限制对某些资源的访问。它可以用于控制线程访问共享资源的数量，这在某些情况下也可以用作线程间通信的机制。

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import java.util.concurrent.Semaphore;

class SemaphoreSharedResource {
    private int data;
    private Semaphore semaphore = new Semaphore(1);

    public void produce(int value) throws InterruptedException {
        semaphore.acquire();
        try {
            this.data = value;
            System.out.println("Produced: " + value);
        } finally {
            semaphore.release();
        }
    }

    public int consume() throws InterruptedException {
        semaphore.acquire();
        try {
            System.out.println("Consumed: " + data);
            return data;
        } finally {
            semaphore.release();
        }
    }
}

public class SemaphoreExample {
    public static void main(String[] args) {
        SemaphoreSharedResource resource = new SemaphoreSharedResource();

        Thread producer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    resource.produce(i);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    resource.consume();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }
}

结论

本文，我们分析了 Java线程通信的5种常见方式：

  • wait()/notify() 是一种低级别的同步机制，适合需要精细控制的场合；
  • BlockingQueue 和 Exchanger 提供了更高层次的抽象，简化了线程间的数据交换；
  • Locks 和 Condition 提供了更灵活的锁机制，适合复杂的同步场景；
  • Semaphore 则用于控制资源访问。

在实际应用中，需要选择哪种方式取决于具体的应用场景和需求。如何你有好的通信方式，欢迎评论区留言。

交流学习

最后，把猿哥的座右铭送给你：投资自己才是最大的财富。 如果你觉得文章有帮助，请帮忙转发给更多的好友，或关注公众号：猿java，持续输出硬核文章。

drawing