myjmsofjms.java

一个基于JBOSS3.26的JMS（queue方式）代码实例

package myjmsproject;

import java.io.Serializable;
import java.util.Hashtable;

import javax.jms.*;
import javax.naming.*;

/**
 <p>Title: messager</p>

 <p>Description: </p>

 <p>Copyright: Copyright (c) 2004</p>

 <p>Company: changwei</p>

 @author 吴昌海
 @version 1.0


 * Use this class to send and receive point-to-point messages.
 * To send a text message:
 * <code>
 * myJMSofJMS myJMSofJMS = new myJMSofJMS();
 * myJMSofJMS.sendText("Hello world");
 * myJMSofJMS.close(); //Release resources
 * </code>

 * To receive a message:
 * <code>
 * myJMSofJMS myJMSofJMS = new myJMSofJMS();
 * myJMSofJMS.getQueueReceiver();
 * </code>
 */
public class myJMSofJMS implements MessageListener {
    public myJMSofJMS() {
        try {
            jbInit();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    private static Context context;
    private boolean transacted = true;
    private int acknowledgementMode = javax.jms.Session.AUTO_ACKNOWLEDGE;
    private QueueConnectionFactory queueConnectionFactory;
    private QueueConnection queueConnection;
    private QueueSession queueSession;
    private QueueSender queueSender;
    private QueueReceiver queueReceiver;
    private Queue queue;
    private String queueConnectionFactoryName = "java://JmsXA";
    private String sendQueueName = "queue/DLQ";
    private String recvQueueName = "queue/DLQ";
    private String strContent=null;

    public void setTransacted(boolean transacted) {
        this.transacted = transacted;
    }

    public void sendObject(Serializable message) throws Exception {
        javax.jms.ObjectMessage objectMessage = getQueueSession().
                                                createObjectMessage();
        objectMessage.clearBody();
        objectMessage.setObject(message);
        getQueueSender().send(objectMessage);
        if (isTransacted()) {
            getQueueSession().commit();
        }
    }

    public void sendText(String message) throws Exception {
          System.out.println("sendText: "+queue.getQueueName());
        javax.jms.TextMessage textMessage = getQueueSession().createTextMessage();

        textMessage.clearBody();
        textMessage.setText(message);
        getQueueSender().send(textMessage);

        if (isTransacted()) {
            getQueueSession().commit();
        }
    }

    public QueueReceiver getQueueReceiver() throws Exception {
        if (queueReceiver == null) {
            queueReceiver = getQueueSession().createReceiver(getRecvQueue());
            queueReceiver.setMessageListener(this);
        }
        return queueReceiver;
    }

    public QueueSender getQueueSender() throws Exception {
        if (queueSender == null) {
            queueSender = getQueueSession().createSender(getSendQueue());
        }
        return queueSender;
    }

    public Queue getRecvQueue() throws Exception {
        if (queue == null) {
            Object obj = getContext().lookup(recvQueueName);
            queue = (Queue) obj;
        }
        return queue;
    }

    public Queue getSendQueue() throws Exception {
        if (queue == null) {
            Object obj = getContext().lookup(sendQueueName);
            queue = (Queue) obj;
        }
        return queue;
    }

    public QueueSession getQueueSession() throws Exception {
        if (queueSession == null) {
            queueSession = getQueueConnection().createQueueSession(isTransacted(),
                    getAcknowledgementMode());
        }
        return queueSession;
    }

    public QueueConnection getQueueConnection() throws Exception {
        if (queueConnection == null) {
            queueConnection = getQueueConnectionFactory().createQueueConnection();
            queueConnection.start();
        }
        return queueConnection;
    }

    public QueueConnectionFactory getQueueConnectionFactory() throws Exception {
        if (queueConnectionFactory == null) {
            Object obj = getContext().lookup(queueConnectionFactoryName);
            queueConnectionFactory = (QueueConnectionFactory) obj;
        }
        return queueConnectionFactory;
    }

    public void setRecvQueueName(String recvQueueName) {
        this.recvQueueName = recvQueueName;
    }

    public String getRecvQueueName() {
        return recvQueueName;
    }

    public void setSendQueueName(String sendQueueName) {
        this.sendQueueName = sendQueueName;
    }

    public String getSendQueueName() {
        return sendQueueName;
    }

    public void setQueueConnectionFactoryName(String queueConnectionFactoryName) {
        this.queueConnectionFactoryName = queueConnectionFactoryName;
    }

    public String getQueueConnectionFactoryName() {
        return queueConnectionFactoryName;
    }

    public void setAcknowledgementMode(int acknowledgementMode) {
        this.acknowledgementMode = acknowledgementMode;
    }

    public int getAcknowledgementMode() {
        return acknowledgementMode;
    }

    public boolean isTransacted() {
        return transacted;
    }

    private Context getInitialContext() throws NamingException {
        Hashtable environment = new Hashtable();

        environment.put(Context.INITIAL_CONTEXT_FACTORY,
                        "org.jnp.interfaces.NamingContextFactory");
        environment.put(Context.URL_PKG_PREFIXES,
                        "org.jboss.naming:org.jnp.interfaces");
        environment.put(Context.PROVIDER_URL, "jnp://10.1.56.72:1099");

        return new InitialContext(environment);
    }

    private Context getContext() throws Exception {
        if (context == null) {
            try {
                context = getInitialContext();
            } catch (Exception ex) {
                ex.printStackTrace();
                throw ex;
            }
        }
        return context;
    }

    public void onMessage(Message message) {
        if (message instanceof javax.jms.BytesMessage) {
            javax.jms.BytesMessage bytesMessage = (javax.jms.BytesMessage)
                                                  message;
            /** @todo Process bytesMessage here */
        } else if (message instanceof javax.jms.MapMessage) {
            javax.jms.MapMessage mapMessage = (javax.jms.MapMessage) message;
            /** @todo Process mapMessage here */
        } else if (message instanceof javax.jms.ObjectMessage) {
            javax.jms.ObjectMessage objectMessage = (javax.jms.ObjectMessage)
                    message;
            /** @todo Process objectMessage here */
        } else if (message instanceof javax.jms.StreamMessage) {
            javax.jms.StreamMessage streamMessage = (javax.jms.StreamMessage)
                    message;
            /** @todo Process streamMessage here */
        } else if (message instanceof javax.jms.TextMessage) {
            javax.jms.TextMessage objectMessage = (javax.jms.TextMessage)
                                                  message;
            try {
                setStrContent(objectMessage.getText());
            } catch (JMSException ex1) {
            }
            /** @todo Process textMessage here */
        }
        if (isTransacted()) {
            try {
                getQueueSession().commit();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    public String getStrContent(){
        return this.strContent;
    }
    public void setStrContent(String str){
        this.strContent=str;
    }
    public void close() throws Exception {
        if (queueSender != null) {
            queueSender.close();
        }
        if (queueReceiver != null) {
            queueReceiver.close();
        }
        if (queueSession != null) {
            queueSession.close();
        }
        if (queueConnection != null) {
            queueConnection.close();
        }
    }

    private void jbInit() throws Exception {
    }
}