Введение в использование BlockingQueue на основе Java

Автор：Eve Cole Время обновления：2024-11-18 14:48:02

В последнее время я веду Java-проект, и в группе мы обсуждаем плюсы и минусы Java! К сожалению, некоторые фанаты Java всегда утверждают, что производительность уже не так плоха, как у C++, а многие стандартные библиотеки написаны мастерами, насколько они стабильны и т. д. Я просто внимательно изучил его, и одна из инструкций, которую мне дали, заключалась в том, что для доставки сообщений между потоками достаточно использовать BlockingQueue, инкапсулированную в java.

Поскольку этого достаточно, давайте напишем код для его тестирования. Просто напишите небольшую программу и проведите небольшое тестирование:

Скопируйте код кода следующим образом:

//Пакет по умолчанию
импортировать java.util.concurrent.*;

импортировать базу.MyRunnable;

тест публичного класса
{
public static void main(String[] args)
{
Очередь BlockingQueue<Integer> = new LinkedBlockingQueue<Integer>();
java.lang.Runnable r = новый MyRunnable (очередь);
Поток t = новый поток (r);
т.старт();

пока (правда)
{
пытаться
{
пока (правда)
{
for(int i =0;i <10000;i++)
{
очередь.предложение (я);
}
}
}
поймать (Исключение e)
{
е.printStackTrace();
}
}
}
}

//Пакеты, которые необходимо добавить
пакетная основа;

импортировать java.lang.Runnable;
импортировать java.util.concurrent.*;
импортировать java.util.*;

публичный класс MyRunnable реализует Runnable
{
public MyRunnable (очередь BlockingQueue<Integer>)
{
this.queue = очередь;
}
публичный недействительный запуск()
{
Дата d = новая дата();
долгое время начала = d.getTime();
System.err.println(время начала);
число интервалов = 0;
пока (правда)
{
пытаться
{
Целое число я = this.queue.poll();
если (я! = ноль)
{
считать++;
}
если (счет == 100000)
{
Дата е = новая дата();
длинное время окончания = e.getTime();
System.err.println(счет);
System.err.println(время окончания);
System.err.print(время окончания - время начала);
перерыв;
}

}
поймать (Исключение e)
{

}
}
}
частная очередь BlockingQueue<Integer>;
}

Для передачи 100 000 фрагментов данных на моей тестовой машине требуется около 50 мс, что неплохо! Просто взгляните на базовую реализацию BlockingQueue.

В приведенном выше тестовом коде я использовал предложение и опрос. Давайте посмотрим на эти две функции реализации. Первая — предложение.

Скопируйте код кода следующим образом:

публичный E-опрос() {
окончательный счетчик AtomicInteger = this.count;
если (count.get() == 0)
вернуть ноль;
Е х = ноль;
интервал с = -1;
окончательный ReentrantLock takeLock = this.takeLock;
takeLock.lock();
пытаться {
если (count.get() > 0) {
х = извлечь();
c = count.getAndDecrement();
если (с > 1)
notEmpty.сигнал();
}
} окончательно {
takeLock.unlock();
}
если (c == емкость)
сигналНеполный();
вернуть х;
}

Он похож на общий поток синхронизации, за исключением того, что добавляется еще один сигнал. Изучая расширенное программирование в среде Unix, я увидел, что для синхронизации между потоками используются условные переменные, и потоки могут синхронизироваться конкурентным способом!
Реализация функции опроса аналогична!

Скопируйте код кода следующим образом:

публичное логическое предложение (E e) {
if (e == null) создать новое исключение NullPointerException();
окончательный счетчик AtomicInteger = this.count;
если (count.get() == емкость)
вернуть ложь;
интервал с = -1;
окончательный ReentrantLock putLock = this.putLock;
putLock.lock();
пытаться {
if (count.get() <емкость) {
вставить(е);
c = count.getAndIncrement();
если (c + 1 < емкость)
notFull.сигнал();
}
} окончательно {
putLock.unlock();
}
если (с == 0)
сигналNotEmpty();
вернуть с >= 0;
}