WSO2 2.2.0 Message Broker 中的消息重试和死信队列答案

【问题标题】：Message retry and Dead Letter Queue in WSO2 2.2.0 Message BrokerWSO2 2.2.0 Message Broker 中的消息重试和死信队列
【发布时间】：2015-08-15 02:24:05
【问题描述】：

我们正在评估 WSO2 堆栈，尤其是 Message Broker v 2.2.0，但无法使消息重试限制起作用。

根据此文档页面，一旦客户端拒绝了一条消息 10 次，它将从队列中删除并放入死信队列。

https://docs.wso2.com/display/MB220/Maximum+Delivery+Attempts

拒绝的定义是：
a) 在使用 Session.CLIENT_ACKNOWLEDGE 或
的情况下不发送确认 b) 在使用事务会话的情况下回滚事务。

使用 WSO2 示例客户端代码，我们无法通过客户端确认模式或引发故障的任何组合来观察此行为。消息在队列中保持活动状态，并且可以从队列中获取任意次数。如您所料，确认或提交会话会将其从队列中删除。

谁能确认这个功能是否真的有效，如果有效，请告诉我们客户必须做什么才能触发它。我们一直在使用 WSO2 提供的示例客户端代码和未经修改的开箱即用服务器配置进行测试：

https://docs.wso2.com/display/MB220/Sending+and+Receiving+Messages+Using+Queues

我们将不胜感激，因为如果不了解系统的这方面的工作原理，我们将无法继续使用 WSO2。

【问题讨论】：

标签： wso2 message broker

【解决方案1】：

此功能按预期工作。为了测试您需要对示例代码中提供的接收器客户端进行一些修改。

添加给定的系统属性
将 abknlowdgment 模式更改为 CLIENT_ACK
获取消息10次不发送ACK给服务器

通过这些更改，您可以满足您的要求。

这里我把修改后的方法贴在QueueReceiver类中

public void receiveMessages() throws NamingException, JMSException {
    Properties properties = new Properties();
    System.setProperty("AndesAckWaitTimeOut", "30000");
    properties.put(Context.INITIAL_CONTEXT_FACTORY, QPID_ICF);
    properties.put(CF_NAME_PREFIX + CF_NAME, getTCPConnectionURL(userName, password));
    System.out.println("getTCPConnectionURL(userName,password) = " + getTCPConnectionURL(userName, password));
    InitialContext ctx = new InitialContext(properties);
    // Lookup connection factory
    QueueConnectionFactory connFactory = (QueueConnectionFactory) ctx.lookup(CF_NAME);
    QueueConnection queueConnection = connFactory.createQueueConnection();
    queueConnection.start();
    QueueSession queueSession =
            queueConnection.createQueueSession(false, QueueSession.CLIENT_ACKNOWLEDGE);
    //Receive message
    Queue queue =  queueSession.createQueue(queueName);
    MessageConsumer queueReceiver = queueSession.createConsumer(queue);
    int count =0;
    while (count < 12) {
        TextMessage message = (TextMessage) queueReceiver.receive();
        System.out.println("Got message ==>" + message.getText());
        count++;
    }
    queueReceiver.close();
    queueSession.close();
    queueConnection.stop();
    queueConnection.close();
}

请注意，此修改只是为了证明该功能正在运行。

【讨论】：