Шаблон параллелизма Java для внешнего общего ресурса (смарт-карт)

Ваше предположение:

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

это не так.

You can set thread pool size dynamically.

Взгляните на следующие API-интерфейсы ThreadPoolExecutor

public void setMaximumPoolSize(int maximumPoolSize)

Устанавливает максимально допустимое количество потоков. Это отменяет любое значение, установленное в конструкторе. Если новое значение меньше текущего значения, избыточные существующие потоки будут завершены, когда они в следующий раз станут простаивающими.

public void setCorePoolSize(int corePoolSize)

Устанавливает основное количество потоков. Это отменяет любое значение, установленное в конструкторе. Если новое значение меньше текущего значения, избыточные существующие потоки будут завершены, когда они в следующий раз станут простаивающими. Если больше, то при необходимости будут запущены новые потоки для выполнения любых задач в очереди.

Core and maximum pool sizes:

ThreadPoolExecutor автоматически настроит размер пула в соответствии с границами, установленными corePoolSize и maximumPoolSize.

Когда новая задача отправляется в методе execute(java.lang.Runnable) и выполняется менее corePoolSize потоков, создается новый поток для обработки запроса, даже если другие рабочие потоки простаивают.

Если запущено больше corePoolSize, но меньше maximumPoolSize потоков, новый поток будет создан только в том случае, если очередь заполнена.

Устанавливая для maximumPoolSize практически неограниченное значение, такое как Integer.MAX_VALUE, вы позволяете пулу размещать произвольное количество одновременных задач. Но я бы не рекомендовал иметь такое большое количество потоков. Будьте осторожны при установке этого значения.

Чаще всего размеры ядра и максимального пула устанавливаются только при создании, но они также могут быть изменены динамически с помощью setCorePoolSize(int) и setMaximumPoolSize(int).

РЕДАКТИРОВАТЬ:

Для лучшего использования пула потоков, если вы знаете, что максимальное количество карт равно 6, вы можете использовать

 ExecutorService executor = Executors.newFixedThreadPool(6);

OR

Рассматривали ли вы вообще использование Apache Commons Pool?

Вам необходимо поддерживать пул объектов SmartcardWrapper, где каждый SmartcardWrapper будет представлять физическую смарт-карту. Всякий раз, когда вам нужно выполнить новое вычисление, вы заимствуете объект из пула, выполняете расчет и возвращаете объект в пуле, чтобы его можно было повторно использовать в следующем потоке.

Сам пул является потокобезопасным и блокируется, когда нет доступных объектов. Все, что вам нужно сделать, это реализовать api для добавления / удаления объектов SmartcardWrapper в пул.

Я мог бы найти разумное простое решение, основанное на следующих предположениях:

• отдельный процесс управляет уведомлениями (системными событиями) для смарт-карт, которые становятся доступными или удаляются.
• клиенту все равно, какую смарт-карту он получит, если он может использовать одну без помех.

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

Я провел лишь минимальное тестирование для двух классов, показанных ниже, и, боюсь, основная часть работы заключается в написании (модульных) тестов, которые доказывают, что пул работает правильно с одновременными клиентскими запросами в сочетании с добавлением и удалением ресурсов смарт-карты. Если вы не хотите этого делать, то, вероятно, лучшим решением будет ответ user769771. Но если вы это сделаете, попробуйте, посмотрите, подходит ли он. Идея состоит в том, что только один экземпляр пула ресурсов создается и используется всеми клиентами и обновляется отдельным процессом, который управляет доступностью смарт-карт.

import java.util.*;
import java.util.concurrent.*;

/**
 * A resource pool that expects shared resources 
 * to be added and removed from the pool by an external process
 * (i.e. not done by the pool itself, see {@link #add(Object)} and {@link #remove(Object)}.
 * <br>A {@link ResourcePoolValidator} can optionally be used. 
 * @param <T> resource type handed out by the pool.
 */
public class ResourcePool<T> {

    private final Set<T> registered = Collections.newSetFromMap(new ConcurrentHashMap<T, Boolean>()); 
    /* Use a linked list as FIFO queue for resources to lease. */
    private final List<T> available = Collections.synchronizedList(new LinkedList<T>()); 
    private final Semaphore availableLock = new Semaphore(0, true); 

    private final ResourcePoolValidator<T> validator;

    public ResourcePool() {
        this(null);
    }

    public ResourcePool(ResourcePoolValidator<T> validator) {
        super();
        this.validator = validator;
    }

    /**
     * Add a resource to the pool.
     * @return true if resource is not already in the pool.
     */
    public synchronized boolean add(T resource) {

        boolean added = false;
        if (!registered.contains(resource)) {
            registered.add(resource);
            available.add(resource);
            availableLock.release();
            added = true;
        }
        return added;
    }

    /**
     * Removes a resource from the pool.
     * The resource might be in use (see {@link #isLeased(Object)})
     * in which case {@link ResourcePoolValidator#abandoned(Object)} will be called 
     * when the resource is no longer used (i.e. released). 
     * @return true if resource was part of the pool and removed from the pool.
     */
    public synchronized boolean remove(T resource) {

        // method is synchronized to prevent multiple threads calling add and remove at the same time 
        // which could in turn bring the pool in an invalid state.
        return registered.remove(resource);
    }

    /**
     * If the given resource is (or was, see also {@link #remove(Object)} part of the pool,
     * a returned value true indicates the resource is in use / checked out.
     * <br>This is a relative expensive method, do not call it frequently.
     */
    public boolean isLeased(T resource) {
        return !available.contains(resource);
    }

