ActiveMQ的简单使用

ActiveMQ的简单使用

ActiveMQ 是一种消息队列,是Apache提供的一个开源消息系统,完全采用Java实现,因此,它能很好地支持J2EE提出的JMS(Java Message Service,即Java消息服务)规范。

第一种方式(点对点)——Queue

生产者
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Producer {
public static void main(String[] args) {
//连接工厂
ConnectionFactory connectionFactory;
//连接
Connection connection = null;
//会话
Session session;
//目的地
Destination destination;
//消息生产者
MessageProducer messageProducer;
try {
connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,ActiveMQConnectionFactory.DEFAULT_PASSWORD,
ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
connection = connectionFactory.createConnection();
connection.start();
session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE); //不开启事务
destination = session.createQueue("Queue-zy");
messageProducer = session.createProducer(destination);
messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); //是否持久化(默认持久化)
for(int i = 1; i<11; i++){
System.out.println("生产者成功发出消息——Queue-zy:序号---"+i);
TextMessage message = session.createTextMessage("消息-----"+i+"-------hello,MQ!");
messageProducer.send(message);
}
} catch (JMSException e) {
e.printStackTrace();
} finally{
try {
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}

控制台打印出发送消息成功的语句,总计10条消息:

1
2
3
4
5
6
7
8
9
10
生产者成功发出消息——Queue-zy:序号---1
生产者成功发出消息——Queue-zy:序号---2
生产者成功发出消息——Queue-zy:序号---3
生产者成功发出消息——Queue-zy:序号---4
生产者成功发出消息——Queue-zy:序号---5
生产者成功发出消息——Queue-zy:序号---6
生产者成功发出消息——Queue-zy:序号---7
生产者成功发出消息——Queue-zy:序号---8
生产者成功发出消息——Queue-zy:序号---9
生产者成功发出消息——Queue-zy:序号---10

image
此时,打开ActiveMQ的管理页面,点击Queues,便可看到一个名为 Queue-zy 的队列:
Number Of Pending Messages :表示等待被消费的消息数为 10

Messages Enqueued: 表示进入队列的消息数为 10 (可叠加)

Messages Dequeued: 表示出队列的消息,此时为0

消费者
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Consumer {
public static void main(String[] args) {
//连接工厂
ConnectionFactory connectionFactory;
//连接
Connection connection = null;
//会话
Session session;
//目的地
Destination destination;
//消息生产者
MessageConsumer messageConsumer ;
try {
connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,ActiveMQConnectionFactory.DEFAULT_PASSWORD,
ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
connection = connectionFactory.createConnection();
connection.start();
session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
destination = session.createQueue("Queue-zy");
messageConsumer = session.createConsumer(destination);
while(true){
TextMessage textMessage = (TextMessage) messageConsumer.receive(10000);
if(textMessage != null){
System.out.println("消费者成功接收到消息: "+textMessage.getText());
}else{
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally{
try {
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}

点击Run启动消费者Main(),可以看到控制台输出,成功接收10条消息:

1
2
3
4
5
6
7
8
9
10
消费者成功接收到消息: 消息-----1-------hello,MQ!
消费者成功接收到消息: 消息-----2-------hello,MQ!
消费者成功接收到消息: 消息-----3-------hello,MQ!
消费者成功接收到消息: 消息-----4-------hello,MQ!
消费者成功接收到消息: 消息-----5-------hello,MQ!
消费者成功接收到消息: 消息-----6-------hello,MQ!
消费者成功接收到消息: 消息-----7-------hello,MQ!
消费者成功接收到消息: 消息-----8-------hello,MQ!
消费者成功接收到消息: 消息-----9-------hello,MQ!
消费者成功接收到消息: 消息-----10-------hello,MQ!

image

再次查看ActiveMQ的管理页面,点击Queues

Number Of Pending Messages :等待被消费的消息数变为 0,此时消息已经全部被消费

Messages Enqueued: 进入队列的消息数为 10 (叠加)

Messages Dequeued: 出队列的消息,此时为10

第二种方式(发布/订阅)——Topic

两种方式的代码只有一处区别:由Session创建的目的地不同

1
2
3
4
//Queue
destination = session.createQueue("Queue-zy");
//Topic
destination = session.createTopic("Topic-zy");

注意:先启动订阅者(消费者),再启动发布者(生产者)————因为在没有实现订阅消息持久化的情况下,只能在订阅者上线后才能接收到发布者发布的消息。

生产者
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Producer {
public static void main(String[] args) {
//连接工厂
ConnectionFactory connectionFactory;
//连接
Connection connection = null;
//会话
Session session;
//目的地
Destination destination;
//消息生产者
MessageProducer messageProducer;
try {
connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,ActiveMQConnectionFactory.DEFAULT_PASSWORD,
ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
connection = connectionFactory.createConnection();
connection.start();
session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE); //不开启事务
//destination = session.createQueue("Queue-zy");
destination = session.createTopic("Topic-zy");
messageProducer = session.createProducer(destination);
messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); //是否持久化(默认持久化)
for(int i = 1; i<11; i++){
System.out.println("生产者成功发出消息——Topic-zy:序号---"+i);
TextMessage message = session.createTextMessage("消息-----"+i+"-------hello,MQ!");
messageProducer.send(message);
}
} catch (JMSException e) {
e.printStackTrace();
} finally{
try {
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
消费者 1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Consumer {
public static void main(String[] args) {
//连接工厂
ConnectionFactory connectionFactory;
//连接
Connection connection = null;
//会话
Session session;
//目的地
Destination destination;
//消息生产者
MessageConsumer messageConsumer ;
try {
connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,ActiveMQConnectionFactory.DEFAULT_PASSWORD,
ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
connection = connectionFactory.createConnection();
connection.start();
session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
//destination = session.createQueue("Queue-zy");
destination = session.createTopic("Topic-zy");
messageConsumer = session.createConsumer(destination);
while(true){
TextMessage textMessage = (TextMessage) messageConsumer.receive(10000);
if(textMessage != null){
System.out.println("消费者 1 成功接收到消息: "+textMessage.getText());
}else{
break;
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
消费者 2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Consumer2 {
public static void main(String[] args) {
//连接工厂
ConnectionFactory connectionFactory;
//连接
Connection connection = null;
//会话
Session session;
//目的地
Destination destination;
//消息生产者
MessageConsumer messageConsumer ;
try {
connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,ActiveMQConnectionFactory.DEFAULT_PASSWORD,
ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
connection = connectionFactory.createConnection();
connection.start();
session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
//destination = session.createQueue("Queue-zy");
destination = session.createTopic("Topic-zy");
messageConsumer = session.createConsumer(destination);
while(true){
TextMessage textMessage = (TextMessage) messageConsumer.receive(10000);
if(textMessage != null){
System.out.println("消费者 2 成功接收到消息: "+textMessage.getText());
}else{
break;
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
}

先启动消费者 1 和 2,再启动生产者,可以看到控制台输出,消费者 1 、2都接收到了消息:

image

image

image

查看ActiveMQ的管理页面,点击Topics,查看名为 Topic-zy 的主题:

Number Of Consumers :表示消费者数目为 2

Messages Enqueued: 进入队列的消息数为 10

Messages Dequeued: 出队列的消息,为20 (因为2个消费者)

以上就是直接使用 ActiveMQ 在点对点发布订阅 两种模式下收发消息的过程。

下一篇,将介绍ActiveMQ结合Spring收发消息的过程。