Spring 3 HornetQ 2.1集成教程

通過Spring框架使用JBoss的新超高性能消息傳遞系統。

HornetQ是一個開放源代碼項目，用于構建多協議，可嵌入，非常高性能的集群異步消息傳遞系統。它是用Java編寫的，并且可以在具有Java 5或更高版本運行時的任何平臺上運行。 HornetQ一流的高性能日志以非持久消息傳遞通常看到的速度提供持久消息傳遞性能。非持久消息傳遞性能也非常高。 HornetQ除了其他“性感”功能外，還提供服務器復制和自動客戶端故障轉移功能，以消除服務器故障時丟失或重復的消息，可以配置成群集使用，其中HornetQ服務器的地理位置分散的群集知道如何負載均衡消息并提供全面的信息。管理API，用于管理和監視所有HornetQ服務器。

在本教程中，我們將向您展示如何通過Spring框架利用HornetQ 。為了使事情變得更有趣，我們將從上一篇關于Spring GWT Hibernate JPA Infinispan集成的文章的結尾處繼續。我們將使用我們的GWTSpringInfinispan項目，并通過消息傳遞功能對其進行授權！當然，您可以閱讀本文，將基于Spring的項目與HornetQ集成。

我們將使用HornetQ 2.1.0.Final版本，您可以從此處下載。我們還將需要jboss-logging-spi庫。將使用JBoss Logging SPI 2.1.1.GA版本，您可以在此處從JBoss Maven存儲庫下載該版本。

為了在運行時正確地集成Spring和HornetQ ，我們必須為Web應用程序提供所有必需的庫。因此，復制下面在/ war / WEB-INF / lib下列出的文件（如果使用的是不同版本，請復制相關文件）

從HornetQ發行

/lib/hornetq-bootstrap.jar
/lib/hornetq-core.jar
/lib/hornetq-jms.jar
/lib/hornetq-logging.jar
/lib/jnpserver.jar
/lib/netty.jar

JBoss Logging SPI庫

jboss-logging-spi-2.1.1.GA.jar

最后，為了使HornetQ在運行時正常運行，Web應用程序的類路徑中必須有幾個配置文件。如本教程的介紹部分所述，我們可以創建HornetQ服務器集群，以實現負載平衡和高可用性消息傳遞，也可以在非集群環境中使用HornetQ 。兩種情況都需要不同的配置。 HornetQ發行版包含/ config目錄下的所有配置文件。我們將使用jboss-as-5集群配置，以便能夠使用消息傳遞平臺的全部功能。將以下文件從/ config / jboss-as-5 / clustered目錄復制到應用程序/ resources包中：

hornetq-configuration.xml –這是主要的HornetQ配置文件
hornetq-jms.xml –服務器端JMS服務配置文件

除非您要在JBoss應用服務器中進行部署，否則請編輯hornetq-configuration.xml文件，并將“ $ {jboss.server.data.dir}”替換為“ $ {data.dir：../ data}”。

將以下文件從/ config / stand-alone / clustered目錄復制到應用程序/ resources包中：

hornetq-users.xml – HornetQ安全管理器的用戶憑證文件

在繼續實際的集成和客戶端實現示例之前，讓我們確定一些有關HornetQ服務器體系結構和上述配置文件的有用信息。

HornetQ服務器不會講JMS ，實際上對JMS一無所知，它是一種協議不可知的消息傳遞服務器，旨在與多種不同的協議一起使用。 HornetQ客戶端（可能在不同的物理計算機上）與HornetQ服務器交互。 HornetQ當前提供兩種API來在客戶端進行消息傳遞：

核心客戶端API。這是一個簡單而直觀的Java API，可在沒有JMS某些復雜性的情況下提供完整的消息傳遞功能集
JMS客戶端API。標準的JMS API在客戶端可用

JMS語義由客戶端上的瘦JMS外觀層實現。當用戶在客戶端上使用JMS API時，所有JMS交互都將轉換為HornetQ核心客戶端API上的操作，然后再使用HornetQ有線格式通過有線傳輸。服務器始終只處理核心API交互。

標準的獨立消息傳遞服務器配置包括核心消息傳遞服務器， JMS服務和JNDI服務。

JMS服務的作用是將任何服務器端hornetq-jms.xml配置文件中的任何JMS Queue，Topic和ConnectionFactory實例部署并綁定到JNDI 。它還提供了用于創建和銷毀隊列，主題和ConnectionFactory實例的簡單管理API，可以通過JMX或連接對其進行訪問。由于核心服務器與JMS無關，因此它是HornetQ核心服務器的一項獨立服務。如果您不想通過服務器端XML配置部署任何JMS Queue，Topic或ConnectionFactory實例，并且不需要在服務器端使用JMS管理API，則可以禁用此服務。

