如何使用 Grails 和 JMS / ActiveMQ 配置独占消费者？

Question

我有一个使用 JMS 插件订阅给定 ActiveMQ 主题的 Grails 应用程序。如何使 TestService 类成为独占消费者？ Details of exclusive consumer here

用例是我在 AWS EC2 上运行消费者，并且 ActiveMQ 提要的持续时间为 5 分钟，如果实例死亡，替换实例需要比这更长的时间。我不能丢失消息并且必须保留消息顺序，因此我希望使用多个实例，其中第一个连接实例将是代理发送每条消息的实例，而其他实例则保留。如果第一个实例死亡，AMQ 代理会将消息发送到其他实例之一。

此外，JMS 使用什么标准来确定独家消费者何时死亡或消失？

// resources.groovy
beans = {
    jmsConnectionFactory(org.apache.activemq.ActiveMQConnectionFactory) { 
    brokerURL top://example.com:1234 
    userName = 'user' 
    password = 'password' 
    } 
} 

class TestService { 

static exposes = ["jms"] 
static destination = "SOME_TOPIC_NAME" 
static isTopic = true 

def onMessage(msg) { 
    // handle message 

    // explicitly return null to prevent unwanted replyTo attempt 
    return null 
} 

} 

Answer 1

首先，您的示例使用了主题，这是行不通的；你想要队列：

class TestService { 
  static expose = ["jms"] 
  static destination = "MYQUEUE" 
  ...
}

在 ActiveMQ 中配置独占消费者很简单：

queue = new ActiveMQQueue("MYQUEUE?consumer.exclusive=true");

..但是使用 Grails 插件可能会很棘手；你可以试试这些：

class TestService { 
  static expose = ["jms"] 
  static destination = "MYQUEUE?consumer.exclusive=true" 
  def onMessage(msg){ ...}
}


class TestService { 
  static expose = ["jms"] 
  @Queue(
     name = "MYQUEUE?consumer.exclusive=true" 
  )
  def handleMessage(msg){ ...}
}

关于代理如何确定消费者是否死亡的问题，我不确定它在 ActiveMQ 中是如何完成的，但在大多数 JMS 实现中，TCP 失败会触发连接上的异常；对等点（在本例中为代理）处理异常并故障转移到下一个可用的消费者。

希望有帮助