Getting started with ActiveMQ on Unix This FlashGuideTM will get you started with ActiveMQ. ActiveMQ implements the JMS 1.1 Specification. It is sponsored by Protique and is licensed under the Apache 2.0 license. 1. Installing ActiveMQ on UNIX
2. Configuring ActiveMQ
3. Testing ActiveMQ
cd ~ mkdir activemqtest export DEVEL_HOME=~/activemqtest cd activemqtest mkdir classes etc lib src
<project default="build" basedir=".">
<property name="activemq_home" value="/usr/local/activemq"/>
<path id="cp">
<pathelement path="${java.class.path}"/>
<fileset dir="${activemq_home}/lib">
<include name="*.jar"/>
</fileset>
<fileset dir="lib">
<include name="*.jar"/>
</fileset>
</path>
<target name="build">
<javac srcdir="src" destdir="classes"
debug="on" optimize="off" deprecation="off">
<classpath refid="cp"/>
</javac>
<jar jarfile="lib/activemqtest.jar">
<fileset dir="classes"
includes="**/*.class"/>
</jar>
</target>
</project>
You may have to change the location of the "activemq_home" property. jms-user=defaultUser jms-pwd=defaultPassword jms-url=tcp://localhost:61616 mode=topic transact-mode=false ack-mode=1 durability=true subject=test-topic consumer-id=ActiveMQClient producer-id=ActiveMQProducer consumer-name=Test
package test.jms;
import org.codehaus.activemq.ActiveMQConnection;
import org.codehaus.activemq.ActiveMQConnectionFactory;
import java.util.Properties;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Session;
public class AMQConnection {
private Connection connection = null;
private Session session = null;
private Destination destination;
private boolean ready = false;
public AMQConnection() {
}
public boolean init(Properties props, String clientId) {
try {
ActiveMQConnectionFactory connectionFactory =
new ActiveMQConnectionFactory(props.getProperty("jms-user"),
props.getProperty("jms-pwd"),
props.getProperty("jms-url"));
Connection connection = connectionFactory.createConnection();
boolean durable =
Boolean.valueOf(props.getProperty("durability")).booleanValue();
if (durable) {
connection.setClientID(clientId);
}
connection.start();
session = connection.createSession(
Boolean.valueOf(props.getProperty("transact-mode")).booleanValue(),
Integer.parseInt(props.getProperty("ack-mode")));
String mode = props.getProperty("mode");
if (mode.equals("topic"))
destination = session.createTopic(props.getProperty("subject"));
else
destination = session.createQueue(props.getProperty("subject"));
ready = true;
} catch (Exception e) {
}
return ready;
}
public void close() {
try {
System.out.println("Closing session ...");
if (session != null) session.close();
} catch (Exception e1) {
System.err.println("AMQConnection.close: " + e1.toString());
}
try {
System.out.println("Closing connection ...");
if (connection != null) connection.close();
} catch (Exception e2) {
System.err.println("AMQConnection.close: " + e2.toString());
}
}
public Session getSession() {
return session;
}
public Destination getDestination() {
return destination;
}
}
package test.jms;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.Hashtable;
import java.util.Properties;
import javax.jms.DeliveryMode;
import javax.jms.MessageProducer;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.Session;
public class AMQTestSender {
private AMQConnection conn = null;
private MessageProducer producer;
public AMQTestSender() {
}
public boolean init() {
String propsFile = System.getProperty("config");
if (propsFile == null || propsFile.equals("")) {
System.err.println("Error: no configuration file specified");
return false;
}
Properties props = new Properties();
try {
props.load(new FileInputStream(propsFile));
} catch (Exception e) {
System.err.println("Unable to load configuration from: " +
propsFile);
return false;
}
conn = new AMQConnection();
if (!conn.init(props, props.getProperty("producer-id"))) return false;
try {
Session session = conn.getSession();
producer = session.createProducer(conn.getDestination());
boolean durable =
Boolean.valueOf(props.getProperty("durability")).booleanValue();
if (durable) {
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
} else {
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
}
} catch (Exception e2) {
System.err.println("Unable to start publisher: " +
e2.toString());
return false;
}
System.out.println("AMQTestSender is waiting ...\n");
return true;
}
public void run() {
BufferedReader is = null;
try {
String input = "";
is = new BufferedReader(new InputStreamReader(System.in));
while (true) {
System.out.println();
System.out.print("Enter message: ");
input = is.readLine();
if (input.equals("") || input.compareToIgnoreCase("quit") == 0 || input.compareToIgnoreCase("exit") == 0)
break;
TextMessage msg = conn.getSession().createTextMessage();
msg.setText(input);
producer.send(msg);
//Thread.sleep(0L);
}
} catch (Exception e) {
System.err.println("AMQTestSender.run: " + e.toString());
} finally {
try {
if (is != null) is.close();
} catch (Exception e2) { }
}
conn.close();
System.exit(1);
}
public static void main(String[] args) {
AMQTestSender ts = new AMQTestSender();
if (ts.init()) ts.run();
}
}
package test.jms;
import java.io.FileInputStream;
import java.util.Hashtable;
import java.util.Properties;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import javax.jms.Session;
import javax.jms.Topic;
public class AMQTestReceiver implements MessageListener {
private Properties props = new Properties();
private AMQConnection conn = null;
private MessageConsumer consumer = null;
private ShutdownHook shutdownHook = null;
public AMQTestReceiver() {
}
public boolean init() {
String propsFile = System.getProperty("config");
if (propsFile == null || propsFile.equals("")) {
System.err.println("Error: no configuration file specified");
return false;
}
try {
props.load(new FileInputStream(propsFile));
} catch (Exception e) {
System.err.println("Unable to load configuration from: " +
propsFile);
return false;
}
conn = new AMQConnection();
if (!conn.init(props, props.getProperty("consumer-id"))) return false;
try {
Session session = conn.getSession();
boolean durable =
Boolean.valueOf(props.getProperty("durability")).booleanValue();
String mode = props.getProperty("mode");
if (durable && mode.equals("topic")) {
consumer = session.createDurableSubscriber(
(Topic)conn.getDestination(),
props.getProperty("consumer-name"));
} else {
consumer = session.createConsumer(conn.getDestination());
}
consumer.setMessageListener(this);
} catch (Exception e2) {
System.err.println("Unable to register listeners: " +
e2.toString());
return false;
}
System.out.println("AMQTestReceiver is waiting ...");
return true;
}
public void onMessage(Message message) {
try {
System.out.println("AMQTestReceiver: received " +
message.getJMSMessageID());
if (message instanceof TextMessage) {
String txt = ((TextMessage)message).getText();
System.out.println("\n" + txt);
if (txt.indexOf("SHUTDOWN") > -1) {
System.err.println("Received shutdown message ...");
consumer.setMessageListener(null);
Thread shutdown = new Thread() {
public void run() {
AMQTestReceiver.this.shutdown();
}
};
shutdown.start();
}
}
} catch (Exception e) {
}
}
public void shutdown() {
System.out.println("Shutting down receiver ...");
try {
if (shutdownHook != null)
Runtime.getRuntime().removeShutdownHook(shutdownHook);
} catch (Throwable t) { }
try {
consumer.close();
} catch (Exception e) { }
conn.close();
System.exit(1);
}
public static void main(String[] args) {
AMQTestReceiver tr = new AMQTestReceiver();
tr.init();
}
protected class ShutdownHook extends Thread {
public void run() {
AMQTestReceiver.this.shutdown();
}
}
}
Notice that the special message containing the string "SHUTDOWN" will cause the client to stop. Typically, you cannot stop the receiver from the onMessage method, and so we create a shutdown Thread to do the job for us.
#!/bin/sh
_ACTIVEMQ_HOME=/usr/local/activemq
_CLASSPATH=
for i in "$_ACTIVEMQ_HOME"/lib/*.jar; do
_CLASSPATH="$_CLASSPATH":"$i"
done
for i in lib/*.jar; do
_CLASSPATH="$_CLASSPATH":"$i"
done
send() {
java -Dconfig=etc/activemq.props -classpath "$_CLASSPATH" test.jms.AMQTestSender
}
receive() {
java -Dconfig=etc/activemq.props -classpath "$_CLASSPATH" test.jms.AMQTestReceiver
}
case "$1" in
'send')
send
;;
'receive')
receive
;;
*)
echo "Usage: $0 { send | receive }"
exit 1
;;
esac
Change the location of _ACTIVEMQ_HOME if necessary. Make sure the script is executable. $ cd $DEVEL_HOME $ ant build $ cd $ACTIVEMQ_HOME/bin $ ./activemq & $ cd $DEVEL_HOME $ ./activemqtest.sh receive CTRL-C $ $ cd $DEVEL_HOME $ ./activemqtest.sh send AMQTestSender is waiting ... Enter message: Hi there Enter message: Hope you get this! Enter message: Bye now Enter message: SHUTDOWN Enter message: Before close session Before close connection $ $ cd $DEVEL_HOME $ ./activemqtest.sh receive Before create durable subscriber AMQTestReceiver is waiting ... AMQTestReceiver: received ID:galadriel-39668-1109644140293-7:0 Hi there AMQTestReceiver: received ID:galadriel-39668-1109644140293-7:1 Hope you get this! AMQTestReceiver: received ID:galadriel-39668-1109644140293-7:2 Bye now AMQTestReceiver: received ID:galadriel-39668-1109644140293-7:3 SHUTDOWN Received shutdown message ... Shutting down receiver ... Closing session ... Closing connection ... $ Back to Table of Contents |
|
|