transactedexample.java

java书 java书 java书 java书 java书 java书

/*
 * Copyright (c) 2006 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) 2006 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.ConnectionFactory;
import javax.jms.Queue;
import javax.jms.Topic;
import javax.jms.Connection;
import javax.jms.Session;
import javax.jms.MessageProducer;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Message;
import javax.jms.MapMessage;
import javax.jms.JMSException;
import java.util.Hashtable;
import java.util.Random;
import javax.annotation.Resource;


public class TransactedExample {
    @Resource(mappedName = "jms/ConnectionFactory")
    private static ConnectionFactory connectionFactory;
    @Resource(mappedName = "jms/AQueue")
    private static Queue vendorOrderQueue;
    @Resource(mappedName = "jms/BQueue")
    private static Queue retailerConfirmQueue;
    @Resource(mappedName = "jms/CQueue")
    private static Queue vendorConfirmQueue;
    @Resource(mappedName = "jms/OTopic")
    private static Topic supplierOrderTopic;

    /**
     * Creates the Retailer and Vendor classes and the two
     * supplier classes, then starts the threads.
     *
     * @param quantity    the quantity specified on the command
     *                    line
     */
    public static void run_threads(int quantity) {
        Retailer r = new Retailer(quantity);
        Vendor v = new Vendor();
        GenericSupplier ms = new GenericSupplier("Monitor");
        GenericSupplier ss = new GenericSupplier("Hard Drive");

        ms.start();
        ss.start();

        // make sure both supplier connections have started
        ms.waitForTopicConsumer();
        ss.waitForTopicConsumer();

        r.start();
        v.start();

        try {
            ms.join();
            ss.join();
            r.join();
            v.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * Reads the order quantity from the command line, then
     * calls the run_threads method to execute the program
     * threads.
     *
     * @param args    the quantity of computers being ordered
     */
    public static void main(String[] args) {
        TransactedExample te = new TransactedExample();
        int quantity = 0;

        if (args.length != 1) {
            System.out.println("Program takes numerical argument.");
            System.err.println(
                    "Program assumes three "
                    + "queues named jms/AQueue, jms/BQueue, and jms/CQueue "
                    + "and one topic named jms/OTopic");
            System.exit(1);
        }

        quantity = (new Integer(args[0])).intValue();
        System.out.println("Quantity to be ordered is " + quantity);

        if (quantity > 0) {
            te.run_threads(quantity);
            System.exit(0);
        } else {
            System.out.println("Quantity must be positive and " + "nonzero");
            System.exit(1);
        }
    }

    /**
     * The GenericSupplier class receives an item order from the
     * vendor and sends a message accepting or refusing it.
     */
    public static class GenericSupplier extends Thread {
        final String PRODUCT_NAME;
        boolean ready = false;
        int quantity = 0;

        /**
         * Constructor.  Instantiates the supplier as the
         * supplier for the kind of item to be ordered.
         *
         * @param itemName    the name of the item being ordered
         */
        public GenericSupplier(String itemName) {
            PRODUCT_NAME = itemName;
        }

        /**
         * Timer method. Completes when ready is set to true, after
         * connection is started.
         */
        void waitForTopicConsumer() {
            while (!(ready)) {
            }
        }

        /**
         * Checks to see if there are enough items in inventory.
         * Rather than go to a database, it generates a random
         * number related to the order quantity, so that some of
         * the time there won't be enough in stock.
         *
         * @return    the number of items in inventory
         */
        public int checkInventory() {
            Random rgen = new Random();

            return (rgen.nextInt(quantity * 5));
        }

        /**
         * Runs the thread.
         */
        public void run() {
            Connection connection = null;
            Session session = null;
            MessageConsumer receiver = null;
            Message inMessage = null;
            MapMessage orderMessage = null;
            MapMessage outMessage = null;

            try {
                connection = connectionFactory.createConnection();
                session = connection.createSession(true, 0);
            } catch (Exception e) {
                System.err.println("Connection problem: " + e.toString());
                System.err.println(
                        "Program assumes three "
                        + "queues named jms/AQueue, jms/BQueue, and jms/CQueue "
                        + "and one topic named jms/OTopic");

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

                System.exit(1);
            }

            /*
             * Create receiver for order topic and start message
             * delivery.
             */
            try {
                receiver = session.createConsumer(supplierOrderTopic);
                connection.start();
            } catch (JMSException je) {
            }

            // Connection has started, set ready to true
            ready = true;

            /*
             * Keep checking supplier order topic for order
             * request until end-of-message-stream message is
             * received. Receive order and send an order
             * confirmation as one transaction.
             */
            while (true) {
                try {
                    inMessage = receiver.receive();

                    if (inMessage instanceof MapMessage) {
                        orderMessage = (MapMessage) inMessage;
                    } else {
                        /*
                         * Message is an end-of-message-stream
                         * message. Send a similar message to
                         * reply queue, commit transaction, then
                         * stop processing orders by breaking out
                         * of loop.
                         */
                        MessageProducer producer = session.createProducer(
                                    inMessage.getJMSReplyTo());
                        producer.send(session.createMessage());
                        session.commit();

                        break;
                    }

                    /*
                     * Extract quantity ordered from order
                     * message.
                     */
                    quantity = orderMessage.getInt("Quantity");
                    System.out.println(
                            PRODUCT_NAME + " Supplier: Vendor ordered "
                            + quantity + " " + PRODUCT_NAME + "(s)");

                    /*
                     * Create sender and message for reply queue.
                     * Set order number and item; check inventory
                     * and set quantity available.
                     * Send message to vendor and commit
                     * transaction.
                     */
                    MessageProducer producer = session.createProducer(
                                (Queue) orderMessage.getJMSReplyTo());
                    outMessage = session.createMapMessage();
                    outMessage.setInt(
                            "VendorOrderNumber",
                            orderMessage.getInt("VendorOrderNumber"));
                    outMessage.setString("Item", PRODUCT_NAME);

                    int numAvailable = checkInventory();

                    if (numAvailable >= quantity) {
                        outMessage.setInt("Quantity", quantity);
                    } else {
                        outMessage.setInt("Quantity", numAvailable);
                    }

                    producer.send(outMessage);
                    System.out.println(
                            PRODUCT_NAME + " Supplier: sent "
                            + outMessage.getInt("Quantity") + " "
                            + outMessage.getString("Item") + "(s)");
                    session.commit();
                    System.out.println(
                            "  " + PRODUCT_NAME
                            + " Supplier: committed transaction");
                } catch (Exception e) {
                    System.err.println(
                            PRODUCT_NAME + " Supplier: Exception occurred: "
                            + e.toString());
                    e.printStackTrace();
                }
            }

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

    /**
     * The Order class represents a Retailer order placed with a
     * Vendor. It maintains a table of pending orders.
     */
    public static class Order {
        private static Hashtable<Integer, Order> pendingOrders = new Hashtable<Integer, Order>();
        private static int nextOrderNumber = 1;
        private static final int PENDING_STATUS = 1;