使用apache的activemq集合JMS处理异步消息

dong5300

1：先去apache下载

http://activemq.apache.org/ ，大约有25M，要有点耐心

直接打开就可以使用；当然它也提供了安装为windows service的方法

2：配置activemq

在conf文件夹中配置

activemq.xml

主要是配置内存和端口

<policyEntry queue=">" memoryLimit="100mb"/>
<policyEntry topic=">" memoryLimit="100mb">
<memoryUsage limit="50 mb"/>

3：使用web方式访问管理

http://localhost:8161/admin/

很是方便

4：编程方式使用，系统安装后，有一个DEMO，你可以先测试一下，以后就照这个DEMO写就行了



5：在应用中的web.xml配置

<context-param>
<param-name>brokerURI</param-name>
<param-value>/WEB-INF/activemq.xml</param-value>
</context-param>
<listener>
<listener-class>org.activemq.web.SpringBrokerContextListener</listener-class>
</listener>

6：工具类

/**
*
*/
package MQ;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
* @author Administrator
* 消息的生产者（发送者）
*/
public class JmsSender {

public static void main(String []args) throws JMSException{
//ConnectionFactory:连接工厂，JMS用它创建连接
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://127.0.0.1:61616");
//JMS客户端到JMS Provider的连接
Connection connection = connectionFactory.createConnection();
connection.start();
//Session:一个发送或接收消息的线程
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
//Destination:消息的目的地；消息发送给谁
//获取session注意参数值my-queue是Query的名字
Destination destination = session.createQueue("my-queue");
//MessageProducer:消息生产者
MessageProducer producer = session.createProducer(destination);
//设置不持久化
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//发送一条信息
sendMsg(session,producer);
session.commit();
connection.close();
}

/**
* 在指定的会话上，通过指定的消息生产者发出一条消息
* @param session 消息会话
* @param producer消息生产者
* @throws JMSException
*/
private static void sendMsg(Session session, MessageProducer producer) throws JMSException {
//创建一条文本消息
TextMessage textMessage = session.createTextMessage("Hello ActiveMQ!");
//通过消息生产者发出消息
producer.send(textMessage);
System.out.println("");
}
}

/**
*
*/
package MQ;

/**
* @author Administrator
* 消息的消费者(接受者)
*/
public class JmsReceiver {

/**
* @param args
* @throws JMSException
*/
public static void main(String[] args) throws JMSException {
//ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://127.0.0.1:61616");
//JMS客户端到JMS Provider的连接
Connection connection = connectionFactory.createConnection();
connection.start();
//session:一个发送或接收消息的线程
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
//Destination:消息的目的地，消息发送给谁
//获取session注意参数值xingbo.xu-queue是一个服务器的queue是一个服务器的queue，须在在ActiveMq的console的配置
Destination destination = session.createQueue("my-queue");
//消息者，消息接收者
MessageConsumer consumer = session.createConsumer(destination);
while(true){
TextMessage message = (TextMessage)consumer.receive(1000);
if(null!=message){
System.out.println("收到消息："+message.getText());
}else{
break;
}
}
consumer.close();
session.close();
connection.close();
}

}