ackequivexample.java

java指南

/*
 * Copyright (c) 2004 Sun Microsystems, Inc.  All rights reserved.  U.S.
 * Government Rights - Commercial software.  Government users are subject
 * to the Sun Microsystems, Inc. standard license agreement and
 * applicable provisions of the FAR and its supplements.  Use is subject
 * to license terms.
 *
 * This distribution may include materials developed by third parties.
 * Sun, Sun Microsystems, the Sun logo, Java and J2EE are trademarks
 * or registered trademarks of Sun Microsystems, Inc. in the U.S. and
 * other countries.
 *
 * Copyright (c) 2004 Sun Microsystems, Inc. Tous droits reserves.
 *
 * Droits du gouvernement americain, utilisateurs gouvernementaux - logiciel
 * commercial. Les utilisateurs gouvernementaux sont soumis au contrat de
 * licence standard de Sun Microsystems, Inc., ainsi qu'aux dispositions
 * en vigueur de la FAR (Federal Acquisition Regulations) et des
 * supplements a celles-ci.  Distribue par des licences qui en
 * restreignent l'utilisation.
 *
 * Cette distribution peut comprendre des composants developpes par des
 * tierces parties. Sun, Sun Microsystems, le logo Sun, Java et J2EE
 * sont des marques de fabrique ou des marques deposees de Sun
 * Microsystems, Inc. aux Etats-Unis et dans d'autres pays.
 */


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


public class AckEquivExample {
    final String CONTROL_QUEUE = "jms/ControlQueue";
    final String queueName = "jms/Queue";
    final String topicName = "jms/Topic";
    final String conFacName = "jms/DurableConnectionFactory";

    /**
     * Instantiates the sender, receiver, subscriber, and
     * publisher classes and starts their threads.
     * Calls the join method to wait for the threads to die.
     */
    public void run_threads() {
        SynchSender synchSender = new SynchSender();
        SynchReceiver synchReceiver = new SynchReceiver();
        AsynchSubscriber asynchSubscriber = new AsynchSubscriber();
        MultiplePublisher multiplePublisher = new MultiplePublisher();

        synchSender.start();
        synchReceiver.start();

        try {
            synchSender.join();
            synchReceiver.join();
        } catch (InterruptedException e) {
        }

        asynchSubscriber.start();
        multiplePublisher.start();

        try {
            asynchSubscriber.join();
            multiplePublisher.join();
        } catch (InterruptedException e) {
        }
    }

    /**
     * Reads the queue and topic names from the command line,
     * then calls the run_threads method to execute the program
     * threads.
     *
     * @param args    the topic used by the example
     */
    public static void main(String[] args) {
        AckEquivExample aee = new AckEquivExample();

        if (args.length != 0) {
            System.out.println("Program takes no arguments.");
            System.exit(1);
        }

        System.out.println("Queue name is " + aee.CONTROL_QUEUE);
        System.out.println("Queue name is " + aee.queueName);
        System.out.println("Topic name is " + aee.topicName);
        System.out.println("Connection factory name is " + aee.conFacName);

        aee.run_threads();
        System.exit(0);
    }

    /**
     * The SynchSender class creates a session in
     * CLIENT_ACKNOWLEDGE mode and sends a message.
     */
    public class SynchSender extends Thread {
        /**
         * Runs the thread.
         */
        public void run() {
            ConnectionFactory connectionFactory = null;
            Connection connection = null;
            Session session = null;
            Queue queue = null;
            MessageProducer producer = null;
            final String MSG_TEXT =
                new String("Here is a client-acknowledge message");
            TextMessage message = null;

            try {
                connectionFactory = SampleUtilities.getConnectionFactory();
                connection = connectionFactory.createConnection();
                session =
                    connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
                queue = SampleUtilities.getQueue(queueName, session);
            } catch (Exception e) {
                System.err.println("Connection problem: " + e.toString());

                if (connection != null) {
                    try {
                        connection.close();
                    } catch (JMSException ee) {
                    }
                }

                System.exit(1);
            }

            /*
             * Create client-acknowledge sender.
             * Create and send message.
             */
            try {
                System.out.println("  SENDER: Created " +
                    "client-acknowledge session");
                producer = session.createProducer(queue);
                message = session.createTextMessage();
                message.setText(MSG_TEXT);
                System.out.println("  SENDER: Sending " + "message: " +
                    message.getText());
                producer.send(message);
            } catch (JMSException e) {
                System.err.println("Exception occurred: " + e.toString());
            } finally {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (JMSException e) {
                    }
                }
            }
        }
    }

    /**
     * The SynchReceiver class creates a session in
     * CLIENT_ACKNOWLEDGE mode and receives the message sent by
     * the SynchSender class.
     */
    public class SynchReceiver extends Thread {
        /**
         * Runs the thread.
         */
        public void run() {
            ConnectionFactory connectionFactory = null;
            Connection connection = null;
            Session session = null;
            Queue queue = null;
            MessageConsumer receiver = null;
            TextMessage message = null;

            try {
                connectionFactory = SampleUtilities.getConnectionFactory();
                connection = connectionFactory.createConnection();
                session =
                    connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
                queue = SampleUtilities.getQueue(queueName, session);
            } catch (Exception e) {
                System.err.println("Connection problem: " + e.toString());

                if (connection != null) {
                    try {
                        connection.close();
                    } catch (JMSException ee) {
                    }
                }

                System.exit(1);
            }

            /*
             * Create client-acknowledge receiver.
             * Receive message and process it.
             * Acknowledge message.
             */
            try {
                System.out.println("  RECEIVER: Created " +
                    "client-acknowledge session");
                receiver = session.createConsumer(queue);
                connection.start();
                message = (TextMessage) receiver.receive();
                System.out.println("  RECEIVER: Processing " + "message: " +
                    message.getText());
                System.out.println("  RECEIVER: Now I'll " +
                    "acknowledge the message");
                message.acknowledge();
            } catch (JMSException e) {
                System.err.println("Exception occurred: " + e.toString());
            } finally {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (JMSException e) {
                    }
                }
            }
        }
    }

    /**
     * The AsynchSubscriber class creates a session in
     * AUTO_ACKNOWLEDGE mode and fetches several messages from a
     * topic asynchronously, using a message listener,
     * TextListener.
     *
     * Each message is acknowledged after the onMessage method