如何在兩個應用程序之間發佈/訂閱JMS消息？

Question

我有兩個Java獨立應用程序。我想從一個應用程序發送消息，並通過兩個客戶端異步接收消息：一個與發送者位於同一個應用程序中。另一個是在不同的應用程序。兩者都與ActiveMQ代理連接。但我只能看到第一個客戶端收到了消息，而另一個客戶卻沒有收到消息。通過JMS連接兩個應用程序的一般方法是什麼？我想我必須對JMS有一些不清楚的概念。我查了一下，但無法弄清楚如何設置我的Spring bean配置文件來在兩個Java應用程序之間發佈/訂閱消息。

這裏是第一個應用程序我的發件人的bean配置文件，也是第一個聽衆的豆這是在同一個應用程序：

<bean id="customerMessageSender" class="com.example.message.CustomerStatusSender"> 
    <property name="jmsTemplate" ref="jsmTemplateBean" /> 
    <property name="topic" ref="topicBean" /> 
</bean> 

<bean id="jsmTemplateBean" class="org.springframework.jms.core.JmsTemplate"> 
    <property name="connectionFactory" ref="connectionFactoryBean"/> 
    <property name="pubSubDomain" value="true"/> 
</bean> 

<bean id="topicBean" class="org.apache.activemq.command.ActiveMQTopic"> 
    <constructor-arg value="CustomerStatusTopic" /> 
</bean> 

<bean id="connectionFactoryBean" class="org.apache.activemq.ActiveMQConnectionFactory"> 
    <property name="brokerURL" value="tcp://localhost:61616" /> 
</bean> 

<bean id="customerStatusListener" class="com.example.message.CustomerStatusListener" /> 


<bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> 
    <property name="connectionFactory" ref="connectionFactoryBean" /> 
    <property name="destination" ref="topicBean" /> 
    <property name="messageListener" ref="customerStatusListener" /> 
</bean> 

下面是第二個偵聽器bean的配置文件，該文件是在不同的應用：

<bean id="topicBean" class="org.apache.activemq.command.ActiveMQTopic"> 
    <constructor-arg value="CustomerStatusTopic" /> 
</bean> 

<bean id="connectionFactoryBean" class="org.apache.activemq.ActiveMQConnectionFactory"> 
    <property name="brokerURL" value="tcp://localhost:61616" /> 
</bean> 

<bean id="anotherCustomerStatusListener" class="com.mydomain.jms.CustomerStatusMessageListener" /> 

<bean id="listenerContainer" 
    class="org.springframework.jms.listener.DefaultMessageListenerContainer"> 
    <property name="connectionFactory" ref="connectionFactoryBean" /> 
    <property name="destination" ref="topicBean" /> 
    <property name="messageListener" ref="anotherCustomerStatusListener" /> 
</bean> 

正如你所看到的，customerStatusListener豆和anotherCustomerStatusListener豆訂閱topicBean。但是，只有第一個偵聽器才能獲取該郵件，因爲它與發件人位於同一個應用程序中，而第二個偵聽器則沒有。通過JMS連接兩個Java應用程序的一般原則是什麼？這樣可以在兩個獨立的應用程序之間發送/接收消息？

編輯：由於類CustomerStatusMessageListener在不同的應用程序中，所以在第一個XML文件中無法添加以下監聽器Bean，因此在第一個（發件人）應用程序的類路徑中不可見。

<bean id="anotherCustomerStatusListener" class="com.mydomain.jms.CustomerStatusMessageListener" /> 

再次編輯：以下是在第二應用程序，它是從第一應用單獨的第二監聽器。它包含一個main方法來實例化偵聽器bean（jms-beans.xml是上面列出的第二個偵聽器的bean配置文件）。

public class CustomerStatusMessageListener implements MessageListener { 
    public void onMessage(Message message) { 
     if (message instanceof TextMessage) { 
      try { 
       System.out.println("Subscriber 2 got you! The message is: " 
         + ((TextMessage) message).getText()); 
      } catch (JMSException ex) { 
       throw new RuntimeException(ex); 
      } 
     } else { 
      throw new IllegalArgumentException(
        "Message must be of type TextMessage"); 
     } 
    } 

    public static void main(String[] args) { 
     ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("jms-beans.xml"); 
     CustomerStatusMessageListener messageListener = (CustomerStatusMessageListener) context.getBean("anotherCustomerStatusListener"); 
     context.close(); 
    } 
} 

Answer 1

您對JMS的理解是正確的。如果你想讓兩個聽衆接收同樣的信息，那麼一個話題就是實現它的方法。如果一個監聽者與發送者在同一個虛擬機上運行而另一個不在，那應該沒有關係。沒有看到你的代碼，你的Spring配置看起來也是正確的。這還剩下幾件事要檢查：

• 這兩個偵聽器是否在同一主機上運行？也就是說，localhost一個地方是一個聽衆，另一個地方是另一個聽衆？
• 您的第二個偵聽器在郵件發送時運行嗎？如果您的第二個偵聽器在發送郵件時未處於活動狀態，則不會查看它是否稍後啓動，除非您使用的持久主題和您的訂閱者至少已連接到代理。

根據您的意見，第二項是您遇到問題的地方。

此blog post說明如何設置持久主題（如果需要消息通過代理重新啓動持續存在，則持久消息）。基本上，這個配置添加到您的消息監聽器：

<property name="subscriptionDurable" value="true"> 
<property name="clientId" value="Some_unique_id"> 
<property name="durableSubscriptionName" value="Some_unique_id"> 

委託人身份證和持久訂閱名必須爲每個用戶不同的，所以你的第一個聽衆會碰到這樣的：

<bean id="listenerContainer" 
class="org.springframework.jms.listener.DefaultMessageListenerContainer"> 
     <property name="connectionFactory" ref="connectionFactoryBean" /> 
     <property name="destination" ref="topicBean" /> 
     <property name="messageListener" ref="anotherCustomerStatusListener" /> 
     <property name="subscriptionDurable" value="true"> 
     <property name="clientId" value="listener1"> 
     <property name="durableSubscriptionName" value="listener1"> 
</bean> 

其次應有：

 <bean id="listenerContainer" 
      class="org.springframework.jms.listener.DefaultMessageListenerContainer"> 
      <property name="connectionFactory" ref="connectionFactoryBean" /> 
      <property name="destination" ref="topicBean" /> 
      <property name="messageListener" ref="anotherCustomerStatusListener" /> 
      <property name="subscriptionDurable" value="true"> 
      <property name="clientId" value="listener2"> 
      <property name="durableSubscriptionName" value="listener2"> 
     </bean> 

請注意，您必須開始你的第二個監聽器至少一次註冊itsel f與代理商聯繫，以便代理人知道其clientId併爲其存儲消息，但您可以關閉它並稍後啓動它，以便在發生任何錯誤消息時將其關閉。

如果您的聽衆長時間在高容量系統上停留，代理將爲其存儲所有消息，最終可能會填滿磁盤或減慢代理。請參閱ActiveMQ documentation for automatically removing durable messages。