AMQP上的ServiceBus关闭了消费者

我们有ServiceBus和Java JMS QPID客户端0.26的本地安装 . 似乎SB中存在错误 - 它在 生产环境 者关闭连接时向消费者发送END命令 . 生产环境 者和消费者都必须在同一台机器上运行 .

首先启动侦听队列的消费者：

static void consumeFromQueueForStackOverflow(Connection connection, Context context)
        throws JMSException, NamingException, InterruptedException {
    Session session = null;
    MessageConsumer consumer = null;
    long RECEIVE_TIMEOUT_MS = 30000;

    try {
        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

        Queue queue = (Queue) context.lookup("JBA_QUEUE");
        consumer = session.createConsumer(queue);

        connection.start();

        int consumed = 0;
        while (true) {
            long startMS = System.currentTimeMillis();
            Message message = consumer.receive(RECEIVE_TIMEOUT_MS);
            if (message != null) {
                consumed++;
                message.acknowledge();
                continue;
            }

            long durationMS = System.currentTimeMillis() - startMS;
            if (durationMS < RECEIVE_TIMEOUT_MS) {
                log.info(String.format(
                        "Connection is closed, timeout: %d[ms], waited: %s[ms] (consumed: %d)",
                        RECEIVE_TIMEOUT_MS, durationMS, consumed));

                break;
            }

            log.info(String.format("Receive timeout, retrying (consumed: %d)", consumed));
            consumed = 0;
        }
    } finally {
        connection.stop();

        if(null != consumer)
            consumer.close();
        if(null != session)
            session.close();
    }
}

然后将1条消息发送到队列：

static void publishToQueueForStackOverflow(Connection connection, Context context)
        throws JMSException, NamingException, InterruptedException {
    Session session = null;
    MessageProducer producer = null;

    try {
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        Queue queue = (Queue) context.lookup("JBA_QUEUE");

        producer = session.createProducer(queue);
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);

        TextMessage message = session.createTextMessage("My Message");
        producer.send(message);
    } finally {
        if(producer != null)
            producer.close();
        if(session != null)
            session.close();
    }

    log.info(String.format("Sent %d messages", count));
}

一旦 生产环境 者进程结束，消费者也会在等待时间小于超时的情况下结束 . receive（timeout）方法在超时到期之前返回null意味着（javadoc）“消息使用者同时关闭” . 或者，您可能会看到IllegalStateException：在调用acknowledge（）期间关闭 .

这是来自消费者的日志 . 您可以看到该服务器向客户端发送了意外的End {}：

RECV[5671|0] : Open{containerId=087d0b7b8a8e4809a686f8b20d5376f5_GPRGXIT002,maxFrameSize=65536,channelMax=255,idleTimeOut=240000}
SEND[5671|0] : null
SEND[5671|0] : Begin{nextOutgoingId=0,incomingWindow=2048,outgoingWindow=2048,handleMax=4294967295}
SEND[5671|0] : Attach{name=IntegrationServiceBus/jba_testing_queue-> (48ff030e-c7be-42b2-9c22-4d0db13aec8f),handle=0,role=receiver,sndSettleMode=unsettled,rcvSettleMode=first,source=Source{address=IntegrationServiceBus/jba_testing_queue,durable=none,expiryPolicy=link-detach},target=Target{}}
RECV[5671|0] : Begin{remoteChannel=0,nextOutgoingId=1,incomingWindow=2048,outgoingWindow=2048,handleMax=7}
RECV[5671|0] : Attach{name=IntegrationServiceBus/jba_testing_queue-> (48ff030e-c7be-42b2-9c22-4d0db13aec8f),handle=0,role=sender,sndSettleMode=unsettled,rcvSettleMode=first,source=Source{address=IntegrationServiceBus/jba_testing_queue,durable=none,expiryPolicy=link-detach},target=Target{},initialDeliveryCount=0,maxMessageSize=262144,properties={com.microsoft:tracking-id=087d0b7b8a8e4809a686f8b20d5376f5_GPRGXIT002_BPRGXIT003;2411:54:55}}
SEND[5671|0] : Flow{nextIncomingId=1,incomingWindow=2048,nextOutgoingId=0,outgoingWindow=2048,handle=0,deliveryCount=0,linkCredit=100,drain=false,echo=false}
RECV[5671|0] : Transfer{handle=0,deliveryId=0,deliveryTag=\x84\xb8.\xf5\xda3\xafF\x89<J\x1bj\xda{<,messageFormat=0,more=false,batchable=true}
RECV[5671|0] : End{}
SEND[5671|0] : Detach{handle=0}
SEND[5671|0] : Disposition{role=receiver,first=0,last=0,settled=true,state=Released{}}
SEND[5671|0] : End{}
SEND[5671|0] : End{}
SEND[5671|0] : Close{}
RECV[5671|0] : Close{}