Java BlockingQueue 线程间消息投递与性能测试

江岸

发布时间：2026-10-14| 5人浏览过

原创

在Java开发中，线程间的数据交换常借助java.util.concurrent.BlockingQueue实现。本文将从实际性能测试出发，验证使用LinkedBlockingQueue进行大量数据投递的效率，并剖析核心方法offer和poll的源码逻辑，揭示其通过重入锁和原子变量实现的线程同步机制。

测试结果：BlockingQueue的高吞吐表现

为了验证Java标准类库在处理线程通信时的性能，我们设计了一个简单的生产者-消费者模型测试。测试环境中，主线程作为生产者，不断向队列中插入整数；辅助线程作为消费者，从队列中取出整数进行处理。主要考察在频繁操作下的时间开销。

第1步：准备测试环境与依赖

测试需要引入JDK自带的并发包。首先创建基础的测试类结构：

import java.util.concurrent.*;
// 自定义的运行任务类
import base.MyRunnable;

public class Test {
    public static void main(String[] args) {
        // 创建有界阻塞队列，此处使用默认容量的LinkedBlockingQueue
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer&gt();
        
        // 启动消费线程
        Thread t = new Thread(new MyRunnable(queue));
        t.start();

        // 生产线程循环投递数据
        while(true) {
            try {
                while(true) {
                    for(int i = 0; i < 10000; i++) {
                        // 尝试将元素放入队列，容量满时返回false而非阻塞
                        queue.offer(i);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

第2步：实现消费者端逻辑

消费者端需要持续轮询获取数据，并在处理一定数量后打印耗时统计。以下是MyRunnable的实现：

package base;

import java.lang.Runnable;
import java.util.concurrent.*;
import java.util.Date;

public class MyRunnable implements Runnable {
    private BlockingQueue<Integer> queue;

    public MyRunnable(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    public void run() {
        Date d = new Date();
        long starttime = d.getTime();
        System.err.println("Start: " + starttime);
        int count = 0;
        
        while(true) {
            try {
                // 非阻塞获取队列头元素，为空则立即返回null
                Integer i = this.queue.poll();
                if(i != null) {
                    count++;
                }
                
                // 当处理达到阈值时输出结果
                if(count == 100000) {
                    Date e = new Date();
                    long endtime = e.getTime();
                    System.err.println("Count: " + count);
                    System.err.println("End: " + endtime);
                    System.err.println("Time cost: " + (endtime - starttime) + " ms");
                    break;
                }
            } catch (Exception e) {
                // 异常处理通常应记录日志而非忽略
            }
        }
    }
}

第3步：观察执行结果

在实际测试机上，处理十万条数据的完整流程大约仅需 50毫秒 左右。这表明对于常规的消息传递场景，直接使用BlockingQueue不仅代码简洁，而且性能足以应对大多数高吞吐需求，无需盲目追求更底层的C++级优化方案。

底层原理：offer 与 poll 的实现机制

BlockingQueue的高效性源于其底层的线程同步机制。以最常用的LinkedBlockingQueue为例，其核心在于ReentrantLock（重入锁）、Condition（条件变量）以及AtomicInteger（原子整数）的配合使用。

1. poll 方法的取数逻辑

poll()用于从队列头部获取并移除元素。如果队列为空，它不会阻塞，而是直接返回null。其核心源码逻辑如下：

public E poll() {
    final AtomicInteger count = this.count;
    // 检查当前计数是否为0，若是则直接返回null，无锁操作
    if (count.get() == 0)
        return null;
    
    E x = null;
    int c = -1;
    final ReentrantLock takeLock = this.takeLock;
    
    // 加锁，确保并发安全
    takeLock.lock();
    try {
        if (count.get() > 0) {
            x = extract();      // 从链表中物理移除节点
            c = count.getAndDecrement(); // 原子递减计数器
            
            // 如果递减前计数大于1，唤醒一个正在等待“非空”条件的线程
            if (c > 1)
                notEmpty.signal();
        }
    } finally {
        takeLock.unlock();
    }
    
    // 如果队列原本已满，现在腾出了空间，唤醒等待“非满”的线程
    if (c == capacity)
        signalNotFull();
    return x;
}

关键点：count.get() == 0的判断不在锁内，减少了不必要的锁竞争。只有在队列非空时才获取锁进行操作，提升了空闲或低负载下的性能。

2. offer 方法的入队逻辑

offer(E e)用于将指定元素插入队列头部。如果队列已满，它将返回false而不是阻塞。逻辑与poll对称：

public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    final AtomicInteger count = this.count;
    // 若已满，直接返回false，不进入锁区
    if (count.get() == capacity)
        return false;
    
    int c = -1;
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        if (count.get() < capacity) {
            insert(e);          // 将元素插入链表
            c = count.getAndIncrement(); // 原子递增计数器
            
            // 如果递增后未满，唤醒等待“非满”条件的线程
            if (c + 1 < capacity)
                notFull.signal();
        }
    } finally {
        putLock.unlock();
    }
    
    // 如果队列原本是空的，现在有了元素，唤醒等待“非空”的线程
    if (c == 0)
        signalNotEmpty();
    
    return c >= 0;
}