java activeMQ消息的发送与接收

刺骨的言语ヽ痛彻心扉 2022-06-08 23:51 415阅读 0赞

activemq是我们经常用到的消息队列之一，比如说速度快，对spring的很好的支持，支持多种协议等等，今天我们就来看一下activeMQ消息的发送与接收的源代码。

我这里使用了两个配置文件，其实在一个配置文件里面就可以完成发送与接收功能，但是为了方便观察日志，我就使用了两个配置文件。在使用代码之前需要搭建好activeMQ消息队列环境。

一、代码目录结构

所建立的工程是maven工程，代码结构如图所示：

![20170822163146931][]

一、maven配置pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    	<groupId>Test</groupId>
    	<artifactId>Test</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    
    
    	<dependencies>
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>fastjson</artifactId>
    			<version>1.2.16</version>
    		</dependency>
    		<dependency>
    		<groupId>org.json</groupId>
    		<artifactId>org.json</artifactId>
    		<version>chargebee-1.0</version>
    	</dependency>
    		<dependency>
    			<groupId>junit</groupId>
    			<artifactId>junit</artifactId>
    			<version>4.12</version>
    			<scope>test</scope>
    		</dependency>
    		<dependency>
    			<groupId>com.cloudhopper.proxool</groupId>
    			<artifactId>proxool</artifactId>
    			<version>0.9.1</version>
    			<exclusions>
    				<exclusion>
    					<artifactId>log4j</artifactId>
    					<groupId>log4j</groupId>
    				</exclusion>
    			</exclusions>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.activemq</groupId>
    			<artifactId>activemq-all</artifactId>
    			<version>5.11.0</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-jms</artifactId>
    			<version>4.1.4.RELEASE</version>
    
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-jdbc</artifactId>
    			<version>4.1.4.RELEASE</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-core</artifactId>
    			<version>4.1.4.RELEASE</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-beans</artifactId>
    			<version>4.1.4.RELEASE</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context</artifactId>
    			<version>4.1.4.RELEASE</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context-support</artifactId>
    			<version>4.1.4.RELEASE</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.cxf</groupId>
    			<artifactId>cxf-rt-frontend-jaxws</artifactId>
    			<version>2.7.3</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.cxf</groupId>
    			<artifactId>cxf-rt-transports-http</artifactId>
    			<version>2.7.3</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.cxf</groupId>
    			<artifactId>cxf-rt-transports-http-jetty</artifactId>
    			<version>2.4.5</version>
    		</dependency>
    		<dependency>
    			<groupId>org.slf4j</groupId>
    			<artifactId>slf4j-api</artifactId>
    			<version>1.7.5</version>
    		</dependency>
    
    
    		<dependency>
    			<groupId>com.h2database</groupId>
    			<artifactId>h2</artifactId>
    			<version>1.3.152</version>
    		</dependency>
    		<dependency>
    			<groupId>org.apache.httpcomponents</groupId>
    			<artifactId>httpclient</artifactId>
    			<version>4.1.1</version>
    		</dependency>
    		<dependency>
    			<groupId>dom4j</groupId>
    			<artifactId>dom4j</artifactId>
    			<version>1.6.1</version>
    		</dependency>
    		<dependency>
    			<groupId>commons-dbutils</groupId>
    			<artifactId>commons-dbutils</artifactId>
    			<version>1.3</version>
    		</dependency>
    		<dependency>
    			<groupId>org.freemarker</groupId>
    			<artifactId>freemarker</artifactId>
    			<version>2.3.16</version>
    		</dependency>
    		<dependency>
    			<groupId>jaxen</groupId>
    			<artifactId>jaxen</artifactId>
    			<version>1.1.1</version>
    		</dependency>
    		<dependency>
    			<groupId>net.sourceforge.saxon</groupId>
    			<artifactId>saxon</artifactId>
    			<version>9.1.0.8</version>
    		</dependency>
    		<dependency>
    			<groupId>net.sourceforge.saxon</groupId>
    			<artifactId>saxon</artifactId>
    			<version>9.1.0.8</version>
    			<classifier>xqj</classifier>
    		</dependency>
    		<dependency>
    			<groupId>net.sourceforge.saxon</groupId>
    			<artifactId>saxon</artifactId>
    			<version>9.1.0.8</version>
    			<classifier>xpath</classifier>
    		</dependency>
    		<dependency>
    			<groupId>net.sourceforge.saxon</groupId>
    			<artifactId>saxon</artifactId>
    			<version>9.1.0.8</version>
    			<classifier>xom</classifier>
    		</dependency>
    		<dependency>
    			<groupId>net.sourceforge.saxon</groupId>
    			<artifactId>saxon</artifactId>
    			<version>9.1.0.8</version>
    			<classifier>sql</classifier>
    		</dependency>
    		<dependency>
    			<groupId>net.sourceforge.saxon</groupId>
    			<artifactId>saxon</artifactId>
    			<version>9.1.0.8</version>
    			<classifier>s9api</classifier>
    		</dependency>
    		<dependency>
    			<groupId>net.sourceforge.saxon</groupId>
    			<artifactId>saxon</artifactId>
    			<version>9.1.0.8</version>
    			<classifier>jdom</classifier>
    		</dependency>
    		<dependency>
    			<groupId>net.sourceforge.saxon</groupId>
    			<artifactId>saxon</artifactId>
    			<version>9.1.0.8</version>
    			<classifier>dom4j</classifier>
    		</dependency>
    		<dependency>
    			<groupId>net.sourceforge.saxon</groupId>
    			<artifactId>saxon</artifactId>
    			<version>9.1.0.8</version>
    			<classifier>dom</classifier>
    		</dependency>
    		<dependency>
    			<groupId>xom</groupId>
    			<artifactId>xom</artifactId>
    			<version>1.1</version>
    		</dependency>
    		<dependency>
    			<groupId>stax</groupId>
    			<artifactId>stax</artifactId>
    			<version>1.2.0</version>
    		</dependency>
    		<dependency>
    			<groupId>org.drools</groupId>
    			<artifactId>drools-core</artifactId>
    			<version>5.3.0.Final</version>
    		</dependency>
    		<dependency>
    			<groupId>org.drools</groupId>
    			<artifactId>drools-compiler</artifactId>
    			<version>5.3.0.Final</version>
    		</dependency>
    		<dependency>
    			<groupId>commons-io</groupId>
    			<artifactId>commons-io</artifactId>
    			<version>2.2</version>
    		</dependency>
    		<dependency>
    			<groupId>javax.xml.bind</groupId>
    			<artifactId>jsr173_api</artifactId>
    			<version>1.0</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-test</artifactId>
    			<version>4.1.4.RELEASE</version>
    		</dependency>
    
    		<dependency>
    			<groupId>corba</groupId>
    			<artifactId>corba-connect</artifactId>
    			<version>0.0.2</version>
    		</dependency>
    
    		<dependency>
    			<groupId>com.ustcinfo.inm.data</groupId>
    			<artifactId>inm-data-spi</artifactId>
    			<version>0.0.1-SNAPSHOT</version>
    		</dependency>
    		<dependency>
    			<groupId>com.taobao.metamorphosis</groupId>
    			<artifactId>metamorphosis-client</artifactId>
    			<version>1.4.4</version>
    		</dependency>
    		<dependency>
    			<groupId>com.googlecode.aviator</groupId>
    			<artifactId>aviator</artifactId>
    			<version>2.2.1</version>
    		</dependency>
    		<dependency>
    			<groupId>com.jcraft</groupId>
    			<artifactId>jsch</artifactId>
    			<version>0.1.50</version>
    		</dependency>
    	</dependencies>
    	<build>
    		<sourceDirectory>src</sourceDirectory>
    		<plugins>
    			<plugin>
    				<artifactId>maven-compiler-plugin</artifactId>
    				<version>3.3</version>
    				<configuration>
    					<source>1.8</source>
    					<target>1.8</target>
    				</configuration>
    			</plugin>
    		</plugins>
    	</build>
    </project>

