ActiveMQ接收消息+发送消息的简单实例

ActiveMQ的一个简单实例-ActiveMQ接收+发送消息 JMS消息框架——ActiveMQ

最近由于公司项目需要 -ActiveMQ接收+发送消息，用的是ActiveMQ。由于这方面网上的例子不是很多，而且有的也不完整。于是经过几天的摸索学习，在网上找到了合适的方案。

我的 IT技术资源库 http://itlib.tk/

ProducerTool.java用于发送消息：

java 代码

packagehomework;

importjavax.jms.Connection;
importjavax.jms.DeliveryMode;
importjavax.jms.Destination;
importjavax.jms.JMSException;
importjavax.jms.MessageProducer;
importjavax.jms.Session;
importjavax.jms.TextMessage;

importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;

publicclassProducerTool{

privateStringuser=ActiveMQConnection.DEFAULT_USER;

privateStringpassword=ActiveMQConnection.DEFAULT_PASSWORD;

privateStringurl=ActiveMQConnection.DEFAULT_BROKER_URL;

privateStringsubject="TOOL.DEFAULT";

privateDestinationdestination=null;

privateConnectionconnection=null;

privateSessionsession=null;

privateMessageProducerproducer=null;

//初始化
privatevoidinitialize()throwsJMSException,Exception{
ActiveMQConnectionFactoryconnectionFactory=newActiveMQConnectionFactory(
user,password,url);
connection=connectionFactory.createConnection();
session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
destination=session.createQueue(subject);
producer=session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
}

//发送消息
publicvoidproduceMessage(Stringmessage)throwsJMSException,Exception{
initialize();
TextMessagemsg=session.createTextMessage(message);
connection.start();
System.out.println("Producer:->Sendingmessage:"+message);
producer.send(msg);
System.out.println("Producer:->Messagesentcomplete!");
}

//关闭连接
publicvoidclose()throwsJMSException{
System.out.println("Producer:->Closingconnection");
if(producer!=null)
producer.close();
if(session!=null)
session.close();
if(connection!=null)
connection.close();
}
}

ConsumerTool.java用于接受消息，我用的是基于消息监听的机制，需要实现MessageListener接口，这个接口有个onMessage方法，当接受到消息的时候会自动调用这个函数对消息进行处理。

java 代码

packagehomework;

importjavax.jms.Connection;
importjavax.jms.Destination;
importjavax.jms.JMSException;
importjavax.jms.MessageConsumer;
importjavax.jms.Session;
importjavax.jms.MessageListener;
importjavax.jms.Message;
importjavax.jms.TextMessage;

importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;

publicclassConsumerToolimplementsMessageListener{

privateStringuser=ActiveMQConnection.DEFAULT_USER;

privateStringpassword=ActiveMQConnection.DEFAULT_PASSWORD;

privateStringurl=ActiveMQConnection.DEFAULT_BROKER_URL;

privateStringsubject="TOOL.DEFAULT";

privateDestinationdestination=null;

privateConnectionconnection=null;

privateSessionsession=null;

privateMessageConsumerconsumer=null;

//初始化
privatevoidinitialize()throwsJMSException,Exception{
ActiveMQConnectionFactoryconnectionFactory=newActiveMQConnectionFactory(
user,password,url);
connection=connectionFactory.createConnection();
session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
destination=session.createQueue(subject);
consumer=session.createConsumer(destination);

}

//消费消息
publicvoidconsumeMessage()throwsJMSException,Exception{
initialize();
connection.start();

System.out.println("Consumer:->Beginlistening...");
//开始监听
consumer.setMessageListener(this);
//Messagemessage=consumer.receive();
}

//关闭连接
publicvoidclose()throwsJMSException{
System.out.println("Consumer:->Closingconnection");
if(consumer!=null)
consumer.close();
if(session!=null)
session.close();
if(connection!=null)
connection.close();
}

//消息处理函数
publicvoidonMessage(Messagemessage){
try{
if(messageinstanceofTextMessage){
TextMessagetxtMsg=(TextMessage)message;
Stringmsg=txtMsg.getText();
System.out.println("Consumer:->Received:"+msg);
}else{
System.out.println("Consumer:->Received:"+message);
}
}catch(JMSExceptione){
//TODOAuto-generatedcatchblock
e.printStackTrace();
}
}
}

如果想主动的去接受消息，而不用消息监听的话，把consumer.setMessageListener(this)改为Message message = consumer.receive()，手动去调用MessageConsumer的receive方法即可。

下面是测试类Test.java：

java 代码

packagehomework;

importjavax.jms.JMSException;

publicclassTest{

/**
*@paramargs
*/
publicstaticvoidmain(String[]args)throwsJMSException,Exception{
//TODOAuto-generatedmethodstub
ConsumerToolconsumer=newConsumerTool();
ProducerToolproducer=newProducerTool();
//开始监听
consumer.consumeMessage();

//延时500毫秒之后发送消息
Thread.sleep(500);
producer.produceMessage("Hello,world!");
producer.close();

//延时500毫秒之后停止接受消息
Thread.sleep(500);
consumer.close();
}
}

ActiveMQ的一个简单实例-ActiveMQ接收+发送消息

IT技术资源库 http://itlib.tk/