還包括一個JNDI服務器，因為在使用JMS查找隊列，主題和ConnectionFactory實例時， JNDI是常見的要求。如果您不需要JNDI，則也可以禁用此服務。 HornetQ允許您直接在客戶端上以編程方式創建JMS和核心對象，而不是從JNDI查找它們，因此JNDI服務器并不總是必需的。

HornetQ附帶了一個基本的安全管理器實現，該實現可獲取用戶憑證
從hornetq-users.xml文件中。該文件包含用戶，密碼和角色信息。

我們將使用HornetQ JMS服務，并在與命名服務器相同的JVM中執行JMS客戶端代碼，因此我們必須創建一個“ jndi.properties”文件，并將其與上述其余HornetQ配置文件一起放在我們的application / resources包下。 “ jndi.properties”文件的內容應如下所示：

java.naming.factory.initial=org.jnp.interfaces.NamingContextFactory

在繼續之前，我們必須注意Eclipse項目的依賴關系。以下jar應包含在項目的Java構建路徑中：

jms.jar

現在讓我們將Spring與HornetQ集成。找到您的applicationContext.xml文件/ war / WEB-INF文件夾，然后添加以下bean：

<bean name="namingServerImpl" class="org.jnp.server.NamingBeanImpl" init-method="start" destroy-method="stop" /><bean name="namingServer" class="org.jnp.server.Main" init-method="start" destroy-method="stop"><property name="namingInfo" ref="namingServerImpl" /><property name="port" value="1099" /><property name="bindAddress" value="localhost" /><property name="rmiPort" value="1098" /><property name="rmiBindAddress" value="localhost" />
</bean><bean name="mbeanServer" class="java.lang.management.ManagementFactory" factory-method="getPlatformMBeanServer" /><bean name="fileConfiguration" class="org.hornetq.core.config.impl.FileConfiguration" init-method="start" destroy-method="stop" /><bean name="hornetQSecurityManagerImpl" class="org.hornetq.spi.core.security.HornetQSecurityManagerImpl" /><!-- The core server -->
<bean name="hornetQServerImpl" class="org.hornetq.core.server.impl.HornetQServerImpl"><constructor-arg ref="fileConfiguration" /><constructor-arg ref="mbeanServer" /><constructor-arg ref="hornetQSecurityManagerImpl" />
</bean><!-- The JMS server -->
<bean name="jmsServerManagerImpl" class="org.hornetq.jms.server.impl.JMSServerManagerImpl" init-method="start" destroy-method="stop" depends-on="namingServer"><constructor-arg ref="hornetQServerImpl" />
</bean>

如果您打算在獨立環境中配置Spring和HornetQ ，則上述配置就足夠了。在我們的例子中，當我們在Apache – Tomcat上部署Web應用程序時，應進行一些小的修改。

Apache – Tomcat為所有已部署的Web應用程序提供JNDI服務，以配置環境屬性和資源。此外，由于環境和資源管理是使用部署描述符文件（例如web.xml和context.xml）完成的，因此在運行時可用的命名上下文是只讀的。另外，在啟動時， Apache – Tomcat使用系統屬性初始化其JNDI環境。結果，使用JNDI InitialContext類（不提供構造函數環境參數）的“在VM中”客戶端執行命名操作，總是檢索Apache – Tomcat JNDI實現Context接口。

為了使HornetQ JNDI服務器與Apache-Tomcat命名服務和HornetQ JMS服務共存，以便將隊列，主題和ConnectionFactory實例綁定到JNDI ，我們必須執行以下操作：

對我們的Web應用程序禁用Apache – Tomcat命名服務
將HornetQ JNDI服務器配置為不使用現有的JNDI服務（如果可用），但始終創建??一個新的服務

要為我們的Web應用程序禁用Apache – Tomcat命名服務，我們必須執行以下操作：

在我們項目的/ war文件夾下創建一個META-INF文件夾
創建一個包含以下context指令的context.xml文件：

<Context override="true" useNaming="false" />

要配置HornetQ JNDI服務器不使用現有的JNDI服務（如果可用），我們必須在Spring bean的“ namingServerImpl”中添加以下屬性：

<property name="useGlobalService" value="false" />

