JBOSS5.1 HA-JMS 集群的JMS服务

 集群的JMS服务
         JBoss AS 3.2.4 和更高的版本支持高可用性的 all 服务器配置里的 JMS（HA-JMS）服务。在 JBoss AS 当前的发布版本里，HA-JMS 服务用一个群集的 singleton fail-over 服务来实现。
注意
       如果你希望自己来配置 HA-JMS，你可以在以前的 JBoss AS 版本里让它运行。我们有一个客户成功地在 JBoss AS 3.0.7 里使用 HA-JMS。如果你有更多问题的话，请联系 JBoss 支持部门。

 高可用性的Singleton Fail-over
             JBoss HA-JMS 服务（也就是消息队列和主题）任何给定时间都只在群集里的单节点运行（主节点）。如果这个节点崩溃了，群集系统简单地选取另外一节点来运行 JMS 服务（fail-over）。这种设置提供了针对服务器瘫痪的冗余措施，但并没有降低 JMS 服务器节点的负载。
注意
          虽然你不能够对 HA-JMS 队列进行负载平衡（只有一个运行这个队列的主节点），但是你可以平衡处理队列里信息的 MDBs 的负载(参考章节 6.1.3, “负载均衡的HA-JMS MDBs”).

服务器端配置
            为了使用 singleton fail-over HA-JMS 服务，你必须在群集里的所有节点上配置相同的 JMS 服务。这包括所有和 JMS 相关的 MBeans 以及部署的 JMS 应用程序。
JMS 服务器被设置成在 DefaultDS 里持久化它的数据。在缺省的情况下，那就是内含的 HSQLDB。然而，在大部分群集环境里，所有节点都需要把数据持久化到一个共享数据库里。因此，在你启动群集 JMS 之前要做的第一件事就是建立一个共享数据库。你需要做如下的事情：
          • 配置DefaultDS来指向你选择的数据库服务器。就是用docs/examples/jca 目录下的xxx-ds.xml文件代替deploy/hsqlsb-ds.xml文件，其中xxx是目标共享数据库的名字（例如，mysql-ds.xml ）。
          • 把server/all/deploy-hasingleton/jms 目录下的hsqldb-jdbc2-service.xml文件用特定数据库的文件代替。例如，MySQL 的文件就是mysql-jdbc2-service.xml。JBoss AS 发行版本捆绑了一些 RDBMS 的配置文件。它们可以在docs/examples/jms下找到。
注意
        你不需要替换server/all/deploy-hasingleton/jms 目录下的hsqldb-jdbc-state-service.xml 文件。尽管它的名字里包含 hsql，它适用于所有兼容 SQL92 的数据库，包括 HSQL，MySQL，SQL Server 和更多数据库。象我们上面配置的那样，它自动使用 DefaultDS 来存放数据。 

HA-JMS客户端
客户和常规的 JMS 客户在两个方面有所不同。.
         • The HA-JMS 客户必须获得 HA-JNDI 里的 JMS connection factories（缺省端口是1100）。
           • 客户连接必须监控服务器异常（exceptions）。当群集系统 fail-over 到另外一个主节点时，所有在当前连接上的的客户端操作都会失败并产生异常（exceptions）。客户端必须知道该重新连接。
注意
           虽然 HA-JMS 连接工厂（connection factory）知道运行 JMS 服务的当前主节点，但并不存在智能的客户端拦截器（client side interceptor）。客户端 stub 只知道固定的主节点，它不能够依服务器拓扑结构的变化而调整。

负载平衡的HA-JMS MDBs
            虽然 HA-JMS 队列（queues）和主题（topics）在同一时间只在单节点上运行，但其他节点上的 MDBs 也能够接收和处理 HA-JMS 主节点上的信息。这种竞争的情况导致 MDBs 的平衡负载行为。为了启用 MDBs 的平衡负载，你可以指定队列的 receiver。这个 receiver 记录哪个节点正在等待信息和信息该按什么样的顺序来处理。JBoss 提供三个 receiver 的实现（implementations）。
• The org.jboss.mq.server.ReceiversImpl是 HashSet 的缺省实现（implementation）。
• The org.jboss.mq.server.ReceiversImplArrayList 是 ArrayList 的实现。
• The org.jboss.mq.server.ReceiversImplLinkedList 是 LinkedList 的实现。
               你可以指定 receiver 实现的类名作为在每个节点上定义永久 JMS Queue 或 DestinationManager 的 MBean 的一个属性。为了获得最好的负载平衡性能，我们建议你使用 ReceiversImplArrayList 或 ReceiversImplArrayList 实现，因为 JVM 里的 HashSet 的实现不是很好。

 

 

