我有一个web服务器服务,其中客户端请求智能卡计算并获得其结果。在服务器正常运行期间,可用的智能卡数量可以减少或增加,例如,我可以从读卡器中物理添加或移除智能卡(或许多其他事件...如异常等)。
智能卡计算可能需要一段时间,因此我必须优化这些作业,以便在存在对web服务器的并发请求时使用所有可用的智能卡。
我想用智能卡线程池来工作。不同寻常的是,至少对我来说,池的大小应该不依赖于客户端请求,而只依赖于智能卡的可用性。
我研究了许多例子:
这是智能卡控件,每个智能卡都有一个SmartcardWrapper,每个智能卡都有自己的插槽编号。
public class SmartcardWrapper{
private int slot;
public SmartcardWrapper(int slot) {
this.slot=slot;
}
public byte[] compute(byte[] input) {
byte[] out=new byte[];
SmartcardApi.computerInput(slot,input,out); //Native method
return out;
}
}
我尝试创建一个线程池,每个智能卡一个线程:
private class SmartcardThread extends Thread{
protected SmartcardWrapper sw;
public SmartcardThread(SmartcardWrapper sw){
this.sw=sw;
}
@Override
public void run() {
while(true){
byte[] input=queue.take();
byte output=sw.compute(input);
// I have to return back the output to the client
}
}
}
每个人都在等待同一个输入队列中的某些东西:
BlockingQueue<byte[]> queue=new BlockingQueue<byte[]>();
但是如何将智能卡线程的输出返回给webserver-client呢?这让我觉得BlockingQueue不是我的解决方案。
如何解决这个问题?我应该遵循哪种并发模式?为每个智能卡分配一个线程是否正确,或者我应该简单地使用信号量吗?
发布于 2015-12-16 01:25:54
你的假设是:
ThreadPoolExecutor:看起来我需要这个,但是我不能改变池的大小,而且每个线程都应该链接到一个智能卡插槽。
是不对的。
You can set thread pool size dynamically.
看一下下面的ThreadPoolExecutor API
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);
或
发布于 2015-12-19 01:31:54
你有没有考虑过使用Apache Commons Pool?
您需要维护一个SmartcardWrapper对象池,其中每个SmartcardWrapper将表示一个物理SmartCard。当您需要进行新的计算时,您可以从池中借用对象,执行计算并返回池中的对象,以便它可以被下一个线程重用。
池本身是线程安全的,并且在没有可用的对象时阻塞。您所需要做的就是实现一个应用程序接口来向池中添加/删除SmartcardWrapper对象。
发布于 2015-12-20 09:20:01
基于以下假设,我可能已经找到了一个合理而简单的解决方案:
这两个假设实际上使创建池化(共享资源)解决方案变得更容易,因为在适当的时候,通常是池本身负责创建和删除资源。如果没有此功能,池化解决方案将变得更简单。我假设从池中获取要使用的智能卡的客户端可以在其自己的执行线程中执行所需的智能卡功能(类似于如何从数据库连接池使用数据库连接来查询数据库中的数据)。
我只对下面显示的两个类做了一些最小的测试,恐怕大部分工作都是在编写(单元)测试中进行的,这些测试证明池在并发客户端请求以及添加和删除智能卡资源的情况下可以正常工作。如果您不想这样做,那么answer from 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
}
}
https://stackoverflow.com/questions/34138795
复制相似问题