為了通過Spring使用HornetQ消息服務，我們可以創建一個連接工廠，也可以從JNDI查找一個。下面提供了連接工廠和“ JmsTemplate”示例：

<bean name="connectionFactory" class="org.hornetq.jms.client.HornetQConnectionFactory" ><constructor-arg><bean class="org.hornetq.api.core.TransportConfiguration"><constructor-arg value="org.hornetq.integration.transports.netty.NettyConnectorFactory" /><constructor-arg><map key-type="java.lang.String" value-type="java.lang.Object"><entry key="port" value="5445"></entry></map></constructor-arg></bean></constructor-arg>
</bean><bean name="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"><property name="connectionFactory" ref="connectionFactory"></property>
</bean>

連接工廠示例的JNDI查找如下所示：

<bean id="inVMConnectionFactory" class="org.springframework.jndi.JndiObjectFactoryBean" depends-on="jmsServerManagerImpl"><property name="jndiName"><value>java:/ConnectionFactory</value></property>
</bean>

我們將使用JNDI查找方法來獲取連接工廠，因此將上述配置添加到applicationContext.xml文件中。

這就是我們要做的所有配置，讓我們繼續使用我們新集成的消息傳遞服務實施假設的業務案例。我們的Web應用程序提供了添加，更新和檢索“員工”數據的功能。假設我們希望在每次添加或更改“員工”數據時都收到通知。為了簡單起見，該通知將是Apache – Tomcat控制臺上的日志。我們將實現一個JMS生產者，以便在用戶每次對“雇員”數據進行更新時將消息發送到“通知”隊列。另外，必須實現JMS使用者，以便處理“通知”隊列消息并登錄到控制臺。

要創建“通知”隊列并將其綁定到名稱為“ / queue / Notifications”的JNDI ，請將以下內容添加到hornetq-jms.xml文件中：

<queue name="Notifications"><entry name="/queue/Notifications"/>
</queue>

為了能夠通過Spring bean使用新創建的“ Notifications”隊列，請將以下JNDI查找指令添加到applicationContext.xml文件中：

<bean id="notificationsQueue" class="org.springframework.jndi.JndiObjectFactoryBean" depends-on="jmsServerManagerImpl"><property name="jndiName"><value>/queue/Notifications</value></property>
</bean>

由于JMS生產者和使用者都是服務器端組件，因此必須將它們放在我們應用程序的/ server子包下。我們選擇在/ server / utils子包下創建它們，因為它們本質上是實用程序類。下面提供了示例JMS生產者和消費者類：

package com.javacodegeeks.gwtspring.server.utils;import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;@Service("notificationsProducer")
public class NotificationsProducer {@AutowiredQueue notificationsQueue;@AutowiredConnectionFactory inVMConnectionFactory;private Connection notificationsQueueConnection;private Session notificationsQueueSession;private MessageProducer notificationsQueueProducer;@PostConstructpublic void init() throws Exception {notificationsQueueConnection = inVMConnectionFactory.createConnection();notificationsQueueSession = notificationsQueueConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);notificationsQueueProducer = notificationsQueueSession.createProducer(notificationsQueue);notificationsQueueProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);}@PreDestroypublic void destroy() throws Exception {if(notificationsQueueConnection != null)notificationsQueueConnection.close();}public void sendNotification(final String message) throws Exception {TextMessage textMessage = notificationsQueueSession.createTextMessage(message);notificationsQueueProducer.send(textMessage);}}

還有消費者

package com.javacodegeeks.gwtspring.server.utils;import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;@Service("notificationsConsumer")
public class NotificationsConsumer implements MessageListener {@AutowiredQueue notificationsQueue;@AutowiredConnectionFactory inVMConnectionFactory;private Connection notificationsQueueConnection;@PostConstructpublic void init() throws Exception {notificationsQueueConnection = inVMConnectionFactory.createConnection();Session notificationsQueueSession = notificationsQueueConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);MessageConsumer notificationsQueueConsumer = notificationsQueueSession.createConsumer(notificationsQueue);notificationsQueueConsumer.setMessageListener(this);notificationsQueueConnection.start();}@PreDestroypublic void destroy() throws Exception {if(notificationsQueueConnection != null)notificationsQueueConnection.close();}@Overridepublic void onMessage(Message message) {if (message instanceof TextMessage) {try {String text = ((TextMessage) message).getText();System.out.println("The Notification Message is : \n" + text);} catch (JMSException ex) {throw new RuntimeException(ex);}} else {throw new IllegalArgumentException("Message must be of type TextMessage");}}}

