ActiveMQ能做什么
大多数情况下ActiveMQ被用于做系统之间的数据交换。
比如说几个不同的系统之间需要进行业务的联办，这个时候我们就需要一个第三方消息中间件来做消息传递。而ActiveMQ就是这个中间件。
使用ActiveMQ的用途和优点：
- 将数据从一个应用程序传送到另一个应用程序，或者从软件的一个模块传送到另外一个模块；
- 负责建立网络通信的通道，进行数据的可靠传送。
- 保证数据不重发，不丢失
- 能够实现跨平台操作，能够为不同操作系统上的软件集成数据传送服务
安装使用ActiveMQ
- 下载ActiveMQ
- 直接解压，启动，在命令行cd到ActiveMQ/bin目录下 ./activemq start
- 检测是否已启动
ActiveMQ默认采用61616端口提供JMS服务，使用8161端口提供管理控制台服务，执行以下命令便可以检验是否已经成功启动ActiveMQ服务
命令行中：netstat -an|find "61616"
直接访问ActiveMQ管理页面http://localhost:8161/admin/ 默认用户名密码admin/admin
- 关闭ActiveMQ可以使用./activemq stop
测试使用ActiveMQ
其实ActiveMQ是按照JMS（java消息服务）规范实现
JMS（java消息服务）就是典型的异步消息处理机制
ActiveMQ默认提供两种通信模式
p2p（点对点）
p2p比较简单，一方发送消息，一方接收消息。相互通信的双方是通过一个类似于队列的方式来进行交流。而在p2p里一个queue只有一个发送者和一个接收者。queue之间是通过名字区别的
//发送消息
public class P2pSend {
public static void main(String[] args) throws JMSException, InterruptedException {
//创建链接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
//创建连接
Connection connection = connectionFactory.createConnection();
connection.start();
//创建一个会话
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//创建生产者 发送消息的人
MessageProducer producer = session.createProducer(null);
Destination destination = session.createQueue("JOBS.1");
Message message = session.createObjectMessage(123);
System.out.println("Sending: id: " + ((ObjectMessage)message).getObject() + " on queue: " + destination);
producer.send(destination, message);
producer.close();
session.close();
connection.close();
}
}
//接收消息
public class P2pReceiver {
public static void main(String[] args) throws JMSException, InterruptedException {
//创建链接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
//创建连接
Connection connection = connectionFactory.createConnection();
connection.start();
//创建一个会话
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("JOBS.1");
MessageConsumer messageConsumer = session.createConsumer(destination);
messageConsumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message arg0) {
try {
Thread.sleep(2000);
System.out.println("1" + " id:" + ((ObjectMessage)arg0).getObject());
} catch (Exception e) {
e.printStackTrace();
}
}
});
//session.close();
//connection.close();
}
}
这里主要就是通过session.createQueue("队列名称")中的队列名称来判断有那个消费者进行消费的。
访问localhost:8161/admin/后台可以看到
publish-subscribe（发布/订阅）
//消息的发送者
public class PubSend {
public static void main(String[] args) throws JMSException {
//创建链接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
//创建连接
Connection connection = connectionFactory.createConnection();
connection.start();
//创建一个会话
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//创建生产者 发送消息的人
MessageProducer producer = session.createProducer(null);
Destination destination = session.createTopic("JOB01");
MapMessage message = session.createMapMessage();
message.setString("name", "shli");
message.setDouble("price", 1.00);
message.setBoolean("up", true);
producer.send(destination, message);
producer.close();
session.close();
connection.close();
}
}
//消息接收者
public class PubReceive {
public static void main(String[] args) throws JMSException {
//创建链接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
//创建连接
Connection connection = connectionFactory.createConnection();
connection.start();
//创建一个会话
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic("JOB01");
MessageConsumer messageConsumer = session.createConsumer(destination);
messageConsumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message arg0) {
MapMessage map = (MapMessage)arg0;
try {
String shli = map.getString("name");
double price = map.getDouble("price");
boolean up = map.getBoolean("up");
System.out.println(shli + "----" + price +"------"+up);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
MessageConsumer messageConsumer1 = session.createConsumer(destination);
messageConsumer1.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message arg0) {
MapMessage map = (MapMessage)arg0;
try {
String shli = map.getString("name");
double price = map.getDouble("price");
boolean up = map.getBoolean("up");
System.out.println(shli + "-1-1-1-" + price +"-1--1-1--"+up);
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//session.close();
//connection.close();
}
}
#######这里主要注意的地方就是
- 消费者只能接收到注册之后，发送端发送的消息
- 发送端发送的消息会分发给所有注册的消费者
- 注册发送端的关键在于session.createTopic("Topic名称")中的Topic名称
访问localhost:8161/admin/ 后台可以看到
说明
我也是第一次学习MQ，记录学习内容，难免会有错误的地方。如果发现，请指出，谢谢~
