ActiveMQ/JMS“丢失”消息-我错过了什么？

Question

我错过了什么？

AMQ版本5.13.2 Java 1.8.0\u 74 Windows 10

给定一个简单的测试用例，传输两条Object消息，一条带有数据，另一条是数据结束标记。只有数据结束标记被接收。

队列在作业开始时创建，并在作业完成后销毁。

如果我运行更多的事务，我会看到大约50%的接收率。

日志清楚地显示接收器在第一条消息被放入队列之前就已启动，两条消息都被放入队列，但实际上只有第二条消息被接收。

发送方和接收方都在同一个JVM上。每个都有自己的会话和连接。

连接和队列设置代码：

@Override
public void beforeJob(JobExecution jobExecution) {
    // TODO Auto-generated method stub
    try {
        jobParameters = jobExecution.getJobParameters();

        readerConnection = connectionFactory.createConnection();
        readerConnection.start();

        writerConnection = connectionFactory.createConnection();
        writerConnection.start();

        jmsQueueManagementSession = writerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        queueName = jobParameters.getString("jobName") + "." + jobExecution.getId();

        queue = jmsQueueManagementSession.createQueue(getQueueName());

    } catch (JMSException ex) {
        throw new MaxisRuntimeException(
                MaxisCodeHelperImpl.generateCode("MXAR", MXMODULE, JMS_RECEIVER_INITIALIZATION_ERROR), null);
    }

}

发件人设置代码：

@Override
public void beforeStep(StepExecution stepExecution) {

    this.stepExecution = stepExecution;
    this.setJobExecution(stepExecution.getJobExecution());
    try {
        this.connection = jmsJobExecutionListener.getWriterConnection();
        this.session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        this.messageProducer = session.createProducer(jmsJobExecutionListener.getQueue());
    } catch (JMSException ex) {
        throw new RuntimeException(ex.getMessage(), ex);
    }
}

接收器设置代码：

@Override
@BeforeStep
public void beforeStep(StepExecution stepExecution) {
    this.stepExecution = stepExecution;
    this.setJobExecution(stepExecution.getJobExecution());
    try {
        this.connection = jmsJobExecutionListener.getReaderConnection();
        this.jmsSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        this.messageConsumer = jmsSession.createConsumer(jmsJobExecutionListener.getQueue());
    }
    catch (JMSException ex)
    {
        throw new RuntimeException(ex.getMessage(), ex);
    }

}

传输代码：

    private void doSomeStuffInTransaction (final Object model) {
        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
            @Override
            protected void doInTransactionWithoutResult(TransactionStatus status) {
                try {
                    doSomeStuff ( model );

                    ObjectMessage message = session.createObjectMessage(
                            (model.getRoot() == null)
                            ? null
                            : model.getRoot().getContents().getId());
                    messageProducer.send(message);
                    logger.debug("Sent: {}", message.toString());
                }catch (Exception e) {
                      //use this to rollback exception in case of exception
                    status.setRollbackOnly();
                    throw new RuntimeException(e.getmessage(), e);
                }   

            }});
    }   

接收方代码：

public Object read() throws Exception,
        UnexpectedInputException, ParseException,
        NonTransientResourceException {

    Object result = null;

    logger.debug("Attempting to receive message on connection: ", connection.toString());

    ObjectMessage msg = (ObjectMessage) messageConsumer.receive();
    logger.debug("Received: {}", msg.toString());
    result = msg.getObject();

    return result;
}

日志剪切：