在JBOSS5.1的配置

分别在两台机器上配置JMS的ConnectionFactory和Queue、Topic。

具体配置如下：
 目录位置：%JBOSS_HOME%/server/%INSTANCE%/deploy/配置消息持久化需要的数据库
 如：hsqldb-ds.xml
 
  
   目录位置：%JBOSS_HOME%/server/%INSTANCE%/deploy/messaging

   其中：connection-factories-service.xml为配置JMS的ConnectionFactory的service的配置文件。其中
   ConnectionFactory为普通的ConectionFactory,不支持自动failover机制和负载均衡，使用于一般的应用项目中ConnectionFactory.
   ClusteredConnectionFactory为JBOSS提供的集群ConnectionFactory支持自动failover机制和负载均衡的创建。  但是对消息驱动bean不支持（MDB），如果需要启动failover和负载均衡机制需要设置
 如下属性为true:
      <attribute name="SupportsFailover">true</attribute>
      <attribute name="SupportsLoadBalancing">true</attribute>   
   ClusterPullConnectionFactory为JBOSS 提供的非ConnectionFactory 的JNDI绑定，只用于在集群中创建ConnectionFactory将消息。
  从一个节点转发到其他的节点上。
   MyExampleConnectionFactory为JBOSS提供的创建ConnectionFactory使用所有属性的实例，用于用户可以参考。

 

hsqldb-persistence-service.xml为普通配置JMS的消息持久化配置文件。这里这样持久化到数据库（db2,hsqldb,mysql,oracle,postgresql等）
具体可以参考%JBOSS_HOME%\docs\examples\jms\相关的配置文件

clustered-hsqldb-persistence-service.xml为配置JMS集群情况下消息持久化配置文件（本人直接使用的%JBOSS_HOME%\docs\examples\jms\下面的copy的），但是官方不推荐采用hsqldb做为消息持久化数据库，可能是事务支持不是很好。

destinations-service.xml为配置JMS的Queue/Topic的service的配置文件。
  其中配置两个默认的队列如：DLQ，ExpiryQueue
 
messaging-service.xml为JBOSS 消息服务部署描述符文件，需要配置默认的消息队列消息。

客户端配置：
 在JBOSS5.1 HA-JMS的使用中 请求的服务器路径配置如下：
        Properties prop = new Properties();
        prop.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");
        prop.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");
        prop.setProperty("java.naming.provider.url", "192.168.70.104:1099,192.168.45.146:1099");

 

 

HA-JMS客户端代码如下：

import java.util.Properties;

import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import javax.naming.Context;

/**
<p> 
</p>
 *  
 * <p>功能描述,该部分必须以中文句号结尾。<p>
 *
 * 创建日期 2013-5-28<br>
 * @author  longgangbai<br>
 * @version $Revision$ $Date$
 * @since   3.0.0
 */
public class CLusterClient {

    /**
     * @param args
     */
    public static void main(String[] args) {
        try {
            //创建JNDI上下文对象
            Context context = getInitialContext();
            //创建队列连接工厂对象
            QueueConnectionFactory obj = (QueueConnectionFactory) context.lookup("ClusteredConnectionFactory");
          //创建队列连接对象
            QueueConnection conn = obj.createQueueConnection();
            //创建队列发送者对象
            QueueSession session = conn.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
            //获取队列
            Queue queue = (Queue) context.lookup("queue/DLQ");
            //
            String text = "hello world";
            //创建消息发送者 message
            QueueSender sender = session.createSender(queue);
            for (int i = 0; i < 100; i++) {
                TextMessage tm = session.createTextMessage(text);
                System.out.println("send text:" + text);
                sender.send(tm);
            }
            //关闭连接对象
            conn.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("-------------OK");
    }

    public static Context getInitialContext() throws Exception {
        try {
            Properties prop = new Properties();
            prop.setProperty("java.naming.factory.initial", "org.jnp.interfaces.NamingContextFactory");
            prop.setProperty("java.naming.factory.url.pkgs", "org.jboss.naming:org.jnp.interfaces");
            prop.setProperty("java.naming.provider.url", "192.168.70.104:1100,192.168.45.146:1100");
            return new javax.naming.InitialContext(prop);
        } catch (Exception e) {
            e.printStackTrace();
            throw e;
        }
    }
}