Запретить потокам запускать одну и ту же задачу

Question

Я хочу запретить потокам Android запускать одну и ту же задачу в разных потоках одновременно . Я создал простое приложение Android, которое читает уведомления с использованием класса NotificationsListenerService и отвечает на полученные уведомления из указанного пакета c. Я использую ThreadPoolExecutor для создания нескольких потоков для задач (уведомлений). Все работает, однако одно и то же сообщение или уведомление помещаются в несколько потоков и выполняются одновременно, поэтому мое приложение отправляет дубликаты ответов пользователю, я понимаю, что это из-за состояния гонки, однако мне не удалось найти решение для этого вопроса. Я включил несколько фрагментов кода, пожалуйста, сообщите мне, где проблема.

NotificationsListenerService. java

public void onNotificationPosted(final StatusBarNotification sbn) {
    if (sbn != null && !sbn.isOngoing() && sbn.getPackageName().equals("com.whatsapp.w4b")) {
        notification = sbn.getNotification();
        if (notification != null) {
            bundle = notification.extras;
            remoteInputs = getRemoteInputs(notification);
            if (remoteInputs != null && remoteInputs.size() > 0) {
                Task task = new Task(getApplicationContext(), notification, bundle, remoteInputs, sbn);
                ThreadManager.getManagerInstance().runTask(task);

            }
        }


    }
}

ThreadManager. java

public class ThreadManager {


    private static final int NUMBER_OF_CORES = Runtime.getRuntime().availableProcessors();
    private static final int CORE_POOL_SIZE = NUMBER_OF_CORES * 2; 
    private static final int MAX_POOL_SIZE = NUMBER_OF_CORES * 2; 
    private static final int KEEP_ALIVE_TIME = 60;

    private static ThreadManager managerInstance = null;

    final BlockingQueue<Runnable> WorkQueue;
    private final ThreadPoolExecutor threadPoolExecutor;
    RejectedMessagesHandler rejectedHandles;

    static {
        managerInstance = new ThreadManager();
    }

    private ThreadManager() {
        WorkQueue = new LinkedBlockingQueue<Runnable>();
        rejectedHandles = new RejectedMessagesHandler();
        threadPoolExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME,
                TimeUnit.MILLISECONDS, WorkQueue);
    }

    public void runTask(Runnable runnable) {
        threadPoolExecutor.execute(runnable);
    }

    public static ThreadManager getManagerInstance() {
        return managerInstance;
    }

}

Задание. java

public class Task implements Runnable {

    private static final String TAG = "Task is Running";

    private Notification notification;
    private Bundle bundle;
    private ArrayList<RemoteInput> remoteInputs;
    final StatusBarNotification sbn;

    Context whatsAppNotification;

    public Task(Context whatsAppNotification, Notification notification, Bundle bundle, ArrayList<RemoteInput> remoteInputs, StatusBarNotification sbn) {
        this.notification = notification;
        this.bundle = bundle;
        this.remoteInputs = remoteInputs;
        this.sbn = sbn;
        this.whatsAppNotification = whatsAppNotification;
    }

    public static String generateRandomStringByUUIDNoDash() {
        return UUID.randomUUID().toString().replace("-", "");
    }

    /*
      Task to be done in background.
       */
    @RequiresApi(api = Build.VERSION_CODES.KITKAT)
    @Override
    public void run() {
        try {
            if (notification != null) {
                String ChatID = generateRandomStringByUUIDNoDash();
                    Object Conversation = bundle.get("android.isGroupConversation");
                    if (Conversation != null) {
                        Object title = bundle.get("android.title");//Phone number
                        Object text = bundle.get("android.text");//Message
                        if (title != null && text != null) {
                            String res = "Hi There :)";
                            Log.e(TAG, "Reply Message for Phone: " + title + "\nMessage :" + text + "\nResponse :" + res);
                            sendMsg(res);
                        }
                    }
                Log.e(logMsg, "Thread End");
            }
            Thread.sleep(500);
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    /*
     * get Remote Input of Notification
     */
    private ArrayList<RemoteInput> getRemoteInputs(Notification notification) {
        ArrayList<RemoteInput> remoteInputs = new ArrayList<>();
        NotificationCompat.WearableExtender wearableExtender = new NotificationCompat.WearableExtender(notification);
        for (NotificationCompat.Action act : wearableExtender.getActions()) {
            if (act != null && act.getRemoteInputs() != null) {
                remoteInputs.addAll(Arrays.asList(act.getRemoteInputs()));
            }
        }
        return remoteInputs;
    }

    /*
     * Send/Reply through Notification
     */
    @RequiresApi(api = Build.VERSION_CODES.KITKAT)
    private void sendMsg(String msg) {
        String LOG = "WAR-Log";
        RemoteInput[] allremoteInputs = new RemoteInput[remoteInputs.size()];
        Intent localIntent = new Intent();
        localIntent.addFlags(Intent.FLAG_ACTIVITY_NEW_TASK);
        Iterator it = remoteInputs.iterator();
        int i = 0;
        while (it.hasNext()) {
            allremoteInputs[i] = (RemoteInput) it.next();
            bundle.putCharSequence(allremoteInputs[i].getResultKey(), msg);
            i++;
        }
        RemoteInput.addResultsToIntent(allremoteInputs, localIntent, bundle);
        try {
            if (notification != null)
                Objects.requireNonNull(replyAction(notification)).actionIntent.send(whatsAppNotification, 0, localIntent);
            else

                Log.e(LOG, "NotificationReceiver notification is null");
        } catch (PendingIntent.CanceledException e) {
            Log.e(LOG, "replyToLastNotification error: " + e.getLocalizedMessage());
        }
    }

    /*
     * Returns Reply Action of Notification
     */
    @RequiresApi(api = Build.VERSION_CODES.KITKAT)
    private NotificationCompat.Action replyAction(Notification notification) {
        NotificationCompat.Action action;
        for (NotificationCompat.Action action2 : new NotificationCompat.WearableExtender(notification).getActions()) {
            if (isAllowFreeFormInput(action2)) {
                return action2;
            }
        }
        if (!(notification == null || notification.actions == null)) {
            for (int i = 0; i < NotificationCompat.getActionCount(notification); i++) {
                action = NotificationCompat.getAction(notification, i);
                if (isAllowFreeFormInput(action)) {
                    return action;
                }
            }
        }
        return null;
    }

    private boolean isAllowFreeFormInput(NotificationCompat.Action action) {
        if (action.getRemoteInputs() == null) {
            return false;
        }
        for (RemoteInput allowFreeFormInput : action.getRemoteInputs()) {
            if (allowFreeFormInput.getAllowFreeFormInput()) {
                return true;
            }
        }
        return false;
    }

}