从 Java 中的多个线程发送消息

Question

我需要从 SomeObjectHandler 实现 handle() 方法，它将 SomeObject 委托给某个外部系统（见下文）。 SomeObject 具有正确的 hashCode 和 equals 方法。方法句柄（SomeObject someObject）可以从许多线程（例如10个）调用。外部系统可以同时操作不等于 someObject 就好了，但是如果系统尝试同时操作等于 someObject 就会中断。我需要实现这个类来防止同时处理相等的 someObject。即使一些 someObject 是相等的，它们都应该被处理。

现在，我在想我需要使用并发库中的队列之类的东西，但我不知道是哪一个。

UPD：我只需要使用标准 Java 库。如果可以实现最大吞吐量。

Answer 1

我不能 100% 确定我是否完全理解了你的问题，但我认为有多种方法可以解决这个问题。

1）正如您已经提到的，您可以使用队列来插入对象并确保外部系统以同步方式处理对象，因为正如您所说，您的外部系统无法同时处理相同的对象。

2) 在发件人代码中自行处理。我已经尝试过多次了。这是一个代码sn-p。这种方法的优点是它只同步相等的对象。只要确保你也处理了 finally 块中的删除部分。对不起，如果代码不整洁。我是新手 :)

Map<SomeObject,Integer> objMap=new ConcurrentHashMap<SomeObject,Integer>();

public void handle(SomeObject someObject) {
synchronized(this.class)
{
Integer count=objMap.get(someObject);  
if(count==null)
{
   count=0;
}
objMap.put(someObject,++count);
}

synchronized(objectMap.get(someObject)
{
  outerSystem.process(someObject);

  Integer count=objMap.get(someObject);  
  if(count>1)
 {
   objMap.put(someObject,--count);
 }
 else
 {
   objectMap.remove(someObject);
 }
}

}

Answer 2

RxJava 可以在这里提供帮助。它非常擅长处理数据流，尤其是涉及异步转换的数据流，并在需要时进行您正在谈论的排队（无需通过同步修饰符阻塞！）。为了解决您的问题，我会这样做：

public class SomeHandler{

   private final OuterSystem outerSystem;

   private final PublishSubject<SomeObject> subject;

   public SomeHandler() {
       subject
          // handle calls from multiple threads (non-blocking)
          .serialized() 
          // buffer in memory if not keeping up       
          .onBackpressureBuffer() 
          // use equals/hashCode to order messages (the queues you referred to)
          .groupBy(x -> x)
          .flatMap(g -> 
              g.doOnNext(x -> outerSystem.process(x))
               // process groups in parallel
               .subscribeOn(Schedulers.computation()))
          // do something if an error occurs
          .doOnError(e -> e.printStackTrace())
          // start consuming data when arrives
          .subscribe();
   }

   public void handle(SomeObject someObject) {
      subject.doOnNext(someObject);
   }
}

Answer 3

如果我正确理解您的问题，您需要保证“相等对象”的串行执行，而“不相等对象”可以并行处理。实现这一点的一种方法是安排 N 个处理器并根据对象的某些确定性特征分散工作负载。

在您的情况下，如果两个对象相等，则它们的哈希码必须相等，因此hasCode() modulo N 可用于将负载分散到 N 个 Executor 上，其中每个 Executor 仅包含一个线程：

public class SomeHandler {
   static int N = ...;
   // Each executor is an Executors.newSingleThreadScheduledExecutor()
   Executor[N] executors = ....;
   OuterSystem system;

   public void handle(SomeObject so) {
       executors[so.hashCode() % N].execute(() -> system.process(so));
   }
}