二、LoadUtil.java加载类

package www.activemq.load; import javax.jms.Destination; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.jms.core.JmsTemplate; /** * 加载配置文件工具类 * @author Administrator * */ public class LoadUtil { ApplicationContext applicationContext; JmsTemplate template;//jsm对象，可以发送和消费消息 Destination destination;//队列名称 /** * 加载方法 * @param path 文件路径名称 */ public void load(String path){ applicationContext = new ClassPathXmlApplicationContext(path); template = (JmsTemplate) applicationContext.getBean("jmsTemplate"); destination = (Destination) applicationContext.getBean("queueDestination"); this.setApplicationContext(applicationContext); this.setDestination(destination); this.setDestination(destination); } public ApplicationContext getApplicationContext() { return applicationContext; } public void setApplicationContext(ApplicationContext applicationContext) { this.applicationContext = applicationContext; } public JmsTemplate getTemplate() { return template; } public void setTemplate(JmsTemplate template) { this.template = template; } public Destination getDestination() { return destination; } public void setDestination(Destination destination) { this.destination = destination; } }

三、QueueProducer.java发送消息到activeMQ类

package www.avtivemq.sendmessage; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; import www.activemq.load.LoadUtil; public class QueueProducer{ /** * 发送消息到activemq的实现方法 * @param msg //发送消息的内容，为字符串类型 */ public void sendMessage(String msg) { LoadUtil lu = new LoadUtil(); lu.load("activemq-config.xml"); JmsTemplate template = lu.getTemplate(); String destination = template.getDefaultDestination().toString(); System.out.println("向队列" + destination + "发送了消息:" + msg); template.send(new MessageCreator() { public Message createMessage(Session session) throws JMSException { return session.createTextMessage(msg); } }); } /** * 主方法 * @param args */ public static void main(String[] args) { QueueProducer producer = new QueueProducer(); producer.sendMessage("zcinfo_test"); } }

四、activemq-config.xml配置文件

五、ActivemqConsumer.java消费类

package www.activemq.receivemessage;
    
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    import www.activemq.load.LoadUtil;
    
    public class ActivemqConsumer implements MessageListener{
    	
    	@Override
    	public void onMessage(Message message) {
    		TextMessage tm = (TextMessage)message;
    		System.out.println("监听到MQ中有数据......");
    		try {
    			System.out.println("获取MQ中数据信息>>>>>>>>>>" +  tm.getText());
    		}catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    	
    	/**
    	 * 加载配置文件之后监听器会自动调用onMessage方法，并且保持服务一直开启很实用
    	 * @param args
    	 */
    	public static void main(String[] args) {
    		new LoadUtil().load("activemq-context.xml");
    	}
    }

六、activemq-context.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" xmlns:jms="http://www.springframework.org/schema/jms" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd"> <context:component-scan base-package="com.starit.analyse" />  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  <property name="connectionFactory" ref="connectionFactory"/> </bean>  <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  <property name="brokerURL" value="failover:(tcp://192.0.81.80:8765)?jms.prefetchPolicy.all=1"/> </bean>  <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  <property name="targetConnectionFactory" ref="targetConnectionFactory"/> </bean>  <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg> <value>UEAP_TO_GZZC_QUEUE_TEST</value> </constructor-arg> </bean>  <bean id="consumerMessageListener" class="www.activemq.receivemessage.ActivemqConsumer"/>  <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="connectionFactory" /> <property name="destination" ref="queueDestination" /> <property name="messageListener" ref="consumerMessageListener" /> <property name="concurrentConsumers" value="1"/> </bean> </beans>

七、测试结果

（1）发送消息

![20170822171033778][]

（2）activemq显示收到消息

![20170822171041983][]

（3）接收消息，服务一直开启接收发送者消息

![20170822171048584][]

[20170822163146931]: /images/20220609/fd9b9651a6274907b6fe1c151aa596c4.png
[20170822171033778]: /images/20220609/07a85767e34d4a21ae9a517e5456b4cf.png
[20170822171041983]: /images/20220609/cba06d921f6d4363b97aafe4c449202f.png
[20170822171048584]: /images/20220609/9a8afe42671c46da9285ba6da0ac8619.png