Java és JMS ActiveMQ módon

Beküldte pzoli - 2011, május 16 - 1:19du

Az Apache Active Message Queue (ActiveMQ) használatára készítettem néhány mintaprogramot. Az első kettő az üzenet körbeszórásra használható. Mindenek előtt a bin\activemq.bat-al el kell indítanunk az MQ szolgáltatást. Aztán a MessageListenerImp felhasználásával egy Main függvénnyel rendelkező osztályt hozzunk létre, és ebbe indítsuk el a messageListenerImpl.run() metódust. Ebben és az onMessage metódusban van a szerverünk lényegi része.

Mindegyik projektünkben szükségünk lesz az alábbi függvénykönyvtárakra

  • activemq-core-5.2.0
  • commons-logging-1.1
  • geronimo-j2ee-management_1.0_spec-1.0
  • javax.jms

    • Ez utóbbit a Java Message Service oldaláról tölthetjük le.

    MessageListenerImpl.java

    package com.infokristaly.messagequeue;
 
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
 
import org.apache.activemq.ActiveMQConnectionFactory;
 
public class MessageListenerImpl implements MessageListener {
    private Connection connection;
    private MessageProducer producer;
    private Session session;
    private int count;
    private long start;
    private Topic topic;
    private Topic control;
 
    private String url = "tcp://localhost:61616";
 
    public void run() throws JMSException {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
        connection = factory.createConnection();
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        topic = session.createTopic("topictest.messages");
        control = session.createTopic("topictest.control");
 
        MessageConsumer consumer = session.createConsumer(topic);
        consumer.setMessageListener(this);
 
        connection.start();
 
        producer = session.createProducer(control);
        System.out.println("Waiting for messages...");
    }
 
    private static boolean checkText(Message msg, String str) {
        try {
            return msg instanceof TextMessage && ((TextMessage)msg).getText().equals(str);
        } catch (JMSException e) {
            e.printStackTrace(System.out);
            return false;
        }
    }
 
    public void onMessage(Message message) {
        if (checkText(message, "SHUTDOWN")) {
            try {
                connection.close();
            } catch (Exception e) {
                e.printStackTrace(System.out);
            }
        } else {
        	if (message instanceof TextMessage)
            try {
                //Publish received message to Publishers
                String msg = (((TextMessage)message).getText());
                producer.send(session.createTextMessage(msg));
            } catch (Exception e) {
                e.printStackTrace(System.out);
            }
        }
    }
 
    public void setUrl(String url) {
        this.url = url;
    }
}

    A klienst egy MessagePublisherMain.java fájlban helyeztem el. A Swing felhasználói felülettel küldhető üzenet a szervernek (Listener), ami aztán a Topic-ra feliratkozottaknak szétszórja azt. A startPublisher() metódusban és az onMessage metódusban van a kliensünk lényegi része.

    /*
 * MessageListenerMain.java
 *
 * Created on 2010. január 2., 11:48
 */
package com.infokristaly.messagelistener;
 
import java.awt.event.KeyEvent;
import javax.jms.*;
import javax.jms.Message;
import org.apache.activemq.*;
 
/**
 *
 * @author  Papp Zoltán
 */
public class MessagePublisherMain extends javax.swing.JFrame implements MessageListener {
 
    private final Object mutex = new Object();
    private Connection connection;
    private Session session;
    private MessageProducer publisher;
    private String topicName = "topictest.messages";
    private String controlName = "topictest.control";
    private Topic topic;
    private Topic control;
    private String url = "tcp://localhost:61616";
 
    private SetupDialog setupDialog;
    /** Creates new form MessageListenerMain */
    public MessagePublisherMain() {
        initComponents();
    }
 
    private void sendMessage(String msg){
        try {
            publisher.send(session.createTextMessage(msg));
        } catch (Exception e) {
 
        }
    }
 
