Apache ActiveMQ消息中间件的基本使用

Apache ActiveMQ是Apache软件基金会所研发的开放源码消息中间件;由于ActiveMQ是一个纯Java程式,因此只需要操作系统支援Java虚拟机,ActiveMQ便可执行。

支持Java消息服务 (JMS) 1.1 版本
Spring Framework
集群 (Clustering)
支持的编程语言包括:C、C++、C#、Delphi、Erlang、Adobe Flash、Haskell、Java、JavaScript、Perl、PHP、Pike、Python和Ruby [1]
协议支持包括:OpenWire、REST、STOMP、WS-Notification、XMPP以及AMQP

好,我们先写个demo来试试 ActiveMQ的效果.

首先我们要下载ActiveMQ,下载地址:

http://www.apache.org/dyn/closer.cgi?path=/activemq/apache-activemq/5.8.0/apache-activemq-5.8.0-bin.zip

解压后,在x:/apache-activemq-5.8.0-bin/bin 目录下执行 activemq.bat即可启动 ActiveMQ,

由于ActiveMQ内置了Jetty web服务器,当ActiveMQ启动成功后,可以通过:http://localhost:8161/admin/访问ActiveMQ的控制台,默认的用户名和密码是:admin。

至此ActiveMQ 服务已经启动了,接下来我们先将x:/apache-activemq-5.8.0-bin/activemq-all-5.8.0.jar拷贝到你的classpath目录下,利用Java写个demo来尝试一下这个消息中间件。

一、显示发送者的对象

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session; import org.apache.activemq.ActiveMQConnectionFactory; public class Sender { public static void main(String[] args) throws Exception {
aa();
}
public static void aa() throws Exception{
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616"); Connection connection = connectionFactory.createConnection();
connection.start(); Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("Test.foo"); MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
for(int i=0; i<100; i++) {
int id = i+1;
ObjectMessage message = session.createObjectMessage();
message.setObject(new User(id, "张三"+id, "123456"));
producer.send(message);
}
System.out.println("ok...");
session.commit();
session.close();
connection.close();
}
}

二、接受者对象

import java.util.concurrent.TimeUnit;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session; import org.apache.activemq.ActiveMQConnectionFactory; public class Receiver { public static void aa() throws Exception{
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616"); Connection connection = connectionFactory.createConnection();
connection.start(); final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("Test.foo"); MessageConsumer consumer = session.createConsumer(destination);
//listener 方式
consumer.setMessageListener(new MessageListener() { public void onMessage(Message msg) {
ObjectMessage message = (ObjectMessage) msg;
//TODO something....
try {
User user = (User) message.getObject();
System.out.println("收到消息:"+user.getName());
} catch (JMSException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
try {
session.commit();
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} });
TimeUnit.MINUTES.sleep(1); session.close();
connection.close();
} public static void main(String[] args) throws Exception{
aa();
}
}

  

 import java.io.Serializable;

 public class User implements Serializable {

     private int id;

     private String name;

     private String pwd;

     public User(int id,String name,String pwd){
this.id=id;
this.name=name;
this.pwd=pwd;
} public int getId() {
return id;
} public void setId(int id) {
this.id = id;
} public String getName() {
return name;
} public void setName(String name) {
this.name = name;
} public String getPwd() {
return pwd;
} public void setPwd(String pwd) {
this.pwd = pwd;
} }

先运行发送者,然后运行接收者即可。。。。

上一篇:Java基础—IO小结(二)缓冲流与其它流的使用


下一篇:Swift Standard Library: Documented and undocumented built-in functions in the Swift standard library – the complete list with all 74 functions