防止消息处理的竞争条件

Question

我有一个通过 Web 服务接收消息（事件）的 J2EE 应用程序。这些消息具有不同的类型（根据类型需要不同的处理）并以特定的顺序发送。它发现了一个问题，即某些消息类型的处理时间比其他消息类型要长。结果是序列中第二个接收到的消息可能会在序列中第一个消息之前被处理。我尝试通过在处理消息的方法周围放置一个同步块来解决这个问题。这似乎可行，但我不确定这是“正确”的方法吗？是否有更合适的替代方案或者这是“可接受的”？我添加了一小段代码来尝试更清楚地解释。 ....任何建议/指导表示赞赏。

public class EventServiceImpl implements EventService {
  public String submit (String msg) {

    if (msg == null)
        return ("NAK");

            EventQueue.getInstance().submit(msg);

    return "ACK";
  }
}


public class EventQueue {
    private static EventQueue instance = null;
    private static int QUEUE_LENGTH = 10000;
    protected boolean done = false;
    BlockingQueue<String> myQueue = new LinkedBlockingQueue<String>(QUEUE_LENGTH);

protected EventQueue() {
    new Thread(new Consumer(myQueue)).start();
}

public static EventQueue getInstance() {
      if(instance == null) {
         instance = new EventQueue();
      }
      return instance;
}

public void submit(String event) {
    try {
        myQueue.put(event);
    } catch (InterruptedException ex) {
    }
}

class Consumer implements Runnable {
    protected BlockingQueue<String> queue;

    Consumer(BlockingQueue<String> theQueue) { this.queue = theQueue; }

    public void run() {
      try {
        while (true) {
          Object obj = queue.take();
          process(obj);
          if (done) {
            return;
          }
        }
      } catch (InterruptedException ex) {
      }
    }

    void process(Object obj) {
        Event event = new Event( (String) obj);
        EventHandler handler = EventHandlerFactory.getInstance(event);
        handler.execute();
    }
}

// Close queue gracefully
public void close() {
    this.done = true;
}

Answer 1

我不确定您正在使用的框架（EJB（MDB）/JMS）是什么。通常应避免在诸如 EJB/JMS 之类的托管环境中使用同步（这不是一个好的做法）。一种解决方法是

• 客户端在发送下一条消息之前应等待服务器的确认。
• 这样您的客户端本身就可以控制事件的顺序。

请注意，如果有多个客户端提交消息，这将不起作用。

编辑：

您遇到的情况是，Web 服务的客户端按顺序发送消息，而不考虑消息处理时间。它只是一个接一个地转储消息。这是基于队列（先进先出） 的解决方案的一个很好的案例。我建议采用以下两种方法来实现此目的

1. 使用 JMS 。这将产生添加 JMS 提供程序 和编写一些管道代码的额外开销。< /p>

2. 使用一些多线程模式，例如 生产者-消费者，其中您的 Web 服务处理程序将是将传入消息转储到队列中，单线程消费者一次将消费一条消息。请参阅使用 java.util.concurrent 包的此示例。

3. 使用数据库。将传入消息转储到数据库中。使用不同的基于调度程序的程序来扫描数据库（基于序列号）并相应地处理消息。

   第一个和第三个解决方案对于此类问题来说是非常标准的。第二种方法速度很快，并且不需要在代码中添加任何其他库。

Answer 2

如果要按特定顺序处理事件，那么为什么不尝试在消息中添加“eventID”和“orderID”字段呢？这样，您的 EventServiceImpl 类就可以排序、排序，然后以正确的顺序执行（无论它们创建和/或传递给处理程序的顺序如何）。

我预计，同步 handler.execute() 块不会获得所需的结果。 synchronized 关键字的作用就是防止多个线程同时执行该块。它在正确排序下一个线程的领域没有做任何事情。

如果 synchronized 块似乎确实使事情正常工作，那么我断言您非常幸运，因为消息正在被创建、传递，然后按正确的顺序执行。在多线程环境下，这个就不能保证了！我会采取措施确保你能够控制这一切，而不是依靠好运。

示例：

1. 消息按照“client01-A”、“client01-C”的顺序创建，
   'client01-B', 'client01-D'
2. 消息按 'client01-D' 的顺序到达处理程序，
   'client01-B', 'client01-A', 'client01-C'
3. EventHandler 可以区分从一个客户端到另一个客户端的消息，并开始缓存 'client01' 的消息。
4. EventHandler recv 的“client01-A”消息并知道它可以处理该消息并执行此操作。
5. EventHandler 在缓存中查找消息“client01-B”，找到它并处理它。
6. EventHandler 找不到“client01-C”，因为它尚未到达。
7. EventHandler接收'client01-C'并处理它。
8. EventHandler 在缓存中查找“client01-D”，找到它、处理它，并认为“client01”交互已完成。

沿着这些思路的一些东西将确保正确的处理并促进多线程的良好使用。