При использовании HornetQ для ведения журнала аудита для запросов к БД начало записи журнала не всегда выполняется перед записью журнала окончания

У меня есть приложение, использующее JBoss AS 7, Hibernate и HornetQ. Мы используем HornetQ для обработки регистрации всех запросов, поступающих в приложение. По мере поступления запроса мы отправляем сообщение “start log entry”, которое при потреблении регистрирует UUID запроса, время его запуска и другие различные данные, относящиеся к запросу. Когда запрос будет завершен, мы отправим сообщение “конечная запись в журнал”, когда потребляется, просматривает начальную запись в журнале (с UUID) из базы данных и добавляет время окончания запроса.

Проблема, которая постоянно возникает, – поиск db, выполненный в “записи в конце журнала”, не находит соответствующую начальную запись, потому что диспетчер сущности фактически не совершил/не удалил начальную запись журнала в базу данных. Мне нужно гарантировать, что запись начала записи находится в базе данных до того, как запись в конце журнала попытается найти ее. Любые идеи, как это сделать, гарантируют это?

Вот соответствующие классы, которые у нас есть:

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 – это просто простой спящий компонент, который я больше не собираюсь тратить на показ своего кода.

Я изначально не хотел отвечать на это, так как мне сложно понять вашу проблему. Но я заметил несколько проблем, которые могут помочь вам определить вашу проблему.

я – Кажется, что вы путаете сообщения и базу данных. это кажется очевидным, но это довольно распространенное недоразумение. Сообщения предназначены для доставки сообщений асинхронно. (никогда не храните). Тот факт, что вам действительно нужно сопоставить базу данных с очередью, может указывать на то, что с вашей архитектурой что-то не так.

II – Производительность сосать этих производителей/потребителей… Они все синхронны, поскольку вы создаете потребителя или производителя каждую запись в журнале. вы должны кэшировать свои объекты и повторно использовать их в потоке. что, возможно, указывает на использование анти-шаблона (i).

III. Я не уверен, что XA используется здесь, но некоторые Transaction Manager (TM) не гарантируют порядок, в котором ветки совершаются в первую очередь. Это фиксируется в новых версиях диспетчера транзакций.

IV – вы создаете фреймворк (ваш инструмент регистрации), используя другую структуру (JPA). Если вам не нужна какая-то динамика. Я бы использовал JDBC напрямую. иначе вы привязаны к фреймворку framework *. Он становится сложным.. но это просто ИМХО.

Я просто пытаюсь оказать вам некоторую помощь здесь, даже если это прямо не отвечает на ваш вопрос.