    /**
     * Try to get a shared resource for usage. 
     * If a resource is acquired, it must be {@link #release(Object)}d in a finally-block.
     * @return A resource that can be exclusively used by the caller.
     * @throws InterruptedException When acquiring a resource is interrupted.
     * @throws TimeoutException When a resource is not available within the given timeout period.
     */
    public T tryAcquire(long timeout, TimeUnit tunit) throws InterruptedException, TimeoutException {

        T resource = null;
        long timeRemaining = tunit.toMillis(timeout);
        final long tend = System.currentTimeMillis() + timeRemaining;
        do {
            if (availableLock.tryAcquire(timeRemaining, TimeUnit.MILLISECONDS)) {
                resource = available.remove(0);
                if (registered.contains(resource)) {
                    boolean valid = false;
                    try {
                        valid = (validator == null ? true : validator.isValid(resource));
                    } catch (Exception e) {
                        // TODO: log exception
                        e.printStackTrace();
                    }
                    if (valid) {
                        break; // return the "checked out" resource
                    } else {
                        // remove invalid resource from pool
                        registered.remove(resource);
                        if (validator != null) {
                            validator.abandoned(resource);
                        }
                    }
                }
                // resource was removed from pool, try acquire again
                // note that this implicitly lowers the maximum available resources
                // (an acquired permit from availableLock goes unused).
                // TODO: retry puts us at the back of availableLock queue but should put us at the front of the queue
                resource = null;
            }
            timeRemaining = tend - System.currentTimeMillis();
        } while (timeRemaining > 0L);
        if (resource == null) {
            throw new TimeoutException("Unable to acquire a resource within " + tunit.toMillis(timeout) + " ms.");
        }
        return resource;
    }

    /**
     * This method must be called by the caller / client whenever {@link #tryAcquire(long, TimeUnit)}
     * has returned a resource. If the caller has determined the resource is no longer valid,
     * the caller should call {@link #remove(Object)} before calling this method.
     * @param resource no longer used.
     */
    public void release(T resource) {

        if (resource == null) {
            return;
        }
        if (registered.contains(resource)) {
            available.add(resource);
            availableLock.release();
        } else {
            if (validator != null) {
                validator.abandoned(resource);
            }
        }
    }

    /** An array (copy) of all resources registered in the pool. */
    @SuppressWarnings("unchecked")
    public T[] getRegisteredResources() {
        return (T[]) registered.toArray(new Object[registered.size()]);
    }

}

И отдельный класс с функциями, относящимися к отдельному процессу, который управляет доступностью смарт-карт.

import java.util.concurrent.TimeUnit;

/**
 * Used by a {@link ResourcePool} to validate a resource before handing it out for lease
 * (see {@link #isValid(Object)} and signal a resource is no longer used (see {@link #abandoned(Object)}). 
 */
public class ResourcePoolValidator<T> {

    /**
     * Overload this method (this method does nothing by default) 
     * to validate a resource before handing it out for lease.
     * If this method returns false or throws an exception (which it preferably should not do), 
     * the resource is removed from the pool.
     * @return true if the resource is valid for leasing
     */
    public boolean isValid(T resource) {
        return true;
    }

    /**
     * Called by the {@link ResourcePool#release(Object)} method when a resource is released by a caller 
     * but the resource was previously removed from the pool and in use.
     * <br>Called by {@link ResourcePool#tryAcquire(long, TimeUnit)} if a resource if not valid 
     * (see {@link #isValid(Object)}.
     * <br>Overload this method (this method does nothing by default) to create a notification of an unused resource,
     * do NOT do any long period of processing as this method is called from a caller (client) thread.
     */
    public void abandoned(T resource) {
        // NO-OP
    }

}

Если посмотреть на требования, то лучшей архитектурой будет отделение вычислений смарт-карты от ваших веб-сервисов.

Если полагаться на то, что веб-службы будут ожидать выполнения задач, интенсивно использующих процессор, это приведет к тайм-аутам.

Лучшее решение - это предварительное вычисление смарт-карты с использованием периодического задания и сохранение этих пар слотов и вычислений на кэш-сервере, таком как Redis.

Задание синхронизатора смарт-карт - это отдельное автономное приложение J2SE, которое периодически проверяет, какая смарт-карта доступна и активна (без ошибок), и обновляет кэш Redis с слотом и вычислениями в виде пары ключ / значение. Если смарт-карта недоступна, она будет удалена из кеша.

Веб-служба просто проверит кеш Redis для определенного ключа слота и, если найдет значение, вернет его или вернет не найденное для этого слота (недоступно или ошибка)

Этот дизайн масштабируется как со стороны смарт-карты, так и со стороны клиентских запросов.

В ответ на ваш вопрос о том, как вернуть результат вызывающему абоненту:

Все ждут чего-то в одной очереди ввода:

BlockingQueue queue = новая BlockingQueue ();

Но как вернуть вывод смарт-карты-потока клиенту-веб-серверу? Это позволило мне думать, что BlockingQueue - не мое решение.

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

Измените свою очередь отправки на:

BlockingQueue<JobSubmitRec> queue=new BlockingQueue<JobSubmitRec>();

и JobSubmitRec будет иметь byte [] и одноразовую очередь для возврата результата:

class JobSubmitRec
{
  byte[] data;
  BlockingQueue<JobSubmitResult> result=new LinkedBlockingQueue<JobSubmitResult>();
}

и ваш рабочий поток будет выглядеть примерно так:

public void run() {
 while(true){
  JobSubmitRec submitrec = queue.take();
  byte[] input = submitrec.data;
  byte output = sw.compute(input);
  submitrec.result.put( new JobSubmitResult(output) );
 }           
}

и клиент, отправивший задание, будет выглядеть так:

JobSubmitRec jsr = new JobSubmitRec( data );
queue.put( jsr );
JobSubmitResult result = jsr.result.take();
// use result here