消费者1（Consumer）--订阅（subcribe）-->主题（Topic）
|
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
|
package
com.java1234.activemq2;
import
javax.jms.Connection;
import
javax.jms.ConnectionFactory;
import
javax.jms.Destination;
import
javax.jms.JMSException;
import
javax.jms.MessageConsumer;
import
javax.jms.Session;
import
org.apache.activemq.ActiveMQConnection;
import
org.apache.activemq.ActiveMQConnectionFactory;
/**
* 消息消费者-消息订阅者一
* @author Administrator
*
*/
public
class
JMSConsumer {
private
static
final
String USERNAME=ActiveMQConnection.DEFAULT_USER;
// 默认的连接用户名
private
static
final
String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;
// 默认的连接密码
private
static
final
String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL;
// 默认的连接地址
public
static
void
main(String[] args) {
ConnectionFactory connectionFactory;
// 连接工厂
Connection connection =
null
;
// 连接
Session session;
// 会话 接受或者发送消息的线程
Destination destination;
// 消息的目的地
MessageConsumer messageConsumer;
// 消息的消费者
// 实例化连接工厂
connectionFactory=
new
ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
try
{
connection=connectionFactory.createConnection();
// 通过连接工厂获取连接
connection.start();
// 启动连接
session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// 创建Session
// destination=session.createQueue("FirstQueue1"); // 创建连接的消息队列
destination=session.createTopic(
"FirstTopic1"
);
messageConsumer=session.createConsumer(destination);
// 创建消息消费者
messageConsumer.setMessageListener(
new
Listener());
// 注册消息监听
}
catch
(JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
|
|
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
|
package
com.java1234.activemq2;
import
javax.jms.JMSException;
import
javax.jms.Message;
import
javax.jms.MessageListener;
import
javax.jms.TextMessage;
/**
* 消息监听-订阅者一
* @author Administrator
*
*/
public
class
Listener
implements
MessageListener{
@Override
public
void
onMessage(Message message) {
// TODO Auto-generated method stub
try
{
System.out.println(
"订阅者一收到的消息："
+((TextMessage)message).getText());
}
catch
(JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
|
消费者2（Consumer）--订阅（subcribe）-->主题（Topic）
生产者（Producer）--发布（publish）-->主题（Topic）
|
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
package
com.java1234.activemq2;
import
javax.jms.Connection;
import
javax.jms.ConnectionFactory;
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
JMSProducer {
private
static
final
String USERNAME=ActiveMQConnection.DEFAULT_USER;
// 默认的连接用户名
private
static
final
String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;
// 默认的连接密码
private
static
final
String BROKEURL=ActiveMQConnection.DEFAULT_BROKER_URL;
// 默认的连接地址
private
static
final
int
SENDNUM=
10
;
// 发送的消息数量
public
static
void
main(String[] args) {
ConnectionFactory connectionFactory;
// 连接工厂
Connection connection =
null
;
// 连接
Session session;
// 会话 接受或者发送消息的线程
Destination destination;
// 消息的目的地
MessageProducer messageProducer;
// 消息生产者
// 实例化连接工厂
connectionFactory=
new
ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
try
{
connection=connectionFactory.createConnection();
// 通过连接工厂获取连接
connection.start();
// 启动连接
session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
// 创建Session
// destination=session.createQueue("FirstQueue1"); // 创建消息队列
destination=session.createTopic(
"FirstTopic1"
);
//创建主题
messageProducer=session.createProducer(destination);
// 创建消息生产者
sendMessage(session, messageProducer);
// 发送消息
session.commit();
}
catch
(Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
finally
{
if
(connection!=
null
){
try
{
connection.close();
}
catch
(JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
/**
* 发送消息
* @param session
* @param messageProducer
* @throws Exception
*/
public
static
void
sendMessage(Session session,MessageProducer messageProducer)
throws
Exception{
for
(
int
i=
0
;i<JMSProducer.SENDNUM;i++){
TextMessage message=session.createTextMessage(
"ActiveMQ 发送的消息"
+i);
System.out.println(
"发送消息："
+
"ActiveMQ 发布的消息"
+i);
messageProducer.send(message);
}
}
}
|
本文转自 hgditren 51CTO博客，原文链接：http://blog.51cto.com/phpme/1952860，如需转载请自行联系原作者