    /** This method is called from within the constructor to
     * initialize the form.
     * WARNING: Do NOT modify this code. The content of this method is
     * always regenerated by the Form Editor.
     */
    // <editor-fold defaultstate="collapsed" desc="Generated Code">
    private void initComponents() {
 
        jScrollPane1 = new javax.swing.JScrollPane();
        jTextArea = new javax.swing.JTextArea();
        edtMessage = new javax.swing.JTextField();
        btnSend = new javax.swing.JButton();
        jScrollPane2 = new javax.swing.JScrollPane();
        jList1 = new javax.swing.JList();
        jMenuBar = new javax.swing.JMenuBar();
        MainMenuItem = new javax.swing.JMenu();
        SetupMenuItem = new javax.swing.JMenuItem();
        ConnectMenuItem = new javax.swing.JMenuItem();
        DisconnectMenuItem = new javax.swing.JMenuItem();
        EditMenuItem = new javax.swing.JMenu();
 
        setDefaultCloseOperation(javax.swing.WindowConstants.EXIT_ON_CLOSE);
        addWindowListener(new java.awt.event.WindowAdapter() {
            public void windowClosing(java.awt.event.WindowEvent evt) {
                formWindowClosing(evt);
            }
        });
 
        jTextArea.setColumns(20);
        jTextArea.setRows(5);
        jTextArea.setFocusable(false);
        jScrollPane1.setViewportView(jTextArea);
 
        edtMessage.setEditable(false);
        edtMessage.addKeyListener(new java.awt.event.KeyAdapter() {
            public void keyPressed(java.awt.event.KeyEvent evt) {
                edtMessageKeyPressed(evt);
            }
        });
 
        btnSend.setText("Send");
        btnSend.setEnabled(false);
        btnSend.addActionListener(new java.awt.event.ActionListener() {
            public void actionPerformed(java.awt.event.ActionEvent evt) {
                btnSendActionPerformed(evt);
            }
        });
 
        jScrollPane2.setViewportView(jList1);
 
        MainMenuItem.setText("Main");
 
        SetupMenuItem.setText("Setup");
        SetupMenuItem.setName("SetupMenu"); // NOI18N
        SetupMenuItem.addActionListener(new java.awt.event.ActionListener() {
            public void actionPerformed(java.awt.event.ActionEvent evt) {
                SetupMenuActionPerformed(evt);
            }
        });
        MainMenuItem.add(SetupMenuItem);
 
        ConnectMenuItem.setText("Connect");
        ConnectMenuItem.setName("ConnectMenu"); // NOI18N
        ConnectMenuItem.addActionListener(new java.awt.event.ActionListener() {
            public void actionPerformed(java.awt.event.ActionEvent evt) {
                ConnectMenuActionPerformed(evt);
            }
        });
        MainMenuItem.add(ConnectMenuItem);
 
        DisconnectMenuItem.setText("Disconnect");
        DisconnectMenuItem.setEnabled(false);
        DisconnectMenuItem.addActionListener(new java.awt.event.ActionListener() {
            public void actionPerformed(java.awt.event.ActionEvent evt) {
                DisconnectMenuItemActionPerformed(evt);
            }
        });
        MainMenuItem.add(DisconnectMenuItem);
 
        jMenuBar.add(MainMenuItem);
 
        EditMenuItem.setText("Edit");
        jMenuBar.add(EditMenuItem);
 
        setJMenuBar(jMenuBar);
 
        javax.swing.GroupLayout layout = new javax.swing.GroupLayout(getContentPane());
        getContentPane().setLayout(layout);
        layout.setHorizontalGroup(
            layout.createParallelGroup(javax.swing.GroupLayout.Alignment.LEADING)
            .addGroup(layout.createSequentialGroup()
                .addContainerGap()
                .addGroup(layout.createParallelGroup(javax.swing.GroupLayout.Alignment.LEADING, false)
                    .addComponent(edtMessage)
                    .addComponent(jScrollPane1, javax.swing.GroupLayout.DEFAULT_SIZE, 358, Short.MAX_VALUE))
                .addGap(18, 18, 18)
                .addGroup(layout.createParallelGroup(javax.swing.GroupLayout.Alignment.LEADING)
                    .addComponent(btnSend, javax.swing.GroupLayout.DEFAULT_SIZE, 91, Short.MAX_VALUE)
                    .addComponent(jScrollPane2, javax.swing.GroupLayout.DEFAULT_SIZE, 91, Short.MAX_VALUE))
                .addContainerGap())
        );
        layout.setVerticalGroup(
            layout.createParallelGroup(javax.swing.GroupLayout.Alignment.LEADING)
            .addGroup(layout.createSequentialGroup()
                .addContainerGap()
                .addGroup(layout.createParallelGroup(javax.swing.GroupLayout.Alignment.TRAILING)
                    .addComponent(jScrollPane2, javax.swing.GroupLayout.DEFAULT_SIZE, 266, Short.MAX_VALUE)
                    .addComponent(jScrollPane1, javax.swing.GroupLayout.DEFAULT_SIZE, 266, Short.MAX_VALUE))
                .addPreferredGap(javax.swing.LayoutStyle.ComponentPlacement.RELATED)
                .addGroup(layout.createParallelGroup(javax.swing.GroupLayout.Alignment.BASELINE)
                    .addComponent(edtMessage, javax.swing.GroupLayout.DEFAULT_SIZE, 26, Short.MAX_VALUE)
                    .addComponent(btnSend))
                .addContainerGap())
        );
 
        pack();
    }// </editor-fold>
    private void btnSendActionPerformed(java.awt.event.ActionEvent evt) {
        sendMessage(edtMessage.getText());
        edtMessage.setText("");
}
 
