WSO2 ESB + activeMQ答案

【问题标题】：WSO2 ESB + activeMQWSO2 ESB + activeMQ
【发布时间】：2016-01-12 19:46:22
【问题描述】：

我是 wso2 esb 和 jms 的新手。我从soapUI 向wso2 esb 发送了一些消息。在我的 wso 序列中，一条已处理的消息会发送到 jms。是否有可能从 wso2 esb 设置此消息的“生存时间”？还是其他方式？

在 AMQ 中我添加了这个：

<policyEntry queue="myQueue">
  <deadLetterStrategy>
    <individualDeadLetterStrategy
      queuePrefix="DLQ." useQueueForQueueMessages="true" />
  </deadLetterStrategy>

类似的东西 <property name="JMSExpiration" value="today+hour_long_value" scope="transport" type="STRING"></property> 按顺序没有效果。

【问题讨论】：

在soapUI中？你见过jms的标题吗？如果没有，值得一试。
我尝试在 wso2 esb 中使用 header Mediator。
但没有任何效果。在 amq 中收到的消息仍然具有默认优先级 4。

标签： wso2 activemq esb ttl amq

【解决方案1】：

如果您使用 JMS 消息存储，您只需设置属性 JMS_PROD_TIME_TO_LIVE。

<property name="JMS_PROD_TIME_TO_LIVE" value="15000" />
<store messageStore="my_jms_message_store" />

使用 WSO2 ESB 4.9.0 测试（使用突触版本 2.1.3-wso2v11）

您可以在JmsProducer code找到更多信息

设置消息优先级的方法相同（属性 JMS_PROD_PRIORITY）。

【讨论】：

【解决方案2】：

我发现唯一可行的方法是创建自己的 Mediator，它将设置消息的生存时间并将其发送到队列。队列名按顺序预设，然后调用mediator：

<property xmlns="http://ws.apache.org/ns/synapse" name="qname" value="your_queue_name" scope="default" type="STRING"></property>

<class xmlns="http://ws.apache.org/ns/synapse" name="com.example.JMSMessageTimeToLiveMediator"></class>

中介类：

public class JMSMessageTimeToLiveMediator extends AbstractMediator implements
    ManagedLifecycle {

private static String CON_FACTORY_NAME = "QueueConnectionFactory";
private static String DEF_PROP_QNAME = "qname";
private static long TIME_TO_LIVE = 60000;

private static QueueConnectionFactory cf;

public boolean mediate(MessageContext context) {
    Connection connection = null;
    Session session = null;
    try {
        connection = cf.createQueueConnection();
        connection.start();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        String queueName = (String) context.getProperty(DEF_PROP_QNAME);
        Destination destination = session.createQueue(queueName);

        MessageProducer producer = session.createProducer(destination);
        producer.setTimeToLive(TIME_TO_LIVE);

        TextMessage message = session.createTextMessage(context
                .getEnvelope().toString());

        producer.send(message);
    } catch (JMSException e) {
        log.error("ProduceJMS ERROR: " + e.getClass() + "   "
                + e.getMessage());
    } catch (Exception e) {
        log.error("ProduceJMS ERROR: " + e.getClass() + "   "
                + e.getMessage());
    } finally {
        try {
            session.close();
            connection.close();
        } catch (JMSException e) {
            log.error("ProduceJMS ERROR: " + e.getMessage());
        }
    }

    return true;
}

public void init(SynapseEnvironment emvironment) {
    Hashtable<String, Object> environment = new Hashtable<String, Object>();
    environment.put("java.naming.factory.initial",
            "org.apache.activemq.jndi.ActiveMQInitialContextFactory");
    log.debug("ProduceJMS INIT");
    try {
        InitialContext ic = new InitialContext(environment);
        cf = (QueueConnectionFactory) ic.lookup(CON_FACTORY_NAME);
    } catch (NamingException e) {
        log.error("ProduceJMS INIT ERROR: " + e.getMessage());
    }
}

public void destroy() {
}

}

【讨论】：