JMS消息中间件原理及ActiveMQ在企业中的应用（接上篇）

时间：2015-08-19 20:42:27 阅读：221 评论：0 收藏：0 [点我收藏+]

代码实现：发送消息---》接受消息---》服务器配置
//1 发送消息（接受回复消息）
public class SenderMessageService {
//发布指定消息到指定地址(在发布之前，建议将消息保存到数据库)
public void publish(String type, Object object) {
try {
InitialContext initCtx = new InitialContext();
//1
Context envContext = (Context) initCtx.lookup("java:comp/env");
//2
ConnectionFactory connectionFactory = (ConnectionFactory) envContext
.lookup("jms/NormalConnectionFactory");
//3
Connection connection = connectionFactory.createConnection();
//4
Session jmsSession = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
//5,6 Destination:需指定其对应的主题（subject）名称
MessageProducer producer = jmsSession
.createProducer((Destination) envContext
.lookup("jms/topic/MyTopic"));
// 设置持久方式：根据Destination创建MessageProducer对象，同时设置其持久模式
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
//7
Message message = jmsSession.createMessage();
message.setObjectProperty(type, object);
Topic topic = jmsSession.createTopic("jms/topic/MyTopic");
//8
message.setJMSReplyTo(topic);
//发送消息
producer.send(message);

//9 接受回复的消息
MessageConsumer consumer = jmsSession.createConsumer(topic);
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
if (message != null && message instanceof TextMessage) {
String messageReceived = null;
try {
messageReceived = ((TextMessage) message).getText();
} catch (JMSException e) {
e.printStackTrace();
}
System.out
.println("reply message received from customer1:"
+ messageReceived);
}
}
});
connection.start();

// 发布刷新帖子消息
// testMessage.clearProperties();
// testMessage.setStringProperty("RefreshThreadId", "331");
// producer.send(testMessage);
} catch (NamingException e) {
e.printStackTrace();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
//2接受消息（发送回复）
import javax.servlet.*;
import javax.servlet.http.*;
import javax.naming.*;
import javax.jms.*;
import com.brightmart.MessageAction;
import com.brightmart.SM;
import com.util.mail.TestSendMail;
// 初始化jms连接，创建topic监听器;指定接收消息时候，做的对应处理
public class JMSListener extends HttpServlet implements MessageListener {
private static final long serialVersionUID = 3963233366687996777L;
//初始化jms连接，创建topic监听器
public void init(ServletConfig config) throws ServletException {
try {
InitialContext initCtx = new InitialContext();
Context envContext = (Context) initCtx.lookup("java:comp/env");// 1
// 根据JNDI获取
ConnectionFactory connectionFactory = (ConnectionFactory) envContext
.lookup("jms/FailoverConnectionFactory");// 2
Connection connection = connectionFactory.createConnection();// 3
// 给connection设置一个clientId
connection.setClientID("MyClient");
// 会话：两个参，事务和应答模式
Session jmsSession = connection.createSession(false,
Session.CLIENT_ACKNOWLEDGE);// 4 AUTO_ACKNOWLEDGE
// 普通消息订阅者，无法接收持久消息// MessageConsumer consumer =
// jmsSession.createConsumer((Destination)//envContext.lookup("jms/topic/MyTopic"));
// 基于Topic创建持久的消息订阅者，前提：Connection必须指定一个唯一的clientId，当前为MyClient
TopicSubscriber consumer = jmsSession.createDurableSubscriber(
(Topic) envContext.lookup("jms/topic/MyTopic"), "MySub");// 5
consumer.setMessageListener(this);
connection.start();
} catch (NamingException e) {
e.printStackTrace();
} catch (JMSException e) {
e.printStackTrace();
}
}
//接收消息，做相应处理
public void onMessage(Message message) {
System.out.println("message in coustomer1.");
if (message == null) {
return;
}
try {
if (message.getObjectProperty("email") != null) {
String emailAddress = (String) message
.getObjectProperty("email");
TestSendMail sendMail = new TestSendMail();
sendMail.sendMail(emailAddress);
message.acknowledge();
Destination d = message.getJMSReplyTo();
Session sessionn = getConnection().createSession(false,
Session.CLIENT_ACKNOWLEDGE);
MessageProducer p = sessionn.createProducer(d);
TextMessage tm = sessionn
.createTextMessage("ustomer1 RECEIVED a email type message");
System.out
.println("customer1 RECEIVED a email type message");
p.send(tm);
} else if (message.getObjectProperty("message") != null) {
MessageAction m = new MessageAction();
SM sm = new SM();
sm.setDestTermId((String) message.getObjectProperty("message"));
sm.setMsgContent("分布式JMS-ActiveMQ系统测试");
m.addSM(sm);
message.acknowledge();
} else {
System.out.println("接收普通消息，不做任何处理！");
}
} catch (JMSException e) {
e.printStackTrace();
}
}
public Connection getConnection() {
InitialContext initCtx;
try {
initCtx = new InitialContext();
Context envContext = (Context) initCtx.lookup("java:comp/env");
// 根据JNDI、url、user、password获取
ConnectionFactory connectionFactory = (ConnectionFactory) envContext
.lookup("jms/FailoverConnectionFactory");
Connection connection = connectionFactory.createConnection();
return connection;
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
}

//3 context.xml
Context中添加配置：
<Resource
name="jms/FailoverConnectionFactory"
auth="Container"
type="org.apache.activemq.ActiveMQConnectionFactory"
description="JMS Connection Factory"
factory="org.apache.activemq.jndi.JNDIReferenceFactory"
brokerURL="failover:(tcp://localhost:61616)?initialReconnectDelay=100&maxReconnectAttempts=5"
brokerName="localhost"
useEmbeddedBroker="false"/>
<Resource
name="jms/NormalConnectionFactory"
auth="Container"
type="org.apache.activemq.ActiveMQConnectionFactory"
description="JMS Connection Factory"
factory="org.apache.activemq.jndi.JNDIReferenceFactory"
brokerURL="tcp://localhost:61616"
brokerName="localhost"
useEmbeddedBroker="false"/>
<Resource name="jms/topic/MyTopic"
auth="Container"
type="org.apache.activemq.command.ActiveMQTopic"
factory="org.apache.activemq.jndi.JNDIReferenceFactory"
physicalName="MY.TEST.FOO"/>
<Resource name="jms/queue/MyQueue"
auth="Container"
type="org.apache.activemq.command.ActiveMQQueue"
factory="org.apache.activemq.jndi.JNDIReferenceFactory"
physicalName="MY.TEST.FOO.QUEUE"/>

4.发送短信或邮件
请参考本博客前两篇文章
注：
1.系统下载ActiveMQ，并允许；
2.ActiveMQ需要融合web服务器，如可以配置tomcat服务器的context.xml;
3在项目中，需要引入ActiveMQ的jar.

标签：消息中间件 jms activemq java

原文地址：http://blog.csdn.net/zeb_perfect/article/details/47783769

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行