Две причины:

1. Вы не ждете завершения потоков, вы просто закрываете пул потоков (то есть заставляете пул потоков отклонять новые задачи, но продолжать обрабатывать существующие задачи).

2. Вы не устанавливаете связь между операциями записи в пуле потоков и операциями чтения в основном потоке.

   Вы можете сделать это (среди других методов):

   1. Acquiring the semaphore before reading a;
   2. Используя submit вместо execute, чтобы получить Future<?> для каждой из отправленных задач, и вызывая метод Future.get() для всех возвращенных фьючерсов. Он задокументирован в Javadoc от ExecutorService что это устанавливает а бывает - раньше.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class DemoApplicationTests {
    final AtomicInteger a = new AtomicInteger(0);
    final AtomicInteger b = new AtomicInteger(0);

    void contextLoads() throws Exception {
        CountDownLatch latch = new CountDownLatch(10);
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        for (int i = 0; i < 10; i++) {
            executorService.execute(() -> {
                add();
                bdd();
                latch.countDown();
            });
        }
        latch.await();
        executorService.shutdown();
        System.out.println("The final value of a：" + a);
        System.out.println("The final value of b：" + b);
    }

    public void add() {
        a.incrementAndGet();
    }
    public void bdd() {
        b.incrementAndGet();
    }

    public static void main(String[] args) throws Exception {
        new DemoApplicationTests().contextLoads();
    }
}

Следующее может (мои знания ограничены и не могу подтвердить в любом случае) быть совершенно законным кодом для размера пула потоков 1 (скопировано из ответ Юджина)

Но когда тот же код был выполнен с размером пула потоков › 1, это приведет к условиям гонки. (опять же намерение состоит в том, чтобы обсудить проблемы с несколькими потоками и видимостью данных как есть, а не проецировать ответ Юджина как неправильный. Ответ Юджина находится в контексте одного потока в пуле потоков и может быть совершенно правильным для однопоточного сценария пула потоков)

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DemoApplicationTests {
    int a = 0;
    int b = 0;

    void contextLoads() throws Exception {
        final int count = 10000;
        ExecutorService executorService = Executors.newFixedThreadPool(100);
        List<Runnable> list = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            Runnable r = () -> {
                add();
                bdd();
            };
            list.add(r);
        }

        CompletableFuture<?>[] futures = list.stream()
            .map(task -> CompletableFuture.runAsync(task, executorService))
            .toArray(CompletableFuture[]::new);

        CompletableFuture.allOf(futures).join();

        executorService.shutdown();
        System.out.println("The final value of a: " + a);
        System.out.println("The final value of b：" + b);
    }

    public void add() {
        a++;
    }
    public void bdd() {
        b++;
    }

    public static void main(String[] args) throws Exception {
        new DemoApplicationTests().contextLoads();
    }
}

Спасибо @Basil Bourque за исправление грамматических ошибок

В вашем пуле есть 1 поток, и вы отправляете в него 10 Runnables. Все они соберутся в queue, пока не наступит их очередь казнить. Вместо того, чтобы ждать, пока все они выполнят finish, вы вызываете shutDown, фактически говоря: больше никаких задач этот пул выполнять не будет. Когда именно это произойдет и сколько задач уже было обработано до того, как произошел вызов shutDown, сказать невозможно. Таким образом, вы получаете очень недетерминированный результат. Вы можете даже увидеть 10 в качестве вывода (иногда), но это не значит, что это правильно.

Вместо этого вы можете подождать, пока пул завершит выполнение всех своих задач:

executorService.awaitTermination(2, TimeUnit.SECONDS);
executorService.shutdown();

Что немного отстойно, так это то, что awaitTermination явно не упоминает, что если он вернет true, он установит отношение happens-before. Таким образом, чтобы быть педантичным с JLS, вам нужно будет работать, например, с этим Semaphore, чтобы установить необходимые гарантии.

У вас есть гонка в вашем коде путем обновления общих a и b из нескольких потоков (даже если вы в настоящее время используете Executors.newFixedThreadPool(1)) без какой-либо синхронизации. Так что это тоже нуждается в исправлении. И Semaphore semaphore = new Semaphore(3); не поможет, поскольку вы все равно позволите 3 параллельным потокам работать с этими переменными; вам понадобится только один permit. Но тогда это действует как Lock, а не как Semaphore.

Другие ответы правильные, с важными моментами. Кроме того, позвольте мне показать, как будущие технологии, разрабатываемые в Project Loom, упростят такой код.

Проект Ткацкий станок

Project Loom внесет некоторые изменения в Java. Экспериментальные сборки технологии Loom, основанные на ранней версии Java 17, уже доступно. Команда Loom запрашивает отзывы.

AutoCloseable и попробуйте с ресурсами

Одно изменение заключается в том, что ExecutorService расширяет AutoCloseable. Это означает, что мы можем использовать синтаксис try-with-resources для удобно и автоматически закрывать сервис после завершения блока try.

Блокировать, пока отправленные задачи не будут выполнены

Кроме того, поток управления блокируется в этом блоке попыток до тех пор, пока все отправленные задачи не будут выполнены/не пройдены/отменены. Нет необходимости отслеживать ход выполнения отдельных задач, если вам это не нужно.

try (
        ExecutorService executorService = Executors.newVirtualThreadExecutor() ;
)
{
    … submit tasks to the executor service, to be run on background threads.
}
// At this point, all submitted are done/failed/canceled. 
// At this point, the executor service is automatically being shut down.

AtomicInteger

Как говорили другие, использование вами примитивов int для переменных a и b в потоках может завершиться неудачей из-за проблем с видимостью в Модель памяти Java. Один из вариантов — пометить их как volatile.

