spring整合使用activemq

末蓝、 2023-10-18 09:14 69阅读 0赞

**activemq简单的demo这里就不做演示了，今天介绍一下如何利用spring整合activemq，也是实际工作中涉及到的，希望对各位伙伴有所帮助，**

1、安装activemq，为演示方便，我已经提前在linux中安装了activemq，安装过程比较简单，应该说activemq是软件中算是最简单的一种了，只需要解压，进入bin目录启动即可，不需做其他的配置；  
![在这里插入图片描述][70]

只需要一条命令，即可启动，注意需要把activemq的端口配置到防护墙的文件中，然后重启防火墙，否则浏览器直接访问是访问不到的，

成功启动后，浏览器输入：[http://192.168.111.130:8161/admin，可进入管控台查看，][http_192.168.111.130_8161_admin]  
![在这里插入图片描述][70 1]

在这个页面可以进行相关的配置，具体可参考官网说明，为后面的使用，我这里提前创建一个队列名称，test\_queue,后面可以直接拿来使用；

接下来是项目和代码的部分；

2、项目结构，  
![在这里插入图片描述][70 2]

为模拟真实的场景，这里我创建了两个maven项目，首先看producer的部分，

![在这里插入图片描述][70 3]  
pom.xml文件，

<dependencies>
    
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-core</artifactId>
    			<version>4.0.0.RELEASE</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context</artifactId>
    			<version>4.0.0.RELEASE</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-orm</artifactId>
    			<version>4.0.0.RELEASE</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-web</artifactId>
    			<version>4.0.0.RELEASE</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-jdbc</artifactId>
    			<version>4.0.0.RELEASE</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-webmvc</artifactId>
    			<version>4.0.0.RELEASE</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-aspects</artifactId>
    			<version>4.0.0.RELEASE</version>
    		</dependency>
    
    		<dependency>
    			<groupId>com.mchange</groupId>
    			<artifactId>c3p0</artifactId>
    			<version>0.9.2</version>
    		</dependency>
    		<dependency>
    			<groupId>cglib</groupId>
    			<artifactId>cglib</artifactId>
    			<version>2.2</version>
    		</dependency>
    		<dependency>
    			<groupId>org.aspectj</groupId>
    			<artifactId>aspectjweaver</artifactId>
    			<version>1.6.8</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.mybatis</groupId>
    			<artifactId>mybatis</artifactId>
    			<version>3.2.8</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.mybatis</groupId>
    			<artifactId>mybatis-spring</artifactId>
    			<version>1.2.2</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>log4j</groupId>
    			<artifactId>log4j</artifactId>
    			<version>1.2.17</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.slf4j</groupId>
    			<artifactId>slf4j-api</artifactId>
    			<version>1.7.7</version>
    		</dependency>
    
    		<dependency>
    			<groupId>mysql</groupId>
    			<artifactId>mysql-connector-java</artifactId>
    			<version>5.1.37</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context-support</artifactId>
    			<version>4.0.0.RELEASE</version>
    		</dependency>
    
    		<dependency>
    			<groupId>commons-collections</groupId>
    			<artifactId>commons-collections</artifactId>
    			<version>3.1</version>
    		</dependency>
    
    		<dependency>
    			<groupId>commons-codec</groupId>
    			<artifactId>commons-codec</artifactId>
    			<version>1.9</version>
    		</dependency>
    
    		<dependency>
    			<groupId>commons-io</groupId>
    			<artifactId>commons-io</artifactId>
    			<version>2.4</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.codehaus.jackson</groupId>
    			<artifactId>jackson-mapper-asl</artifactId>
    			<version>1.9.2</version>
    		</dependency>
    
    		<dependency>
    			<groupId>commons-fileupload</groupId>
    			<artifactId>commons-fileupload</artifactId>
    			<version>1.3.1</version>
    		</dependency>
    
    		<dependency>
    			<groupId>javax.servlet</groupId>
    			<artifactId>servlet-api</artifactId>
    			<version>2.5</version>
    		</dependency>
    
    		<dependency>
    			<groupId>javax.servlet.jsp</groupId>
    			<artifactId>jsp-api</artifactId>
    			<version>2.1.3-b06</version>
    		</dependency>
    
    		<dependency>
    			<groupId>jstl</groupId>
    			<artifactId>jstl</artifactId>
    			<version>1.2</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-test</artifactId>
    			<version>4.3.7.RELEASE</version>
    			<scope>test</scope>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>fastjson</artifactId>
    			<version>1.1.41</version>
    		</dependency>
    
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>druid</artifactId>
    			<version>1.0.9</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.aspectj</groupId>
    			<artifactId>aspectjweaver</artifactId>
    			<version>1.8.8</version>
    		</dependency>
    
    		
    		
    
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-jms</artifactId>
    			<version>4.0.0.RELEASE</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.apache.activemq</groupId>
    			<artifactId>activemq-core</artifactId>
    			<version>5.7.0</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.apache.activemq</groupId>
    			<artifactId>activemq-pool</artifactId>
    			<version>5.13.3</version>
    		</dependency>
    		
    	</dependencies>

大家可以根据自己需要，把主要的配置文件引入即可，整合spring的时候，倒数第三个是必须的；

log4j.properties老套路，这里不多解释，请看mq.properties，这个里面是连接linux中的activemq的主要配置信息，

## MQ
    mq.brokerURL=tcp\://192.168.111.130\:61616
    mq.userName=admin
    mq.password=admin
    mq.pool.maxConnections=10
    #queueName
    queueName=test_queue

spring-mq.xml，这个配置文件里，配置的项如ConnectionFactory，池化连接，activeMqJmsTemplate（队列模板）等，主要是方便项目代码中的连接使用，方便spring统一管理，

<?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:p="http://www.springframework.org/schema/p"
    	xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans  
               http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  
               http://www.springframework.org/schema/aop   
               http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  
               http://www.springframework.org/schema/tx  
               http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  
               http://www.springframework.org/schema/context  
               http://www.springframework.org/schema/context/spring-context-3.2.xsd"
    	default-autowire="byName" default-lazy-init="false">
    	
    	
    	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    		
            <property name="brokerURL" value="${mq.brokerURL}" />
            <property name="userName" value="${mq.userName}"></property>
            <property name="password" value="${mq.password}"></property> 
    	</bean>
    
        
    	<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
    		<property name="connectionFactory" ref="targetConnectionFactory" />
    		<property name="maxConnections" value="${mq.pool.maxConnections}" />
    	</bean>
    
    	
    	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
    		
    		<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
    	</bean>
    	
    	
    	
    	<bean id="activeMqJmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
    	      
    	    <property name="connectionFactory" ref="connectionFactory"/>  
    	    <property name="defaultDestinationName" value="${queueName}"></property>
    	</bean> 
    
    </beans>

接下来是和spring配置文件整合在一起，请看spring-context.xml配置文件，

里面主要包括，引入mq的配置文件，扫描包，加载spring-mq的配置文件，比较简单，相信大家一眼就能看明白；

下面是具体的代码，这里为了模拟生产者，使用简单的点对点的方式进行消息的发送；  
MailParam是个实体类，可以作为一个对象进行传输，

public class MailParam {
    	
    	/** 发件人 **/
    	private String from;
    	/** 收件人 **/
    	private String to;
    	/** 主题 **/
    	private String subject;
    	/** 邮件内容 **/
    	private String content;
    
    	public MailParam() {
    	}
    
    	public MailParam(String to, String subject, String content) {
    		this.to = to;
    		this.subject = subject;
    		this.content = content;
    	}
    
    	public String getFrom() {
    		return from;
    	}
    
    	public void setFrom(String from) {
    		this.from = from;
    	}
    
    	public String getTo() {
    		return to;
    	}
    
    	public void setTo(String to) {
    		this.to = to;
    	}
    
    	public String getSubject() {
    		return subject;
    	}
    
    	public void setSubject(String subject) {
    		this.subject = subject;
    	}
    
    	public String getContent() {
    		return content;
    	}
    
    	public void setContent(String content) {
    		this.content = content;
    	}
    }

MQProducer：

@Service("mqProducer")
    public class MQProducer {
    	
    	@Autowired
    	private JmsTemplate activeMqJmsTemplate;
    	
    	public void sendMessage(final MailParam mail){
    		activeMqJmsTemplate.send(new MessageCreator() {
    			
    			public Message createMessage(Session session) throws JMSException {
    				return session.createTextMessage(JSONObject.toJSONString(mail));
    			}
    		});
    	}
    	
    }

在test目录下创建一个测试类MQProducerTest，

public class MQProducerTest {
    	
    	private static final Log log = LogFactory.getLog(MQProducerTest.class);
    	
    	public static void main(String[] args) {
    		try {
    			
    			ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring/spring-context.xml");
    			context.start();
    			
    			MQProducer mqProducer = (MQProducer) context.getBean("mqProducer");
    			
    			MailParam mail = new MailParam();
    			mail.setTo("2639378441@qq.com");
    			mail.setSubject("ActiveMQ测试");
    			mail.setContent("通过ActiveMQ异步发送邮件！");
    			
    			mqProducer.sendMessage(mail);
    			
    		} catch (Exception e) {
    			log.error("==>MQ context start error:", e);
    			System.exit(0);
    		}finally{
    			log.info("===>System.exit");
    			System.exit(0);
    		}
    	}
    	
    	
    }

再看消费端，  
![在这里插入图片描述][70 4]

pom文件和producer基本一致，不再罗列，配置文件部门也差不多，从项目结构可以看出来，在和spring进行整合的时候，消费端常常需要设定一个监听器，进行异步的监听消费，这个和rocketmq或rabbitmq思想类似，log4j.properties不再罗列，mq.properties如下，

## MQ
    mq.brokerURL=tcp\://192.168.111.130\:61616
    mq.userName=admin
    mq.password=admin
    mq.pool.maxConnections=10
    #queueName
    queueName=test_queue

注意到，生产者个消费端都采用了连接池，用于管理并发的连接数，方便管理和控制，这一点请注意；

spring-mq.xml，这个和producer中类似，多加了消息监听器的配置部分，

<?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:p="http://www.springframework.org/schema/p"
    	xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans  
               http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  
               http://www.springframework.org/schema/aop   
               http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  
               http://www.springframework.org/schema/tx  
               http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  
               http://www.springframework.org/schema/context  
               http://www.springframework.org/schema/context/spring-context-3.2.xsd"
    	default-autowire="byName" default-lazy-init="false">
    
    	
    	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    		
            <property name="brokerURL" value="${mq.brokerURL}" />
            <property name="userName" value="${mq.userName}"></property>
            <property name="password" value="${mq.password}"></property> 
    	</bean>
    
        
    	<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
    		<property name="connectionFactory" ref="targetConnectionFactory" />
    		<property name="maxConnections" value="${mq.pool.maxConnections}" />
    	</bean>
    
    	
    	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
    		
    		<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
    	</bean>
    	
    	
    	
    	
    	<bean id="activeMqJmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
    	      
    	    <property name="connectionFactory" ref="connectionFactory"/>  
    	    <property name="defaultDestinationName" value="${queueName}"></property>
    	</bean> 
    	
    	
    	<bean id="sessionAwareQueue" class="org.apache.activemq.command.ActiveMQQueue">
    		<constructor-arg>
    			<value>${queueName}</value>
    		</constructor-arg>
    	</bean>
    
    	
    	<bean id="consumerSessionAwareMessageListener" class="wusc.edu.demo.mqtest.listener.ConsumerSessionAwareMessageListener"></bean>
    
    	<bean id="sessionAwareListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    		<property name="connectionFactory" ref="connectionFactory" />
    		<property name="destination" ref="sessionAwareQueue" />
    		<property name="messageListener" ref="consumerSessionAwareMessageListener" />
    	</bean>
    
    </beans>

spring-context.xml，用于引入外部配置文件并整合activemq的配置文件，

接下来，请看消息监听部分的代码，ConsumerSessionAwareMessageListener，这个类需要实现一个接口，SessionAwareMessageListener，

@Component
    public class ConsumerSessionAwareMessageListener implements SessionAwareMessageListener<Message> {
    
    	private static final Log log = LogFactory.getLog(ConsumerSessionAwareMessageListener.class);
    
    	@Autowired
    	private JmsTemplate activeMqJmsTemplate;
    	
    	@Autowired
    	private Destination sessionAwareQueue;
    	
    	@Autowired
    	private MailBiz bailBiz;
    	
    	public synchronized void onMessage(Message message, Session session) {
    		try {
    			ActiveMQTextMessage msg = (ActiveMQTextMessage) message;
    			final String ms = msg.getText();
    			log.info("==>receive message:" + ms);
    			MailParam mailParam = JSONObject.parseObject(ms, MailParam.class);// 转换成相应的对象
    			if (mailParam == null) {
    				return;
    			}
    
    			try {
    				System.out.println("收到的消息内容是==============>>>>>>>>>>>"+mailParam.getContent());
    				//bailBiz.mailSend(mailParam);
    			} catch (Exception e) {
    				// 发送异常，重新放回队列
    //				activeMqJmsTemplate.send(sessionAwareQueue, new MessageCreator() {
    //					public Message createMessage(Session session) throws JMSException {
    //						return session.createTextMessage(ms);
    //					}
    //				});
    				log.error("==>MailException:", e);
    			}
    		} catch (Exception e) {
    			log.error("==>", e);
    		}
    	}
    	
    }

最后是消费端的启动测试类，MQConsumer，

public class MQConsumer {
    	
    	private static final Log log = LogFactory.getLog(MQConsumer.class);
    
    	public static void main(String[] args) {
    		try {
    			ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring/spring-context.xml");
    			context.start();
    			System.out.println("consumer is starting.....");
    		} catch (Exception e) {
    			log.error("==>MQ context start error:", e);
    			System.exit(0);
    		}
    	}
    }

接下来就可以启动两个测试类，进行测试啦，首先启动消费端的代码，  
![在这里插入图片描述][70 5]

此时，消费者成功启动，消息监听器启动，等待接收消息，再启动生产者，  
![在这里插入图片描述][70 6]

此时，生产者也启动成功，并且成功发送一条消息到消息队列中，此时再看生产者端控制台输出的结果，  
![在这里插入图片描述][70 7]

再看管控台，  
![在这里插入图片描述][70 8]

可以看到，消费端成功收到来自生产者的消息，到此，spring整合activemq结束，需要代码的伙伴们可以下载：[https://download.csdn.net/download/zhangcongyi420/10687136][https_download.csdn.net_download_zhangcongyi420_10687136]

**大家还可以继续在此基础上尝试在ssm项目中整合activemq，原理基本类似，最后感谢大家观看！！！**

[70]: https://img-blog.csdn.net/20180925222734862?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[http_192.168.111.130_8161_admin]: http://192.168.111.130:8161/admin%EF%BC%8C%E5%8F%AF%E8%BF%9B%E5%85%A5%E7%AE%A1%E6%8E%A7%E5%8F%B0%E6%9F%A5%E7%9C%8B%EF%BC%8C
[70 1]: https://img-blog.csdn.net/20180925223005627?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 2]: https://img-blog.csdn.net/20180925223353300?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 3]: https://img-blog.csdn.net/20180925230129312?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 4]: https://img-blog.csdn.net/20180926080200760?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 5]: https://img-blog.csdn.net/20180926184424896?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 6]: https://img-blog.csdn.net/20180926184629240?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 7]: https://img-blog.csdn.net/20180926184817863?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 8]: https://img-blog.csdn.net/20180926185419532?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[https_download.csdn.net_download_zhangcongyi420_10687136]: https://download.csdn.net/download/zhangcongyi420/10687136