DEBUG com.mylib.SelectedDataJmsReader - Attempting to receive message on connection: 
... snip ...
*** First message ***
DEBUG org.apache.activemq.broker.region.Queue - localhost Message ID:zip-56502-1457640572818-4:2:2:1:1 sent to queue://Stuff via SQL.402
DEBUG org.apache.activemq.broker.region.Queue - queue://Stuff via SQL.402, subscriptions=2, memory=0%, size=1, pending=0 toPageIn: 1, Inflight: 0, pagedInMessages.size 0, pagedInPendingDispatch.size 0, enqueueCount: 1, dequeueCount: 0, memUsage:2214
DEBUG com.maxis.mxmove.core.SelectedDataJmsWriter - Sent: ActiveMQObjectMessage {commandId = 0, responseRequired = false, messageId = ID:zip-56502-1457640572818-4:2:2:1:1, originalDestination = null, originalTransactionId = null, producerId = null, destination = queue://Stuff via SQL.402, transactionId = null, expiration = 0, timestamp = 1457640610215, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@78cb27fd, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false}
INFO  com.maxis.mxmove.core.SelectedDataJmsWriter - Committed 1 stuff to redo log and JMS queue

*** Second Message ***
INFO  com.maxis.mxmove.core.SourceSelectionReaderImpl - Returning empty stuff and end-of-stream placeholder.
DEBUG org.apache.activemq.broker.region.Queue - localhost Message ID:zip-56502-1457640572818-4:2:2:1:2 sent to queue://Stuff via SQL.402
DEBUG org.apache.activemq.broker.region.Queue - queue://Stuff via SQL.402, subscriptions=2, memory=0%, size=2, pending=0 toPageIn: 1, Inflight: 1, pagedInMessages.size 1, pagedInPendingDispatch.size 0, enqueueCount: 2, dequeueCount: 0, memUsage:3155
DEBUG com.maxis.mxmove.core.SelectedDataJmsWriter - Sent: ActiveMQObjectMessage {commandId = 0, responseRequired = false, messageId = ID:zip-56502-1457640572818-4:2:2:1:2, originalDestination = null, originalTransactionId = null, producerId = null, destination = queue://Stuff via SQL.402, transactionId = null, expiration = 0, timestamp = 1457640610375, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false}
INFO  com.maxis.mxmove.core.SelectedDataJmsWriter - Committed 1 stuff to redo log and JMS queue

*** We received the last message, not the first.  We show two enqueues, and one dequeue.. ***
DEBUG com.maxis.mxmove.core.SelectedDataJmsReader - Received: ActiveMQObjectMessage {commandId = 7, responseRequired = true, messageId = ID:zip-56502-1457640572818-4:2:2:1:2, originalDestination = null, originalTransactionId = null, producerId = ID:zip-56502-1457640572818-4:2:2:1, destination = queue://Stuff via SQL.402, transactionId = null, expiration = 0, timestamp = 1457640610375, arrival = 0, brokerInTime = 1457640610375, brokerOutTime = 1457640610376, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 1024, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false}
INFO  com.maxis.mxmove.core.SelectedDataJmsReader - executed read, found end-of-stream marker, returning null
DEBUG org.apache.activemq.broker.region.Queue - queue://Stuff via SQL.402, subscriptions=2, memory=0%, size=1, pending=0 toPageIn: 0, Inflight: 1, pagedInMessages.size 1, pagedInPendingDispatch.size 0, enqueueCount: 2, dequeueCount: 1, memUsage:1107

Answer 1

在接收器设置代码中，请注意beforeStep（）方法是用@beforeStep注释的。我认为这意味着接收器被设置了两次，并且可能进行了预取优化。这是经过验证的，因为日志显示了两个订阅。我不是一个笨重的JMS用户，我有一个错误的印象，一个是收件人，另一个是发件人。

@Override
@BeforeStep
public void beforeStep(StepExecution stepExecution) {
    this.stepExecution = stepExecution;
    this.setJobExecution(stepExecution.getJobExecution());

删除@BeforeStep注释后，日志仅显示一个订阅

DEBUG org.apache.activemq.broker.region.Queue - queue://Workorders via SQL.408, subscriptions=1, memory=0%, size=1, pending=0 toPageIn: 0, Inflight: 1, pagedInMessages.size 1, pagedInPendingDispatch.size 0, enqueueCount: 1370, dequeueCount: 1369, memUsage:1024

下次有人告诉我你使用注入时代码有多“干净”，我可能会考虑武功交换的应用