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-->
生產者的啟動程式碼
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">
<!-- 此處可以不要，如果不使用 -->
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