我有一个生产者和许多消费者。
- 制作人速度快，产生了很多结果
- 具有相同值的令牌需要按顺序处理
- 具有不同值的令牌必须并行处理
- 创建新的Runnable将非常昂贵，而且生产代码可以使用100k的Token（为了创建Runnable，我必须将一些复杂的构建对象传递给构造函数）
我能用一个更简单的算法获得同样的结果吗？用可重入锁嵌套同步块似乎有点不自然。有什么比赛条件你可能会注意到吗？
更新：我发现的第二个解决方案是使用3个集合。一个是缓存生产者的结果，第二个是阻塞队列，第三个是使用列表跟踪正在进行的任务。又有点复杂了。
我的代码版本
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
public class Main1 {
static class Token {
private int order;
private String value;
Token() {
}
Token(int o, String v) {
order = o;
value = v;
}
int getOrder() {
return order;
}
String getValue() {
return value;
}
}
private final static BlockingQueue<Token> queue = new ArrayBlockingQueue<Token>(10);
private final static ConcurrentMap<String, Object> locks = new ConcurrentHashMap<String, Object>();
private final static ReentrantLock reentrantLock = new ReentrantLock();
private final static Token STOP_TOKEN = new Token();
private final static List<String> lockList = Collections.synchronizedList(new ArrayList<String>());
public static void main(String[] args) {
ExecutorService producerExecutor = Executors.newSingleThreadExecutor();
producerExecutor.submit(new Runnable() {
public void run() {
Random random = new Random();
try {
for (int i = 1; i <= 100; i++) {
Token token = new Token(i, String.valueOf(random.nextInt(1)));
queue.put(token);
}
queue.put(STOP_TOKEN);
}catch(InterruptedException e){
e.printStackTrace();
}
}
});
ExecutorService consumerExecutor = Executors.newFixedThreadPool(10);
for(int i=1; i<=10;i++) {
// creating to many runnable would be inefficient because of this complex not thread safe object
final Object dependecy = new Object(); //new ComplexDependecy()
consumerExecutor.submit(new Runnable() {
public void run() {
while(true) {
try {
//not in order
Token token = queue.take();
if (token == STOP_TOKEN) {
queue.add(STOP_TOKEN);
return;
}
System.out.println("Task start" + Thread.currentThread().getId() + " order " + token.getOrder());
Random random = new Random();
Thread.sleep(random.nextInt(200)); //doLongRunningTask(dependecy)
lockList.remove(token.getValue());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}});
}
}}
您可以预先创建一组Runnables，它将拾取传入任务（令牌），并根据它们的顺序值将它们放置在队列中。
正如评论中所指出的，不能保证具有不同值的令牌总是并行执行（总而言之，你至少受到盒子中物理核心数量的限制）。但是，可以保证具有相同顺序的令牌将按到达顺序执行。
样本代码：
/**
* Executor which ensures incoming tasks are executed in queues according to provided key (see {@link Task#getOrder()}).
*/
public class TasksOrderingExecutor {
public interface Task extends Runnable {
/**
* @return ordering value which will be used to sequence tasks with the same value.<br>
* Tasks with different ordering values <i>may</i> be executed in parallel, but not guaranteed to.
*/
String getOrder();
}
private static class Worker implements Runnable {
private final LinkedBlockingQueue<Task> tasks = new LinkedBlockingQueue<>();
private volatile boolean stopped;
void schedule(Task task) {
tasks.add(task);
}
void stop() {
stopped = true;
}
@Override
public void run() {
while (!stopped) {
try {
Task task = tasks.take();
task.run();
} catch (InterruptedException ie) {
// perhaps, handle somehow
}
}
}
}
private final Worker[] workers;
private final ExecutorService executorService;
/**
* @param queuesNr nr of concurrent task queues
*/
public TasksOrderingExecutor(int queuesNr) {
Preconditions.checkArgument(queuesNr >= 1, "queuesNr >= 1");
executorService = new ThreadPoolExecutor(queuesNr, queuesNr, 0, TimeUnit.SECONDS, new SynchronousQueue<>());
workers = new Worker[queuesNr];
for (int i = 0; i < queuesNr; i++) {
Worker worker = new Worker();
executorService.submit(worker);
workers[i] = worker;
}
}
public void submit(Task task) {
Worker worker = getWorker(task);
worker.schedule(task);
}
public void stop() {
for (Worker w : workers) w.stop();
executorService.shutdown();
}
private Worker getWorker(Task task) {
return workers[task.getOrder().hashCode() % workers.length];
}
}
根据代码的性质，确保以串行方式处理相同的值是等待STOP_TOKEN到达。
您需要单生产者单消费者设置，消费者收集和排序代币的价值（比如说，进入Multimap）。
只有这样，您才能知道哪些令牌可以串行处理，哪些令牌可以并行处理。
无论如何，我建议您看看LMAX Disruptor，它提供了在线程之间共享数据的非常有效的方法。
它不受执行器同步开销的影响，因为它是无锁的（这可能会给您带来很好的性能优势，这取决于您处理数据的方式）。
使用两个Disruptor的解决方案
// single thread for processing as there will be only on consumer
Disruptor<InEvent> inboundDisruptor = new Disruptor<>(InEvent::new, 32, Executors.newSingleThreadExecutor());
// outbound disruptor that uses 3 threads for event processing
Disruptor<OutEvent> outboundDisruptor = new Disruptor<>(OutEvent::new, 32, Executors.newFixedThreadPool(3));
inboundDisruptor.handleEventsWith(new InEventHandler(outboundDisruptor));
// setup 3 event handlers, doing round robin consuming, effectively processing OutEvents in 3 threads
outboundDisruptor.handleEventsWith(new OutEventHandler(0, 3, new Object()));
outboundDisruptor.handleEventsWith(new OutEventHandler(1, 3, new Object()));
outboundDisruptor.handleEventsWith(new OutEventHandler(2, 3, new Object()));
inboundDisruptor.start();
outboundDisruptor.start();
// publisher code
for (int i = 0; i < 10; i++) {
inboundDisruptor.publishEvent(InEventTranslator.INSTANCE, new Token());
}
入站中断器上的事件处理程序只收集传入的令牌。当收到STOP令牌时，它会将一系列令牌发布到出站中断器以进行进一步处理：
public class InEventHandler implements EventHandler<InEvent> {
private ListMultimap<String, Token> tokensByValue = ArrayListMultimap.create();
private Disruptor<OutEvent> outboundDisruptor;
public InEventHandler(Disruptor<OutEvent> outboundDisruptor) {
this.outboundDisruptor = outboundDisruptor;
}
@Override
public void onEvent(InEvent event, long sequence, boolean endOfBatch) throws Exception {
if (event.token == STOP_TOKEN) {
// publish indexed tokens to outbound disruptor for parallel processing
tokensByValue.asMap().entrySet().stream().forEach(entry -> outboundDisruptor.publishEvent(OutEventTranslator.INSTANCE, entry.getValue()));
} else {
tokensByValue.put(event.token.value, event.token);
}
}
}
出站事件处理程序按顺序处理相同值的令牌：
public class OutEventHandler implements EventHandler<OutEvent> {
private final long order;
private final long allHandlersCount;
private Object yourComplexDependency;
public OutEventHandler(long order, long allHandlersCount, Object yourComplexDependency) {
this.order = order;
this.allHandlersCount = allHandlersCount;
this.yourComplexDependency = yourComplexDependency;
}
@Override
public void onEvent(OutEvent event, long sequence, boolean endOfBatch) throws Exception {
if (sequence % allHandlersCount != order ) {
// round robin, do not consume every event to allow parallel processing
return;
}
for (Token token : event.tokensToProcessSerially) {
// do procesing of the token using your complex class
}
}
}
所需基础设施的其余部分（目的在Disruptor文档中描述）：
public class InEventTranslator implements EventTranslatorOneArg<InEvent, Token> {
public static final InEventTranslator INSTANCE = new InEventTranslator();
@Override
public void translateTo(InEvent event, long sequence, Token arg0) {
event.token = arg0;
}
}
public class OutEventTranslator implements EventTranslatorOneArg<OutEvent, Collection<Token>> {
public static final OutEventTranslator INSTANCE = new OutEventTranslator();
@Override
public void translateTo(OutEvent event, long sequence, Collection<Token> tokens) {
event.tokensToProcessSerially = tokens;
}
}
public class InEvent {
// Note that no synchronization is used here,
// even though the field is used among multiple threads.
// Memory barrier used by Disruptor guarantee changes are visible.
public Token token;
}
public class OutEvent {
// ... again, no locks.
public Collection<Token> tokensToProcessSerially;
}
public class Token {
String value;
}
如果你有很多不同的令牌，那么最简单的解决方案是创建一些数量的单线程执行器（大约是你的核心数量的两倍），然后将每个任务分配给一个由其令牌的哈希确定的执行器。
这样，所有具有相同令牌的任务都将转到同一个执行器并按顺序执行，因为每个执行器只有一个线程。
如果您对调度公平性有一些未说明的要求，那么通过让生产者线程在分发请求之前对其请求（或块）进行排队，直到每个执行器的未决请求少于10个，就可以很容易地避免任何严重的不平衡。
以下解决方案将仅使用生产者和消费者使用的单个Map来按顺序处理每个订单号的订单，同时并行处理不同的订单号。这是代码：
public class Main {
private static final int NUMBER_OF_CONSUMER_THREADS = 10;
private static volatile int sync = 0;
public static void main(String[] args) {
final ConcurrentHashMap<String,Controller> queues = new ConcurrentHashMap<String, Controller>();
final CountDownLatch latch = new CountDownLatch(NUMBER_OF_CONSUMER_THREADS);
final AtomicBoolean done = new AtomicBoolean(false);
// Create a Producer
new Thread() {
{
this.setDaemon(true);
this.setName("Producer");
this.start();
}
public void run() {
Random rand = new Random();
for(int i =0 ; i < 1000 ; i++) {
int order = rand.nextInt(20);
String key = String.valueOf(order);
String value = String.valueOf(rand.nextInt());
Controller controller = queues.get(key);
if (controller == null) {
controller = new Controller();
queues.put(key, controller);
}
controller.add(new Token(order, value));
Main.sync++;
}
done.set(true);
}
};
while (queues.size() < 10) {
try {
// Allow the producer to generate several entries that need to
// be processed.
Thread.sleep(5000);
} catch (InterruptedException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
}
// System.out.println(queues);
// Create the Consumers
ExecutorService consumers = Executors.newFixedThreadPool(NUMBER_OF_CONSUMER_THREADS);
for(int i = 0 ; i < NUMBER_OF_CONSUMER_THREADS ; i++) {
consumers.submit(new Runnable() {
private Random rand = new Random();
public void run() {
String name = Thread.currentThread().getName();
try {
boolean one_last_time = false;
while (true) {
for (Map.Entry<String, Controller> entry : queues.entrySet()) {
Controller controller = entry.getValue();
if (controller.lock(this)) {
ConcurrentLinkedQueue<Token> list = controller.getList();
Token token;
while ((token = list.poll()) != null) {
try {
System.out.println(name + " processing order: " + token.getOrder()
+ " value: " + token.getValue());
Thread.sleep(rand.nextInt(200));
} catch (InterruptedException e) {
}
}
int last = Main.sync;
queues.remove(entry.getKey());
while(done.get() == false && last == Main.sync) {
// yield until the producer has added at least another entry
Thread.yield();
}
// Purge any new entries added
while ((token = list.poll()) != null) {
try {
System.out.println(name + " processing order: " + token.getOrder()
+ " value: " + token.getValue());
Thread.sleep(200);
} catch (InterruptedException e) {
}
}
controller.unlock(this);
}
}
if (one_last_time) {
return;
}
if (done.get()) {
one_last_time = true;
}
}
} finally {
latch.countDown();
}
}
});
}
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
consumers.shutdown();
System.out.println("Exiting.. remaining number of entries: " + queues.size());
}
}
请注意，Main类包含一个队列实例，该实例是Map。map键是要由使用者按顺序处理的订单id。该值是一个Controller类，它将包含与该订单id关联的所有订单。
生产者将生成订单，并将订单（令牌）添加到其关联的控制器中。使用者将对队列映射值进行迭代，并调用Controller锁方法来确定它是否可以处理该特定订单id的订单。如果锁返回false，它将检查下一个Controller实例。如果锁返回true，它将处理所有订单，然后检查下一个Controller。
已更新添加了同步整数，该整数用于确保从队列映射中删除控制器实例时。它的所有条目都将被消耗掉。在调用unlock方法的使用者代码中存在逻辑错误。
Token类与您在这里发布的类似。
class Token {
private int order;
private String value;
Token(int order, String value) {
this.order = order;
this.value = value;
}
int getOrder() {
return order;
}
String getValue() {
return value;
}
@Override
public String toString() {
return "Token [order=" + order + ", value=" + value + "]n";
}
}
下面的Controller类用于确保线程池中只有一个线程将处理订单。锁定/解锁方法用于确定哪些线程将被允许处理订单。
class Controller {
private ConcurrentLinkedQueue<Token> tokens = new ConcurrentLinkedQueue<Token>();
private ReentrantLock lock = new ReentrantLock();
private Runnable current = null;
void add(Token token) {
tokens.add(token);
}
public ConcurrentLinkedQueue<Token> getList() {
return tokens;
}
public void unlock(Runnable runnable) {
lock.lock();
try {
if (current == runnable) {
current = null;
}
} finally {
lock.unlock();
}
}
public boolean lock(Runnable runnable) {
lock.lock();
try {
if (current == null) {
current = runnable;
}
} finally {
lock.unlock();
}
return current == runnable;
}
@Override
public String toString() {
return "Controller [tokens=" + tokens + "]";
}
}
有关实施的其他信息。它使用CountDownLatch来确保在流程退出之前处理所有生成的订单。done变量与STOP_TOKEN变量一样。
实现中确实包含一个您需要解决的问题。存在的问题是，在处理完所有订单后，它不会清除订单id的控制器。这将导致线程池中的线程被分配给不包含订单的控制器。这将浪费可用于执行其他任务的cpu周期。
您所需要的只是确保不同时处理具有相同值的令牌吗？你的代码太乱了，无法理解你的意思（它不编译，并且有很多未使用的变量、锁和映射，这些都是创建的，但从未使用过）。看起来你想得太多了。您只需要一个队列和一个映射。我想是这样的：
class Consumer implements Runnable {
ConcurrentHashMap<String, Token> inProcess;
BlockingQueue<Token> queue;
public void run() {
Token token = null;
while ((token = queue.take()) != null) {
if(inProcess.putIfAbsent(token.getValue(), token) != null) {
queue.put(token);
continue;
}
processToken(token);
inProcess.remove(token.getValue());
}
}
}
具有相同值的令牌需要按顺序处理
确保任何两件事按顺序发生的方法是在同一个线程中进行。
不管有多少工作线程，我都会有一个集合，并且我会有一张Map。每当我得到一个以前从未见过的令牌时，我都会随机选择一个线程，并将令牌和线程输入到映射中。从那时起，我将使用同一个线程来执行与该令牌相关联的任务。
创建新的Runnables将是非常昂贵的
Runnable是一个接口。创建实现Runnable的新对象不会比创建任何其他类型的对象花费更多。
也许我误解了什么。但最初似乎更容易将具有相同值的令牌从具有不同值的令牌过滤到两个不同的队列中。
然后将Stream与map或foreach一起用于序列。只需使用并行流版本即可完成其余操作
如果生产环境中的令牌是延迟生成的，并且一次只得到一个，那么只需制作某种过滤器，将它们分发到两个不同的队列。
如果你能用Streams实现它，我会这么做，因为它们简单、易用、快速！
https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html
我举了一个简短的例子说明我的意思。在这种情况下，数字代币是人工构建的，但这与重点无关。此外，流都是在主线程上启动的，这可能也不理想。
public static void main(String args[]) {
ArrayList<Token> sameValues = new ArrayList<Token>();
ArrayList<Token> distinctValues = new ArrayList<Token>();
Random random = new Random();
for (int i = 0; i < 100; i++) {
int next = random.nextInt(100);
Token n = new Token(i, String.valueOf(next));
if (next == i) {
sameValues.add(n);
} else {
distinctValues.add(n);
}
}
distinctValues.stream().parallel().forEach(token -> System.out.println("Distinct: " + token.value));
sameValues.stream().forEach(token -> System.out.println("Same: " + token.value));
}
我不完全确定我是否理解了这个问题，但我会尝试一下算法。
参与者是：
- 任务的
queue - 自由
executors的一个pool - 当前正在处理的
in-process令牌中的set controller
然后，
-
最初，所有
executors可用，而set为空 -
controller选择一个可用的executor，并通过queue寻找具有不在in-process set中的令牌的task- 将令牌添加到
in-process集合 - 分配CCD_ 18来处理CCD_
- 返回到队列的开头
- 将令牌添加到
-
CCD_ 20在完成处理时从CCD_ 21中移除令牌，并将其自身添加回池
实现这一点的一种方法是让一个执行器用于序列处理，另一个用于并行处理。我们还需要一个单线程管理器服务，它将决定需要提交给哪个服务令牌进行处理。//要由两个线程共享的队列。包含生产者生成的令牌
BlockingQueue标记列表=新的ArrayBlockingQueue（10）；
private void startProcess() {
ExecutorService producer = Executors.newSingleThreadExecutor();
final ExecutorService consumerForSequence = Executors
.newSingleThreadExecutor();
final ExecutorService consumerForParallel = Executors.newFixedThreadPool(10);
ExecutorService manager = Executors.newSingleThreadExecutor();
producer.submit(new Producer(tokenList));
manager.submit(new Runnable() {
public void run() {
try {
while (true) {
Token t = tokenList.take();
System.out.println("consumed- " + t.orderid
+ " element");
if (t.orderid % 7 == 0) { // any condition to check for sequence processing
consumerForSequence.submit(new ConsumerForSequenceProcess(t));
} else {
ConsumerForParallel.submit(new ConsumerForParallelProcess(t));
}
}
}
catch (InterruptedException e) { // TODO Auto-generated catch
// block
e.printStackTrace();
}
}
});
}
我认为这项任务背后隐藏着一个更基本的设计问题，但没关系。我无法从你的问题描述中弄清楚你是想按顺序执行，还是只想对由单个令牌描述的任务进行原子/事务操作。我在下面提出的建议更像是对这个问题的"快速解决方案"，而不是真正的解决方案。
对于真正的"有序执行"情况，我提出了一种基于排序输出的队列代理的解决方案：
-
定义Queue的实现，该实现提供了一种生成代理队列的工厂方法，该代理队列由该单个队列对象表示给生产者；工厂方法还应该注册这些代理队列对象。如果将元素添加到输入队列中与某个输出队列中的某个元素匹配，则应将其直接添加到某一输出队列中。否则，将其添加到任何（最短的）输出队列中。（有效地进行检查）。或者（稍微好一点）：当添加元素时不要这样做，但当任何输出队列都为空时。
-
为每个可运行的使用者提供一个字段，存储一个单独的队列接口（而不是访问单个对象）。通过上面定义的工厂方法初始化此字段。
对于事务情况，我认为跨越比核心更多的线程更容易（使用统计数据来计算），并在较低的（对象）级别上实现阻塞机制。