注意:所有的配置檔案都在src檔案下
聲明: 在這裡不講activemq 是什麼,本人隻是根據平時用到的東西整理一下希望對大家有所幫助。
首先是生産者的配置檔案
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:jms="http://www.springframework.org/schema/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.0.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd">
<!-- Default Destination Queue Definition-->
<bean id="defaultDestination" class="org.apache.activemq.command.ActiveMQTopic"> <!-- 如果為點對點的方式此處ActiveMQTopic 應該改為ActiveMQQueue-->
<constructor-arg index="0" value="test.topic"/> </bean> <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <property name="sessionCacheSize" value="10" /> <property name="targetConnectionFactory" > <bean class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="failover:(tcp://localhost:61616?wireFormat.maxInactivityDuration=0)&maxReconnectDelay=1000" /> </bean> </property> </bean> <!-- JmsTemplate Definition --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="connectionFactory"/> <property name="defaultDestination" ref="defaultDestination"/> <!-- "pubSubDomain" true表示釋出/訂閱 模式--> <property name="pubSubDomain" value="true" /> </bean> <!-- 自定義發送消息對象 --> <!-- Message Sender Definition --> <bean id="messageSender" class="com.zyc.core.activemq.handle.MessageSender"> <constructor-arg index="0" ref="jmsTemplate"/> </bean> </beans>
生産者的啟動代碼
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Scanner;
public class SenderApp {
public static void main(String[] args) throws Exception {
start("JMSConfiguration.xml");
}
public static void start(String configLocation) throws Exception {
MessageSender sender = getMessageSender(configLocation);
System.out.println("Please input your message:");
String text = null;
int i = 0;
boolean isTrue = true;
while (isTrue) {
text = "abc" + i;
System.out.println(String.format("send message: %s", text));
sender.send(text);
if (i == 10)
isTrue = false;
i++;
}
}
public static MessageSender getMessageSender(String configLocation) {
ApplicationContext context = new ClassPathXmlApplicationContext(
configLocation);
return (MessageSender) context.getBean("messageSender");
}
}
生産者的核心發送代碼
package com.zyc.core.activemq.handle;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;
public class MessageSender {
private final JmsTemplate jmsTemplate;
public MessageSender(final JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public void send(final String text) throws Exception {
//jmsTemplate.convertAndSend("test.topic",text);
jmsTemplate.convertAndSend(text);
System.out.println("send: " + text);
}
}
消費者的配置檔案
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:jms="http://www.springframework.org/schema/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.0.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd">
<!-- 此處可以不要,如果不使用 -->
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> <property name="order" value="1" /> <property name="ignoreUnresolvablePlaceholders" value="true" /> <property name="location"> <value>classpath:activemq.properties</value> </property> </bean><bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"> <property name="connectionFactory" > <bean class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="failover:(tcp://localhost:61616?wireFormat.maxInactivityDuration=0)&maxReconnectDelay=1000" /> </bean> </property> <property name="maxConnections" value="1"></property> </bean> <!-- JmsTemplate Definition --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="connectionFactory"/> <property name="pubSubDomain" value="true" /> </bean> <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto"><jms:listener destination="test.topic" ref="messageReceiver"/><jms:listener destination="test.topic" ref="messageReceiver2"/></jms:listener-container> <!-- 自定義接受消息對象 --> <!-- Message Receiver Definition --> <bean id="messageReceiver" class="com.zyc.core.activemq.handle.MessageReceiver"> </bean> <bean id="messageReceiver2" class="com.zyc.core.activemq.handle.MessageReceiver2"> </bean> </beans>
消費者的監聽器 messageReceiver2 和messageReceiver 一樣這裡隻是想用2個消費者測試 package com.zyc.core.activemq.handle;
import java.text.SimpleDateFormat;
import java.util.Date;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class MessageReceiver implements MessageListener {
private SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
public void onMessage(Message message) {
if(message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
String text = textMessage.getText();
System.out.println(String.format("Received: %s",text+",time:"+sf.format(new Date())));
textMessage.acknowledge();
try {
Thread.sleep(0);
} catch (InterruptedException e) {
e.printStackTrace();
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
消費者的啟動代碼
package com.zyc.core.app;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class ReceiverApp {
public static void main( String[] args )
{
new ClassPathXmlApplicationContext("JMSConfigurationReceive.xml");
}
}
用到的jar如下
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>4.1.6.RELEASE</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.8.0</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.8.0</version>
</dependency>
<dependency>
<groupId>commons-pool</groupId>
<artifactId>commons-pool</artifactId>
<version>1.6</version>
</dependency>
注:5.4以上版本 activemq 預設的持久化存儲是kahaDB