glassfish+mdb+activeMQ

1.在active的控制台，新建queue：batchAdjust

2.用脚本配置好glassfish的配置
脚本如下：

@echo off

if "%1" == "" goto :ERROR

echo prepare for mq
call %1\bin\asadmin delete-jmsdest  --desttype queue ActiveMqQueue
call %1\bin\asadmin delete-admin-object jms/ActiveMqQueue
call %1\bin\asadmin delete-connector-resource jms/ActiveMQConnectionFactory
call %1\bin\asadmin delete-connector-connection-pool jms/ActiveMQpool

call %1\bin\asadmin undeploy activemq-rar-5.1.0

echo finished prepare for mq


echo start config for mq
#发布activemq的rar，这个包一般在apache-activemq-5.1.0-bin\apache-activemq-5.1.0\lib\optional目录下
# activemq-rar-5.1.0.rar包中的activemq-rar-5.1.0.rar\META-INF\ra.xml可以改连接哪个地址的active MQ
call %1\bin\asadmin deploy activemq-rar-5.1.0.rar
echo finished deploying activemq-rar-5.1.0.rar

call %1\bin\asadmin create-admin-object --raname activemq-rar-5.1.0 --restype javax.jms.Queue --property DestinationJndiName=ActiveMqQueue jms/ActiveMqQueue
echo finished create admin object jms/ActiveMqQueue

call %1\bin\asadmin create-jmsdest --desttype queue ActiveMqQueue
echo finished create jmsdest ActiveMqQueue

call %1\bin\asadmin create-connector-connection-pool --raname activemq-rar-5.1.0 --connectiondefinition javax.jms.ConnectionFactory --transactionsupport  XATransaction jms/ActiveMQpool
echo finished create connector connection pool jms/ActiveMQpool

#建jms/ActiveMQConnectionFactory给客户端使用
call %1\bin\asadmin create-connector-resource --poolname jms/ActiveMQpool jms/ActiveMQConnectionFactory
echo finished create connector resource jms/ActiveMQConnectionFactory

echo Setup complete
goto :END


:ERROR
echo Please set glassfish base path
echo e.g. config-new-glassfish c:\glassfish

:END

2.然后用netbean6.1新增加mdb




在生成的MDB中，加入红色部分。

@MessageDriven(mappedName = "jms/batchAdjust", activationConfig =  {
        @ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),
        @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
        @ActivationConfigProperty(propertyName="destination", propertyValue="batchAdjust") 

    })

MDB完整代码

package com.outblaze.pointsystem.mdb;

import com.outblaze.pointsystem.model.HashMapKey;
import com.outblaze.pointsystem.model.MessageType;
import com.outblaze.pointsystem.pojo.UploadPoint;
import com.outblaze.pointsystem.session.pointadjust.LoyaltyPointAdjustLocal;
import com.outblaze.pointsystem.utils.AppLog;
import java.util.HashMap;
import java.util.Map;
import javax.annotation.Resource;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.EJB;
import javax.ejb.MessageDriven;
import javax.ejb.MessageDrivenContext;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.TextMessage;

/**
 * batch adjust all user balance
 * @author whiskey.yang
 */
@MessageDriven(mappedName = "jms/ActiveMqQueue", activationConfig ={ @ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),@ActivationConfigProperty(propertyName = "destination", propertyValue = MessageType.QUEUE_BALANCE_LOYALTY_ADJUST_WHOLE)})
public class QBalanceLoyaltyAdjustWholeMB implements MessageListener
{
    /**
     * Session Facade for LoyaltyPointAdjustLocal 
     */
    @EJB private LoyaltyPointAdjustLocal  loyaltyPointAdjustLocal ;

    @Resource
    private MessageDrivenContext mdc;

    /** QBalanceLoyaltyAdjustWholeMB */
    public QBalanceLoyaltyAdjustWholeMB()
    {
    }