    private void ConnectMenuActionPerformed(java.awt.event.ActionEvent evt) {
        startPublisher();        
}
 
    private void showSetupDialog(){
        if (setupDialog==null)
            setupDialog = new SetupDialog(this,true);
        setupDialog.setDefaultCloseOperation(HIDE_ON_CLOSE);
        setupDialog.setVisible(true);
    }
    private void SetupMenuActionPerformed(java.awt.event.ActionEvent evt) {
        showSetupDialog();
}
 
    private void Disconnect(){
        if (connection != null)
        try {
            connection.stop();
            connection.close();
        } catch (Exception e) {
        }
        btnSend.setEnabled(false);
        ConnectMenuItem.setEnabled(true);
        DisconnectMenuItem.setEnabled(false);
        SetupMenuItem.setEnabled(true);
        edtMessage.setEditable(false);
    }
    private void DisconnectMenuItemActionPerformed(java.awt.event.ActionEvent evt) {
        Disconnect();   
    }
 
    private void formWindowClosing(java.awt.event.WindowEvent evt) {
        Disconnect();
    }
 
    private void edtMessageKeyPressed(java.awt.event.KeyEvent evt) {
        if (evt.getKeyCode() == KeyEvent.VK_ENTER){
            sendMessage(edtMessage.getText());
            edtMessage.setText("");
        }
}
 
    public void startPublisher() {
        if (setupDialog==null || (setupDialog.ModalResult==setupDialog.btnCancel) ||
                setupDialog.edtName.getText().isEmpty())
            showSetupDialog();
        if (setupDialog.ModalResult==setupDialog.btnOK)
        try {
            jTextArea.append("Connecting "+setupDialog.edtName.getText()+"... ("+url+") with topic: "+topicName+"\r\n");
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);
            connection = factory.createConnection();
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            topic = session.createTopic(topicName);
            control = session.createTopic(controlName);
 
            publisher = session.createProducer(topic);
            publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            try {
              publisher.send(session.createTextMessage("[+"+setupDialog.edtName.getText()+"]"));
            } catch (Exception e){
 
            }
            btnSend.setEnabled(true);
            ConnectMenuItem.setEnabled(false);
            DisconnectMenuItem.setEnabled(true);
            SetupMenuItem.setEnabled(false);
            edtMessage.setEditable(true);
            session.createConsumer(control).setMessageListener(this);
            connection.start();
            jTextArea.append("Connected.\r\n");
        } catch (Exception e) {
 
        }
    }
 
    private static class runnableMain implements Runnable {
        public void run() {
                MessagePublisherMain main = new MessagePublisherMain();
                main.setVisible(true);
            }
    }
    /**
     * @param args the command line arguments
     */
    public static void main(String args[]) {
        java.awt.EventQueue.invokeLater(new runnableMain());
    }
 
    public void onMessage(Message message) {
        if (message instanceof TextMessage) {
            try {
                jTextArea.append(((TextMessage) message).getText() + "\r\n");
            } catch (Exception e) {
 
            }
        }
    }
    // Variables declaration - do not modify
    private javax.swing.JMenuItem ConnectMenuItem;
    private javax.swing.JMenuItem DisconnectMenuItem;
    private javax.swing.JMenu EditMenuItem;
    private javax.swing.JMenu MainMenuItem;
    private javax.swing.JMenuItem SetupMenuItem;
    private javax.swing.JButton btnSend;
    private javax.swing.JTextField edtMessage;
    private javax.swing.JList jList1;
    private javax.swing.JMenuBar jMenuBar;
    private javax.swing.JScrollPane jScrollPane1;
    private javax.swing.JScrollPane jScrollPane2;
    private javax.swing.JTextArea jTextArea;
    // End of variables declaration
}

    C# mintakód itt.