当使用 HornetQ 对 DB 请求进行审计日志记录时，开始日志条目并不总是在结束日志条目之前提交

Question

我有一个使用 JBoss AS 7、Hibernate 和 HornetQ 的应用程序。我们使用 HornetQ 来处理进入应用程序的所有请求的日志记录。当一个请求进来时，我们发送一个“开始日志条目”消息，当使用该消息时，记录请求的 UUID、它的开始时间以及与请求相关的其他各种数据。当请求完成时，我们发送一条“结束日志条目”消息，当被消费时从数据库中查找起始日志条目（带有 UUID），并将结束时间添加到请求中。

经常发生的问题是在“结束日志条目”中完成的数据库查找找不到匹配的起始条目，因为实体管理器实际上还没有将起始日志条目提交/刷新到数据库。在结束日志条目尝试查找它之前，我需要保证开始日志条目在数据库中。有什么想法可以确保发生这种情况吗？

以下是我们拥有的相关类：

LogFilter.java - 这是 web.xml 中定义的过滤器，用于拦截所有请求并记录它们

public class LogFilter implements Filter {
    @Inject
    LoggingService ls;

    public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException {
      // do stuff to create the log entry

      try {
        ls.startLogEntry(logEntry);
        filterChain.doFilter(servletRequest, servletResponse);
      } finally {
          try {
            ls.endLogEntry(logEntry);
          } catch (Exception e2) {
            // log stacktrace to the server logs
          }
      }
   }
}

LoggingService.java - 注入 LogFilter 的服务

public class LoggingService {
  private static Log logger = LogFactory.getLog(LoggingService.class);
  @PersistenceContext(unitName="p1")
  private EntityManager em;

  // find a log entry by its generated uuid when the request started
  public LogEntry getLogEntry(String uuid) {
    CriteriaBuilder cb = em.getCriteriaBuilder();
    CriteriaQuery<LogEntry> c = cb.createQuery(LogEntry.class);
    Root<LogEntry> root = c.from(LogEntry.class);

    c.where(cb.equal(root.get(LogEntry_.uuid), uuid));

    TypedQuery<LogEntry> tq = em.createQuery(c);
    return JpaUtil.getSingleResultOrNull(tq);
  }

  // send message to write the log entry without the end time
  public void startLogEntry(LogEntry logEntry) {
    AuditLogProducer.sendLogEntryToQueue(logEntry);
  }

  public void endLogEntry(LogEntry logEntry) {
    if (logEntry != null) {
      logEntry.setRequestEnd(new Date());
      AuditLogProducer.sendLogEntryToQueue(logEntry);
    }
  }
}

AuditLogProducer.java - 负责将消息发送到队列的类

public class AuditLogProducer {
  private static final String QUEUE_LOOKUP = "jboss/queue/AuditLogQueue";
  private static final String CONNECTION_FACTORY = "ConnectionFactory";
  private static Log log = LogFactory.getLog(AuditLogProducer.class);

  public static void sendLogEntryToQueue(LogEntry logEntry) {
    QueueSession session = null;
    QueueConnection connection = null;

    try {
      Context context = new InitialContext();
      QueueConnectionFactory factory = (QueueConnectionFactory) context.lookup(CONNECTION_FACTORY);
      connection = factory.createQueueConnection();
      session = connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
      Queue queue = (Queue) context.lookup(QUEUE_LOOKUP);

      QueueSender sender = session.createSender(queue);
      ObjectMessage message = session.createObjectMessage();

      message.setObject(logEntry);
      sender.send(message);
      session.close();
    } catch (Exception e) {
      log.error("problem sending message to queue: " + e.getMessage());
    } finally {
      if (session != null) {
        try {
          session.close();
          connection.close();
        } catch (Exception e) { }
      }
    }
  }
}

AuditLogConsumer.java - 负责从队列中消费消息的类

@MessageDriven(
  activationConfig={
      @ActivationConfigProperty(
        propertyName="destinationType",
        propertyValue="javax.jms.Queue"
      ), @ActivationConfigProperty(
        propertyName="destination",
        propertyValue="queue/AuditLogQueue"
      )
    }
)
public class AuditLogConsumer implements MessageListener {
  private static Log log = LogFactory.getLog(AuditLogConsumer.class);

  @Inject
  LogBuilder logBuilder;

  @Inject
  LogPolisher logPolisher;

  @Override
  public void onMessage(Message message) {
    if (message instanceof ObjectMessage) {
      ObjectMessage msg = (ObjectMessage) message;
      LogEntry logEntry = null;

      try {
        logEntry = (LogEntry) msg.getObject();
      } catch (JMSException e) {
        log.error("Problem retrieving JMS message from the AuditLogQueue: " + e.getMessage(), e);
        return;
      }

      if (logEntry.getRequestEnd() == null) {
        logBuilder.insertLogEntry(logEntry);
      } else {
        logPolisher.updateLogEntry(logEntry);
      }
    }
  }
}

LogBuilder.java - 单独负责将日志条目合并到数据库中的类 - 曾经属于一个类，但希望将创建和更新分为 2 个类，以确保将开始写入数据库准时（最终没有工作）

@Singleton
public class LogBuilder {
  @PersistenceContext(unitName="p1")
  private EntityManager em;

  @TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
  public void insertLogEntry(LogEntry log) {
    em.merge(log);
    em.flush();
  }
}

LogPolisher.java - 单独负责查找开始日志条目、设置结束时间以及将新日志合并到数据库的类 - newLog.setRequestEnd(requestEnd) 有时会在上一行返回 null 时失败，因为它没有找到开始日志条目（因为它还没有被写入）。

@Singleton
public class LogPolisher {
  private static Log log = LogFactory.getLog(LogPolisher.class);
  @Inject
  LoggingService loggingService;

  @PersistenceContext(unitName="p1")
  private EntityManager em;

  @TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
  public void updateLogEntry(LogEntry logEntry) {
    Date requestEnd = logEntry.getRequestEnd();

    try {
      LogEntry newLog = loggingService.getLogEntry(logEntry.getUuid());

      newLog.setRequestEnd(requestEnd);
      em.merge(newLog);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }
}

LogEntry.java 只是一个普通的休眠 bean，我不会在这里浪费更多的空间来展示它的代码。

Answer 1

我最初不想回答这个问题，因为我很难理解您的问题。但我发现了一些可能有助于您确定问题的问题。

我 - 您似乎混淆了消息传递和数据库。这似乎很明显，但这是一个很常见的误解。消息传递用于异步传递消息。 （从不存储）。您实际上需要将数据库与队列匹配这一事实可能表明您的架构存在问题。

II - 性能会影响这些生产者/消费者...它们都是同步的，因为您正在为每个日志条目创建消费者或生产者。您应该缓存对象并在每个线程中重用它们。这可能表明正在使用反模式 (i)。

III - 我不确定这里是否使用了 XA，但某些事务管理器 (TM) 不能保证首先提交分支的顺序。新版本的事务管理器正在修复此问题。

IV - 您正在使用另一个框架 (JPA) 构建一个框架（您的日志工具）。除非你需要动态的东西。我会直接使用 JDBC。否则你会被绑定在一个框架*框架上。它变得复杂......但这只是恕我直言。

我只是想在这里为您提供一些帮助.. 即使它没有直接回答您的问题。