    /**
     * onmessage
     * @param message Message
     */
    public void onMessage(Message message)
    {

        TextMessage textMessage;
        MapMessage msg;
        ObjectMessage objMsg = null;
        try
        {
            AppLog.getInstance().getEjbLog().info("MDB:"+MessageType.QUEUE_BALANCE_LOYALTY_ADJUST_WHOLE);
            AppLog.getInstance().getEjbLog().info("receive message---------------" + message.getJMSMessageID());
            if (message instanceof ObjectMessage)
            {
                objMsg = (ObjectMessage) message;
                Map map = (HashMap) objMsg.getObject();
                
                //logice method
                UploadPoint uploadPoint=(UploadPoint)map.get(HashMapKey.ADJUST_LOYALTY_ALL_USER_POJO);
                loyaltyPointAdjustLocal.updateAllLoyaltyPointBatch(uploadPoint);
            }
            else if (message instanceof TextMessage)
            {
                textMessage = (TextMessage) message;
                System.out.println("Message Received:" + textMessage.getText());
            }
        }
        catch (JMSException ex)
        {
            AppLog.getInstance().getEjbLog().error("Deal with message JMSException", ex);
            mdc.setRollbackOnly();
        }
        catch (Exception ex)
        {
            AppLog.getInstance().getEjbLog().error("Deal with message Exception", ex);
            mdc.setRollbackOnly();
        }
    }
}

POJO

package com.outblaze.pointsystem.model;

/**
 * the message for active MQ
 * @author whiskey.yang
 */
public class MessageType
{

    /**
     * the type for batch adjust all loyalty user balance
     */
    public static final String QUEUE_BALANCE_LOYALTY_ADJUST_WHOLE = "queueBalanceLoyaltyAdjustWhole";
    /**
     * the type for batch adjust specified loyalty user balance
     */
    public static final String QUEUE_BALANCE_LOYALTY_ADJUST_SPECIFIED = "queueBalanceLoyaltyAdjustSpecified";
}

＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝


客户端代码：
＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝
import com.outblaze.pointsystem.utils.AppLog;
import java.util.HashMap;
import javax.annotation.Resource;
import javax.ejb.Stateless;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

/**
 * The session for message client
 * @author whiskey.yang
 */
@Stateless
public class MessageClientBean implements MessageClientLocal
{

    @Resource(mappedName = "jms/ActiveMQConnectionFactory")
    private ConnectionFactory deployQueueFactory;

    /**
     * Send the message to deploy queue.
     * @param map HashMap
     * @param queueName the queue name
     * @return boolean if message is sent successfully return true.else return false
     */
    public boolean sendMessageToActiveMQ(HashMap map, String queueName)
    {
        boolean returnflag=false;
        Connection connect = null;
        Session session = null;
        MessageProducer messageProducer = null;
        Queue queue = null;
        try
        {
            connect = deployQueueFactory.createConnection();
            connect.start();
            session = connect.createSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);
            queue = session.createQueue(queueName);
            messageProducer = session.createProducer(queue);
            javax.jms.ObjectMessage objMessage = session.createObjectMessage();
            objMessage.setObject(map);
            messageProducer.send(objMessage);
            AppLog.getInstance().getEjbLog().info("success send message to queueName:" + queueName + ".............................................");
            returnflag=true;


//            javax.jms.ObjectMessage objMessage = session.createObjectMessage();
//            objMessage.setObject(map);
//            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); 
//                   Connection connection = factory.createConnection();     
//                   Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
//                   Topic topic = session.createTopic("TestTopic"); 
//                   MessageProducer publisher = session.createProducer(topic); 
//                   publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT); 
//               connection.start(); 
//               for(int x=0;x<5;x++){ 
//              TextMessage message = session.createTextMessage("Test message "+x); 
//              System.out.println("Sending message: "+message.getText()); 
//              publisher.send(message); 
        }
        catch (JMSException ex)
        {
            AppLog.getInstance().getEjbLog().error("send message to active MQ JMSException:" + ex);
        }
        catch(Exception ex)
        {
            AppLog.getInstance().getEjbLog().error("send message to active MQ Exception:" + ex);
        }
        finally
        {
            try
            {
                if (messageProducer != null)
                {
                    messageProducer.close();
                }
                if (session != null)
                {
                    session.close();
                }
                if(connect!=null)
                {
                    connect.close();
                }
                
            }
            catch (JMSException ex)
            {
                AppLog.getInstance().getEjbLog().error("Finished sending message to active MQ. But colse the resource exception:" + ex);
            }

        }
        return returnflag;
    }
}