Spring + JMS + ActiveMQ实现简单的消息队列（监听器异步实现）

首先声明：以下内容均是在网上找别人的博客综合学习而成的，可能会发现某些代码与其他博主的相同，由于参考的文章比较多，这里对你们表示感谢，就不一一列举，如果有侵权的地方，请通知我，我可以把该文章删除。

1、jms-xml Spring配置文件

<?xml version="1.0" encoding="UTF-8"?>
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<bean id = "connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name = "brokerURL" value = "tcp://localhost:61616"/>
</bean>
<bean id = "topicDestination" class="org.apache.activemq.command.ActiveMQTopic"
autowire="constructor">
<constructor-arg value="com.spring.xkey.jms.topic"/>
</bean>
<bean id="sendMessage" class="com.spring.xkey.jms.SendMessage">
<property name="username" value="xkey"/>
<property name="password" value="1234567890"/>
</bean>
<bean id = "jmsMessageConverter" class="com.spring.xkey.jms.JmsMessageConverter">
<property name="sendMessage" ref="sendMessage"/>
</bean>
<bean id = "jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="receiveTimeout" value="6000"/>
<property name="messageConverter" ref="jmsMessageConverter"/>
</bean>
<bean id = "jmsMessageListener" class="com.spring.xkey.jms.JmsMessageListener">
</bean>
<bean id = "publisher" class="com.spring.xkey.jms.Publisher">
<property name="jmsTemplate" ref="jmsTemplate"/>
<property name="destinations" ref="topicDestination" />
<property name="sendMessage" ref="sendMessage"/>
</bean>
<bean id = "consumer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="topicDestination" />
<property name="messageListener" ref="jmsMessageListener" />
</bean>
</beans>

2、Listener代码

package com.spring.xkey.jms;
import java.util.Date;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import org.apache.activemq.command.ActiveMQMapMessage;
public class JmsMessageListener implements MessageListener {
public void onMessage(Message message) {
ActiveMQMapMessage msg = null;
//System.out.println("ONMessage-----------------" + message.toString());
try {
if (message instanceof ActiveMQMapMessage) {
msg = (ActiveMQMapMessage) message;
String username = msg.getString("username");
String password = msg.getString("password");
System.out.println("Message::: "+username+", "+password);
// msg = (ActiveMQMapMessage) message;
// String sentDate = msg.getString("date");
// String reMessage = msg.getString("message");
// int sentCount = msg.getInt("count");
// System.out
// .println("-------------New Message Arrival-----------"
// + new Date());
// System.out.println("It's " + sentCount + " time From Darcy: "
// + reMessage + " ---Send time :" + sentDate);
}
} catch (JMSException e) {
System.out.println("JMSException in onMessage(): " + e.toString());
} catch (Throwable t) {
System.out.println("Exception in onMessage():" + t.getMessage());
}
}
}

3、Converter代码

package com.spring.xkey.jms;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.jms.support.converter.MessageConverter;
public class JmsMessageConverter implements MessageConverter{
private SendMessage sendMessage;
public void setSendMessage(SendMessage sendMsg){
this.sendMessage = sendMsg;
}
public Object fromMessage(Message message) throws JMSException,
MessageConversionException {
// TODO Auto-generated method stub
MapMessage mapmessage= (MapMessage)message;
this.sendMessage.setUsername(mapmessage.getString("username"));
this.sendMessage.setPassword(mapmessage.getString("password"));
System.out.println("First");
return sendMessage;
}
public Message toMessage(Object arg0, Session session) throws JMSException,
MessageConversionException {
// TODO Auto-generated method stub
this.sendMessage = (SendMessage)arg0;
MapMessage mapmessage= (MapMessage) session.createMapMessage();
mapmessage.setString("username", this.sendMessage.getUsername());
mapmessage.setString("password", this.sendMessage.getPassword());
System.out.println("Second");
return mapmessage;
}
}

4、Publisher代码

package com.spring.xkey.jms;
import java.util.Scanner;
import javax.jms.Destination;
import org.springframework.jms.core.JmsTemplate;
public class Publisher {
private JmsTemplate template;
private Destination[] destinations;
private SendMessage sendMessage;
public void chart()
{
boolean chart = true;
int count = 0;
while(chart)
{
count ++;
Scanner cin=new Scanner(System.in);
System.out.println("输入聊天内容，输入N停止聊天");
String text=cin.nextLine();
if(text.equals("N"))
{
chart = false;
}
System.out.println("我："+text);
sendChartMessage(count,text);
}
}
public void sendMsgCon(){
Scanner cin=new Scanner(System.in);
String username = cin.nextLine();
String password = cin.nextLine();
this.sendMessage.setUsername(username);
this.sendMessage.setPassword(password);
sendConvertor(this.sendMessage);
}
public void sendConvertor(SendMessage sendMsg){
template.convertAndSend(destinations[0],sendMsg);
}
protected void sendChartMessage(int count , String strMessage)
{
MyMessageCreator creator = new MyMessageCreator(count,strMessage);
template.send(destinations[0], creator);
}
public JmsTemplate getJmsTemplate() {
return template;
}
public void setJmsTemplate(JmsTemplate template) {
this.template = template;
}
public Destination[] getDestinations() {
return destinations;
}
public void setDestinations(Destination[] destinations) {
this.destinations = destinations;
}
public void setSendMessage(SendMessage sendMsg){
this.sendMessage = sendMsg;
}
public SendMessage getSendMessage(){
return this.sendMessage;
}
}

5、SendMessage代码

package com.spring.xkey.jms;
public class SendMessage {
private String username;
private String password;
public void setUsername(String user){
this.username = user;
}
public void setPassword(String pass){
this.password = pass;
}
public String getUsername(){
return this.username;
}
public String getPassword(){
return this.password;
}
}

6、Test类

package com.spring.xkey.jms;
import javax.jms.JMSException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
public class Test {
/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
ApplicationContext context =
new ClassPathXmlApplicationContext("jms.xml");
/**Sender sender = (Sender)context.getBean("sender");
sender.SendInfo();
Receiver receiver = (Receiver)context.getBean("receiver");
try {
System.out.println(receiver.receiverInfo());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}*/
Publisher pub = (Publisher)context.getBean("publisher");
DefaultMessageListenerContainer consumer =
(DefaultMessageListenerContainer)context.getBean("consumer");
consumer.start();
pub.sendMsgCon();
//pub.chart();
}
}

秒客网

Spring + JMS + ActiveMQ实现简单的消息队列（监听器异步实现）

相关文章