總結我們的示例業務案例，我們必須修改“ employeeService” Spring bean，以便在用戶每次請求保存或更新“ employee”數據時使用“ notificationsProducer”實用程序bean發送通知消息。我們使用“ @Autowire”注釋在“ employeeService”內部連接“ notificationProducer”，并從“ notificationProducer”調用“ sendNotification”操作，以便每次請求“ employeeService”的saveOrUpdateEmployee“操作時發送通知。完整的代碼如下所示：

package com.javacodegeeks.gwtspring.server.services;import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;import com.javacodegeeks.gwtspring.server.dao.EmployeeDAO;
import com.javacodegeeks.gwtspring.server.utils.NotificationsProducer;
import com.javacodegeeks.gwtspring.shared.dto.EmployeeDTO;
import com.javacodegeeks.gwtspring.shared.services.EmployeeService;@Service("employeeService")
public class EmployeeServiceImpl implements EmployeeService {@Autowiredprivate EmployeeDAO employeeDAO;@AutowiredNotificationsProducer notificationsProducer;@PostConstructpublic void init() throws Exception {}@PreDestroypublic void destroy() {}@Transactional(propagation=Propagation.SUPPORTS, rollbackFor=Exception.class)public EmployeeDTO findEmployee(long employeeId) {return employeeDAO.findById(employeeId);}@Transactional(propagation=Propagation.REQUIRED, rollbackFor=Exception.class)public void saveEmployee(long employeeId, String name, String surname, String jobDescription) throws Exception {EmployeeDTO employeeDTO = employeeDAO.findById(employeeId);if(employeeDTO == null) {employeeDTO = new EmployeeDTO(employeeId, name,surname, jobDescription);employeeDAO.persist(employeeDTO);}}@Transactional(propagation=Propagation.REQUIRED, rollbackFor=Exception.class)public void updateEmployee(long employeeId, String name, String surname, String jobDescription) throws Exception {EmployeeDTO employeeDTO = employeeDAO.findById(employeeId);if(employeeDTO != null) {employeeDTO.setEmployeeName(name);employeeDTO.setEmployeeSurname(surname);employeeDTO.setJob(jobDescription);}}@Transactional(propagation=Propagation.REQUIRED, rollbackFor=Exception.class)public void deleteEmployee(long employeeId) throws Exception {EmployeeDTO employeeDTO = employeeDAO.findById(employeeId);if(employeeDTO != null)employeeDAO.remove(employeeDTO);}@Transactional(propagation=Propagation.REQUIRED, rollbackFor=Exception.class)public void saveOrUpdateEmployee(long employeeId, String name, String surname, String jobDescription) throws Exception {EmployeeDTO employeeDTO = new EmployeeDTO(employeeId, name,surname, jobDescription);employeeDAO.merge(employeeDTO);notificationsProducer.sendNotification("Save Or Update Employee with values : \nID : " + employeeId + "\nName : " + name + "\nSurname : " + surname + "\nJob description : " + jobDescription);}}

而已！要部署Web應用程序，只需將/ war文件夾復制到Apache – Tomcat “ webapps”文件夾中。您可以將war文件夾的名稱更改為您喜歡的名稱，最好在項目名稱之后將其重命名，例如GWTSpringInfinispanHornetQ

在午餐之前，應用程序不要忘記創建數據庫模式，這里是“ javacodegeeks”。

午餐應用程序將您的瀏覽器指向以下地址

http：// localhost：8080 / GWTSpringInfinispanHornetQ /

如果一切順利，您應該會看到您的主頁。應該顯示兩個文本框，每個文本框后面都有一個按鈕。在第一個文本框中，您可以將雇員保存或更新到數據庫。作為輸入，提供ID，名稱，姓氏和職位描述，并用空格字符分隔。單擊“ SaveOrUpdate”按鈕，將提供的信息存儲到數據庫中。對于現有的“員工”條目（相同的ID），將執行更新。在這兩種情況下，都應記錄一個通知日志。日志格式應如下：

通知消息為：
使用值保存或更新員工：
編號：xxx 名稱：xxx 姓：xxx 職位描述：xxx

其中“ xxx”應為您提供的“員工”信息。請查看日志文件（catalina.out）。第二個文本框用于檢索現有的“雇員”條目。提供一個“雇員” ID，然后單擊“檢索”按鈕。如果“雇員”存在，則應看到“雇員” ID，姓名，姓氏和職位描述。

您可以從此處下載該項目（如開頭所述，并且不包含先前的文章，所需的第三方庫）

玩得開心！

賈斯汀

相關文章：