消费者端的Spring JMS 连接ActiveMQ接收生产者Oozie Server发送的Oozie作业执行结果

 <property>
        <name>oozie.jms.producer.connection.properties</name>
        <value>java.naming.factory.initial#org.apache.activemq.jndi.ActiveMQInitialContextFactory;java.naming.provider.url#tcp://l
ocalhost:61616;connectionFactoryNames#ConnectionFactory</value>
 </property>



<!-- JMSAccessorService -->

    <property>

        <name>oozie.service.JMSAccessorService.connectioncontext.impl</name>

        <value>

        org.apache.oozie.jms.DefaultConnectionContext

        </value>

        <description>

        Specifies the Connection Context implementation

        </description>

    </property>

  <property>
        <name>oozie.service.JMSTopicService.topic.name</name>
        <value>
        default=${username}
        </value>
        <description>
        Topic options are ${username} or ${jobId} or a fixed string which can be specified as default or for a
        particular job type.
        For e.g To have a fixed string topic for workflows, coordinators and bundles,
        specify in the following comma-separated format: {jobtype1}={some_string1}, {jobtype2}={some_string2}
        where job type can be WORKFLOW, COORDINATOR or BUNDLE.
        e.g. Following defines topic for workflow job, workflow action, coordinator job, coordinator action,
        bundle job and bundle action
        WORKFLOW=workflow,
        COORDINATOR=coordinator,
        BUNDLE=bundle
        For jobs with no defined topic, default topic will be ${username}
        </description>
    </property>

<bean id="jndiTemplate" class="org.springframework.jndi.JndiTemplate">
        <property name="environment">
            <props>
                <prop key="java.naming.factory.initial">
                    org.apache.activemq.jndi.ActiveMQInitialContextFactory
                </prop>
                <prop key="java.naming.provider.url">
                    tcp://192.168.121.35:61616
                </prop>
                <prop key="java.naming.security.principal">
                    system
                </prop>
                <prop key="java.naming.security.credentials">
                    manager
                </prop>
            </props>
        </property>
    </bean>

    <bean id="jndiTopicConnectionFactory" class="org.springframework.jndi.JndiObjectFactoryBean">
        <property name="jndiTemplate" ref="jndiTemplate" />
        <property name="jndiName" value="ConnectionFactory" />
    </bean>

//获得Oozie中关于JMS的相关配置信息,如Transport Connectors
        OozieClient oc = new OozieClient("http://192.168.121.35:11000/oozie");
        JMSConnectionInfo jmsInfo = oc.getJMSConnectionInfo();

        Properties jndiProperties = jmsInfo.getJNDIProperties();
        Context jndiContext = new InitialContext(jndiProperties);

{java.naming.provider.url=tcp://192.168.121.35:61616, 
java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory, 
connectionFactoryNames=ConnectionFactory}

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

<bean id="jmsContainer"
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="topicConnectionFactory"></property>
        <property name="destinationResolver" ref="destinationResolver"></property>
        <property name="concurrentConsumers" value="1"></property>
        <property name="destination" ref="notifyTopic"></property>
        <property name="messageListener" ref="messageListener"></property>
    </bean>

    <bean id="messageListener" class="com.schedule.tools.message.SimpleJMSReceiver" />

AUTO_ACKNOWLEDGE
Automatic message acknowledgment before listener execution; no redelivery in case of exception thrown.

CLIENT_ACKNOWLEDGE
Automatic message acknowledgment after successful listener execution; no redelivery in case of exception thrown.

DUPS_OK_ACKNOWLEDGE
Lazy message acknowledgment during or after listener execution; potential redelivery in case of exception thrown.

@Override
    public void onMessage(Message message) {
        String parentJobId = null;
        String jobId = null;
        String errorMessage = null;
        String status = null;
        Date startTime = null;
        Date endTime = null;
        long runTime = -1;//-1 means job run error
        
        try {
            // 普通用户作业和解释作业
            
            if (message.getStringProperty(JMSHeaderConstants.APP_TYPE).equals(
                    AppType.WORKFLOW_JOB.name())) {
                
                WorkflowJobMessage wfJobMessage = JMSMessagingUtils
                        .getEventMessage(message);

                // 是普通作业
                jobId = wfJobMessage.getId();
                errorMessage = wfJobMessage.getErrorMessage();
                status = wfJobMessage.getStatus().toString();
                startTime = wfJobMessage.getStartTime();
                endTime = wfJobMessage.getEndTime();
                
                
                if(endTime != null){
                    runTime = endTime.getTime() - startTime.getTime();
                    System.out.println(jobId + "执行了:" + (endTime.getTime()-startTime.getTime())/1000 + "s");                    
                }



//other code.....