Я предпочитаю альтернативу, используя классы Atomic…. Замените эти переменные int на AtomicInteger для переноса увеличивающегося числа счетчиков.

Отметьте эти поля-члены final, чтобы каждый экземпляр никогда не заменялся.

// Member fields
final AtomicInteger a, b;

// Constructor
public Incrementor ( )
{
    this.a = new AtomicInteger();
    this.b = new AtomicInteger();
}

Чтобы увеличить на единицу значение в AtomicInteger, мы вызываем incrementAndGet. Этот вызов возвращает новое увеличенное число. Поэтому я изменил сигнатуру ваших методов добавления, чтобы показать, что мы можем вернуть новое значение, если это когда-либо понадобится.

// Logic
public int addA ( )
{
    return this.a.incrementAndGet();
}

public int addB ( )
{
    return this.b.incrementAndGet();
}

Виртуальные потоки

Еще одна функция, появившаяся в Project Loom, — это виртуальные потоки, также известные как волокна. Многие из этих облегченных потоков отображаются для работы в потоках платформы/ядра. Если ваш код часто блокируется, то использование виртуальных потоков значительно повысит производительность вашего приложения. Используйте новую функцию, вызвав Executors.newVirtualThreadExecutor.

try (
        ExecutorService executorService = Executors.newVirtualThreadExecutor() ;
)
{ … }

Пример класса

Я написал класс Incrementor, похожий на ваш. Использование такого класса выглядит так:

        Incrementor incrementor = new Incrementor();
        try (
                ExecutorService executorService = Executors.newVirtualThreadExecutor() ;
        )
        {
            for ( int i = 0 ; i < 10 ; i++ )
            {
                executorService.submit( ( ) -> {
                    int newValueA = incrementor.addA();
                    int newValueB = incrementor.addB();
                    System.out.println( "Thread " + Thread.currentThread().getId() + " incremented a & b to: " + newValueA + " & " + newValueB + " at " + Instant.now() );
                } );
            }
        }

Координация a и b

Предостережение: как уже отмечали другие, такой код не увеличивает атомарно a и b вместе синхронно. Судя по всему, вам это не нужно, поэтому я игнорирую этот вопрос. Вы можете увидеть это поведение в действии в примере вывода запуска, показанном внизу ниже, где два потока чередуются во время их доступа к a и b. Выдержка здесь:

Thread 24 incremented a & b to: 10 & 9 at 2021-02-09T02:21:30.270246Z
Thread 23 incremented a & b to: 9 & 10 at 2021-02-09T02:21:30.270246Z

Полный код класса

Соберите весь этот код.

Обратите внимание на простоту кода, когда (а) используется Loom и (б) используются константы Atomic…. Нет необходимости ни в семафорах, ни в защелках, ни в CompletableFuture, ни в вызове ExecutorService#shutdown.

package work.basil.example;

import java.time.Instant;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class Incrementor
{
    // Member fields
    final AtomicInteger a , b ;

    // Constructor
    public Incrementor ( )
    {
        this.a = new AtomicInteger();
        this.b = new AtomicInteger();
    }

    // Logic
    public int addA ( )
    {
        return this.a.incrementAndGet();
    }

    public int addB ( )
    {
        return this.b.incrementAndGet();
    }
}

И метод main для демонстрации использования этого класса.

    public static void main ( String[] args )
    {
        // Exercise this class by instantiating, then incrementing ten times.
        System.out.println( "INFO - `main` starting the demo. " + Instant.now() );

        Incrementor incrementor = new Incrementor();
        try (
                ExecutorService executorService = Executors.newVirtualThreadExecutor() ;
        )
        {
            for ( int i = 0 ; i < 10 ; i++ )
            {
                executorService.submit( ( ) -> {
                    int newValueA = incrementor.addA();
                    int newValueB = incrementor.addB();
                    System.out.println( "Thread " + Thread.currentThread().getId() + " incremented a & b to: " + newValueA + " & " + newValueB + " at " + Instant.now() );
                } );
            }
        }

        System.out.println( "INFO - At this point all submitted tasks are done/failed/canceled, and executor service is shutting down. " + Instant.now() );
        System.out.println( "incrementor.a.get() = " + incrementor.a.get() );
        System.out.println( "incrementor.b.get() = " + incrementor.b.get() );
        System.out.println( "INFO - `main` ending. " + Instant.now() );
    }

Когда бег.

INFO - `main` starting the demo. 2021-02-09T02:21:30.173816Z
Thread 18 incremented a & b to: 4 & 4 at 2021-02-09T02:21:30.245812Z
Thread 14 incremented a & b to: 1 & 1 at 2021-02-09T02:21:30.242306Z
Thread 20 incremented a & b to: 6 & 6 at 2021-02-09T02:21:30.246784Z
Thread 21 incremented a & b to: 8 & 8 at 2021-02-09T02:21:30.269666Z
Thread 22 incremented a & b to: 7 & 7 at 2021-02-09T02:21:30.269666Z
Thread 17 incremented a & b to: 3 & 3 at 2021-02-09T02:21:30.243580Z
Thread 24 incremented a & b to: 10 & 9 at 2021-02-09T02:21:30.270246Z
Thread 23 incremented a & b to: 9 & 10 at 2021-02-09T02:21:30.270246Z
Thread 16 incremented a & b to: 2 & 2 at 2021-02-09T02:21:30.242335Z
Thread 19 incremented a & b to: 5 & 5 at 2021-02-09T02:21:30.246646Z
INFO - At this point all submitted tasks are done/failed/canceled, and executor service is shutting down. 2021-02-09T02:21:30.279542Z
incrementor.a.get() = 10
incrementor.b.get() = 10
INFO - `main` ending. 2021-02-09T02:21:30.285862Z