ActiveMQ的傳送與接收 簡單例子和監聽配置
簡單傳送例項：
package com.xuwei.activemq;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Sender {
private static final int SEND_NUMBER = 5;
public static void main(String[] args) {
// ConnectionFactory ：連線工廠，JMS 用它建立連線
ConnectionFactory connectionFactory;
// Connection ：JMS 客戶端到JMS Provider 的連線
Connection connection = null;
// Session： 一個傳送或接收訊息的執行緒
Session session;
// Destination ：訊息的目的地;訊息傳送給誰.
Destination destination;
// MessageProducer：訊息傳送者
MessageProducer producer;
// TextMessage message;
// 構造ConnectionFactory例項物件，此處採用ActiveMq的實現jar
connectionFactory = new ActiveMQConnectionFactory(
"admin","admin",
// ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://192.168.215.128:61616");
try {
// 構造從工廠得到連線物件
connection = connectionFactory.createConnection();
// 啟動
connection.start();
// 獲取操作連線
session = connection.createSession(Boolean.TRUE,
Session.AUTO_ACKNOWLEDGE);
// 獲取session注意引數值xingbo.xu-queue是一個伺服器的queue，須在在ActiveMq的console配置
//destination = session.createQueue("zhangqiao1");
destination = session.createTopic("ea_mq_teaching_group");
// 得到訊息生成者【傳送者】
producer = session.createProducer(destination);
// 設定不持久化，此處學習，實際根據專案決定
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 構造訊息，此處寫死，專案就是引數，或者方法獲取
sendMessage(session, producer);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
public static void sendMessage(Session session, MessageProducer producer)
throws Exception {
for (int i = 1; i <= SEND_NUMBER; i++) {
TextMessage message = session
.createTextMessage("ActiveMq 傳送的訊息" + i);
// 傳送訊息到目的地方
System.out.println("傳送訊息：" + "ActiveMq 傳送的訊息" + i);
producer.send(message);
}
}
}
簡單接收例項：
package com.jetsen.platform.prepareLesson.receiveMQ;
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
import net.sf.json.JSONObject;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Receivermq implements MessageListener{
private static boolean stop = false;
public void execute() throws Exception {
//連線工廠
ConnectionFactory connFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://http://192.168.215.128:61616");
//連線到JMS提供者
Connection conn = connFactory.createConnection();
conn.start();
//事務性會話，自動確認訊息
Session session = conn.createSession(true, Session.AUTO_ACKNOWLEDGE);
//訊息的來源地
Destination destination = session.createQueue("ea_mq_teaching_group"); //createQueue點對點
//訊息消費者
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(this);
//等待接收訊息
while(!stop){
// Thread.sleep(5000);
}
session.commit();
consumer.close();
session.close();
conn.close();
}
public void onMessage(Message m) {
try{
if(m instanceof TextMessage){ //接收文字訊息
TextMessage message = (TextMessage)m;
System.out.println("文字訊息"+message.getText());
}else if(m instanceof MapMessage){ //接收鍵值對訊息
MapMessage message = (MapMessage)m;
// System.out.println(message.getLong("age"));
// System.out.println(message.getDouble("sarray"));
// System.out.println(message.getString("username"));
System.out.println("鍵值對訊息 ");
}else if(m instanceof StreamMessage){ //接收流訊息
StreamMessage message = (StreamMessage)m;
System.out.println("流訊息"+message.readString());
System.out.println("流訊息"+message.readLong());
}else if(m instanceof BytesMessage){ //接收位元組訊息
byte[] b = new byte[1024];
int len = -1;
BytesMessage message = (BytesMessage)m;
while((len=message.readBytes(b))!=-1){
System.out.println("位元組訊息"+new String(b, 0, len));
String jsonstr = new String(b, 0, len);
JSONObject jsonObject = JSONObject.fromObject(jsonstr);
}
}else if(m instanceof ObjectMessage){ //接收物件訊息
ObjectMessage message = (ObjectMessage)m;
// User user = (User)message.getObject();
System.out.println("物件訊息");
}else{
System.out.println(m);
}
stop = true;
}catch(JMSException e){
stop = true;
e.printStackTrace();
}
}
}
以上傳送完和啟動接收就可以看到效果。
下面進行監聽配置MessageListener：
配置一個傳送的xml：
<?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:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-2.5.xsd"
default-autowire="byName">
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<!-- <property name="brokerURL" value="${activemq.brokerURL}" />
<property name="userName" value="${activemq.userName}" />
<property name="password" value="${activemq.password}" /> -->
<!-- <property name="brokerURL" value="tcp://198.9.6.39:61616" /> -->
<!-- 我的activemq -->
<property name="brokerURL" value="tcp://192.168.215.128:61616" />
<property name="userName" value="admin" />
<property name="password" value="admin" />
</bean>
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
</bean>
<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
<!-- 設定訊息佇列的名字 -->
<!-- <constructor-arg index="0" value="${activemq.messageName}" /> -->
<constructor-arg index="0" value="ea_mq_teaching_group" />
</bean>
</beans>
然後傳送程式碼：
package com.jetsen.platform.prepareLesson.receiveMQ;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.FileSystemXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
public class Sendmq{
/**
* @param args
* jmsTemplate和destination都是在spring配置檔案中進行配製的
* Sender只使用了配置檔案中的jmsFactory，jmsTemplate，還有destination這三個屬性
*/
public static void main(String[] args) {
ApplicationContext applicationContext = new FileSystemXmlApplicationContext("/WebRoot/WEB-INF/platform-sendavtivemq.xml");
JmsTemplate template = (JmsTemplate) applicationContext.getBean("jmsTemplate");
Destination destination = (Destination) applicationContext.getBean("destination");
template.send(destination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage("傳送的內容");
}
});
System.out.println("成功傳送了一條JMS訊息");
}
}
配置接收的xml（platform-reveiveavtivemq.xml）:
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:task="http://www.springframework.org/schema/task"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:jaxrs="http://cxf.apache.org/jaxrs"
xmlns:jaxws="http://cxf.apache.org/jaxws"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop.xsd
http://www.springframework.org/schema/task
http://www.springframework.org/schema/task/spring-task-3.0.xsd
http://www.springframework.org/schema/util
http://www.springframework.org/schema/util/spring-util-2.5.xsd
http://cxf.apache.org/jaxws http://cxf.apache.org/schemas/jaxws.xsd
http://cxf.apache.org/jaxrs http://cxf.apache.org/schemas/jaxrs.xsd
http://cxf.apache.org/jaxws http://cxf.apache.org/schemas/jaxws.xsd
http://cxf.apache.org/transports/http/configuration
http://cxf.apache.org/schemas/configuration/http-conf.xsd">
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<!-- <property name="brokerURL" value="${activemq.brokerURL}" />
<property name="userName" value="${activemq.userName}" />
<property name="password" value="${activemq.password}" /> -->
<property name="brokerURL" value="tcp://192.168.215.128:61616" />
<property name="userName" value="admin" />
<property name="password" value="admin" />
</bean>
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
</bean>
<bean id="mqDestination" class="org.apache.activemq.command.ActiveMQQueue"> <!-- org.apache.activemq.command.ActiveMQQueue 點對點的包-->
<!-- <constructor-arg>
<value>${activemq.messageName}</value>
</constructor-arg> -->
<!-- 設定訊息佇列的名字 -->
<constructor-arg index="0" value="ea_mq_teaching_group" />
<!-- <constructor-arg index="0" value="${activemq.messageName}" /> -->
</bean>
<bean id="messageListener" class="com.jetsen.platform.prepareLesson.receiveMQ.Receiver"></bean> <!-- class中就是傳送程式碼路徑-->
<bean id="listenerContainer" class="org.springframework.jms.listener.SimpleMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="mqDestination" />
<property name="messageListener" ref="messageListener" />
</bean>
</beans>
傳送程式碼：
package com.jetsen.platform.prepareLesson.receiveMQ;
import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
import net.sf.json.JSONObject;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.FileSystemXmlApplicationContext;
//監聽程式碼
public class Receiver implements MessageListener{
@Override
public void onMessage(Message m) {
try{
if(m instanceof TextMessage){ //接收文字訊息
TextMessage message = (TextMessage)m;
System.out.println("文字訊息:"+message.getText());
}else if(m instanceof MapMessage){ //接收鍵值對訊息
MapMessage message = (MapMessage)m;
// System.out.println(message.getLong("age"));
// System.out.println(message.getDouble("sarray"));
// System.out.println(message.getString("username"));
System.out.println("鍵值對訊息 ");
}else if(m instanceof StreamMessage){ //接收流訊息
StreamMessage message = (StreamMessage)m;
System.out.println("流訊息:"+message.readString());
System.out.println("流訊息:"+message.readLong());
}else if(m instanceof BytesMessage){ //接收位元組訊息
byte[] b = new byte[1024];
int len = -1;
BytesMessage message = (BytesMessage)m;
while((len=message.readBytes(b))!=-1){
System.out.println("位元組訊息:"+new String(b, 0, len));
String jsonstr = new String(b, 0, len);
JSONObject jsonObject = JSONObject.fromObject(jsonstr);
System.out.println("接收成功");
}
}else if(m instanceof ObjectMessage){ //接收物件訊息
ObjectMessage message = (ObjectMessage)m;
// User user = (User)message.getObject();
System.out.println("物件訊息");
}else{
System.out.println(m);
}
}catch(JMSException e){
e.printStackTrace();
}
}
}
在web.xml中配置啟動專案就監聽：
在web.xml中找到
<context-param>
<param-name>contextConfigLocation</param-name>
<param-value>platform-reveiveavtivemq.xml</param-value>
</context-param>
將需要啟動載入的xml放入<param-value></param-value>中。