active mq 多個消費者實戰(釋出訂閱模式)
阿新 • • 發佈:2019-02-17
注意:所有的配置檔案都在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