ActiveMQ PTP/发布订阅者模式及与Spring结合（四）

本是古典何须时尚 2021-09-27 01:30 277阅读 0赞

**PTP**

PTP的过程理解起来简单。它好比两个人打电话，这两个人是独享这一条通信链路的。一方是发送消息，另外一方接收，就这么简单。在实际应用中因为有多个用户对使用PTP的链路，他的通信场景如下

![70][]

**Publish-Subscribe**

发布订阅模式有点类似于我们日常生活中订阅报纸。每年到年尾的时候，邮局就会发一本报纸集合让我们来选择订阅哪一个。在这个表里头裂了所有出版发行的报纸，那么对于我们每一个订阅的报纸就相当于发布订阅模式里的topic。有很多个人订阅报纸，也有人可能和我订阅了相同的报纸。那么在这里相当于我们在同一个topic里注册了。对于一份报纸发行方来说，它和所有的订阅者构成了一个一对多的关系。这种关系如下图所示：

![70 1][]

**activeMQ与spring进行整合**

activeMQ可以很轻松的与spring进行整合，Sping提供了一系列的接口类，非常的好用。

我们接下来使用spring框架整合activeMQ，里有消息中间件，异步处理任务的机制，比如异步消费数据、异步发送邮件、异步做查询操作等。

**Coding：**

**activemq-consumer**

使用的是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/maven-v4_0_0.xsd">
 <modelVersion>4.0.0</modelVersion>
 <groupId>activemq</groupId>
 <artifactId>activemq-consumer</artifactId>
 <packaging>war</packaging>
 <version>0.0.1-SNAPSHOT</version>
 <name>activemq-consumer Maven Webapp</name>
 <url>http://maven.apache.org</url>
 
 <repositories> 
 <repository> 
 <snapshots> 
 <enabled>true</enabled> 
 </snapshots> 
 <id>public</id> 
 <name>Public Repositories</name> 
 <url>http://localhost:8081/nexus/content/groups/public/</url> 
 </repository> 
 </repositories> 
 <pluginRepositories> 
 <pluginRepository> 
 <id>public</id> 
 <name>Public Repositories</name> 
 <url>http://localhost:8081/nexus/content/groups/public/</url> 
 </pluginRepository> 
 </pluginRepositories> 
 
 
 <properties>
 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 <junit.version>4.11</junit.version>
 <spring.version>3.2.0.RELEASE</spring.version>
 <httpclient.version>4.3.1</httpclient.version>
 </properties> 
 
 
 <dependencies>
 	<dependency>
 		<groupId>org.springframework</groupId>
 		<artifactId>spring-test</artifactId>
 		<version>${spring.version}</version>
 		<scope>test</scope> 
 	</dependency>
 	<dependency>
 		<groupId>org.springframework</groupId>
 		<artifactId>spring-webmvc</artifactId>
 		<version>${spring.version}</version>
 	</dependency>
 	<dependency>
 		<groupId>org.springframework</groupId>
 		<artifactId>spring-core</artifactId>
 		<version>${spring.version}</version>
 	</dependency>
 	<dependency>
 		<groupId>org.springframework</groupId>
 		<artifactId>spring-context</artifactId>
 		<version>${spring.version}</version>
 	</dependency>		
 	<dependency>
 		<groupId>org.springframework</groupId>
 		<artifactId>spring-context-support</artifactId>
 		<version>${spring.version}</version>
 	</dependency>	
 	<dependency>
 		<groupId>org.springframework</groupId>
 		<artifactId>spring-jms</artifactId>
 		<version>${spring.version}</version>
 	</dependency>
 	<dependency>
 		<groupId>org.springframework</groupId>
 		<artifactId>spring-oxm</artifactId>
 		<version>${spring.version}</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.1.1</version>
 	</dependency> 
 
 	<dependency>
 		<groupId>org.apache.activemq</groupId>
 		<artifactId>activemq-all</artifactId>
 		<version>5.11.1</version>
 	</dependency>
 	<dependency>
 		<groupId>org.apache.activemq</groupId>
 		<artifactId>activemq-pool</artifactId>
 		<version>5.11.1</version>
 	</dependency>
 	<dependency>
 		<groupId>com.alibaba</groupId>
 		<artifactId>druid</artifactId>
 		<version>0.2.9</version>
 	</dependency>
 	<dependency>
 		<groupId>org.aspectj</groupId>
 		<artifactId>aspectjweaver</artifactId>
 		<version>1.7.1</version>
 	</dependency>		
 	<dependency>
 		<groupId>org.codehaus.jackson</groupId>
 		<artifactId>jackson-mapper-asl</artifactId>
 		<version>1.9.11</version>
 	</dependency>
 	<dependency>
 		<groupId>commons-fileupload</groupId>
 		<artifactId>commons-fileupload</artifactId>
 		<version>1.2.2</version>
 	</dependency>
 	<dependency>
 		<groupId>org.apache.commons</groupId>
 		<artifactId>commons-lang3</artifactId>
 		<version>3.3.1</version>
 	</dependency>
 	<dependency>
 		<groupId>commons-io</groupId>
 		<artifactId>commons-io</artifactId>
 		<version>2.4</version>
 	</dependency>
 	<dependency>
 		<groupId>org.apache.httpcomponents</groupId>
 		<artifactId>httpclient-cache</artifactId>
 		<version>${httpclient.version}</version>
 	</dependency>
 	<dependency>
 		<groupId>org.apache.httpcomponents</groupId>
 		<artifactId>httpclient</artifactId>
 		<version>${httpclient.version}</version>
 	</dependency>
 	<dependency>
 		<groupId>org.apache.httpcomponents</groupId>
 		<artifactId>httpcore</artifactId>
 		<version>${httpclient.version}</version>
 	</dependency>	
 	<dependency>
 		<groupId>com.alibaba</groupId>
 		<artifactId>fastjson</artifactId>
 		<version>1.1.26</version>
 	</dependency> 
 	<dependency>
 		<groupId>org.slf4j</groupId>
 		<artifactId>slf4j-api</artifactId>
 		<version>1.7.9</version>
 	</dependency>
 	<dependency>
 		<groupId>org.slf4j</groupId>
 		<artifactId>jul-to-slf4j</artifactId>
 		<version>1.7.6</version>
 	</dependency>
 	<dependency>
 		<groupId>org.slf4j</groupId>
 		<artifactId>jcl-over-slf4j</artifactId>
 		<version>1.7.6</version>
 	</dependency>
 	<dependency>
 		<groupId>javax.servlet</groupId>
 		<artifactId>jstl</artifactId>
 		<version>1.2</version>
 	</dependency>
 	<dependency>
 		<groupId>javax.servlet</groupId>
 		<artifactId>servlet-api</artifactId>
 		<version>3.0-alpha-1</version>
 		<scope>provided</scope>
 	</dependency>
 	<dependency> 
 <groupId>com.github.miemiedev</groupId> 
 <artifactId>mybatis-paginator</artifactId> 
 <version>1.2.10</version> 
 </dependency>
 	<dependency> 
 		<groupId>com.ojdbc7</groupId>
 		<artifactId>ojdbc7</artifactId>
 		<version>1.0</version>
 	</dependency>	
 	<dependency>
 		<groupId>mysql</groupId>
 		<artifactId>mysql-connector-java</artifactId>
 		<version>5.1.21</version>
 	</dependency>
 	<dependency>
 		<groupId>org.jboss.marshalling</groupId>
 		<artifactId>jboss-marshalling</artifactId>
 		<version>1.3.0.CR9</version>
 	</dependency> 			
 	<dependency>
 		<groupId>org.jboss.marshalling</groupId>
 		<artifactId>jboss-marshalling-serial</artifactId>
 		<version>1.3.0.CR9</version>
 	</dependency>		
 	<dependency>
 		<groupId>junit</groupId>
 		<artifactId>junit</artifactId>
 		<version>${junit.version}</version>
 		<scope>test</scope>
 	</dependency>	
 		<dependency>
 		<groupId>org.apache.curator</groupId>
 		<artifactId>curator-framework</artifactId>
 		<version>2.4.2</version>
 	</dependency> 	
 		<dependency>
 		<groupId>org.apache.curator</groupId>
 		<artifactId>curator-recipes</artifactId>
 		<version>2.4.2</version>
 	</dependency>
 
 </dependencies>
 
 <build>
 <finalName>activemq-consumer</finalName>
 </build>
 </project>

config.properties

## ActiveMQ Configuration
    activemq.brokerURL=tcp\://192.168.1.200\:61616
    activemq.userName=xxx
    activemq.password=xxx
    activemq.pool.maxConnections=10
    #queueName
    activemq.queueName=mailqueue
    
    ## SMTP Configuration
    mail.host=smtp.163.com
    ##mail.port=21
    mail.username=***@163.com
    mail.password=
    mail.smtp.auth=true
    mail.smtp.timeout=30000
    mail.default.from=***@163.com

spring-activemq.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: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="${activemq.brokerURL}" />
 <property name="userName" value="${activemq.userName}"></property>
 <property name="password" value="${activemq.password}"></property> 
 	</bean>
 	
 
 	<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
 		<property name="connectionFactory" ref="targetConnectionFactory" />
 		<property name="maxConnections" value="${activemq.pool.maxConnections}" />
 	</bean>
 
 	
 	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
 		
 		<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
 	</bean>
 	
 	
 	
 	
 	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> 
 	  
 	 <property name="connectionFactory" ref="connectionFactory"/> 
 	 <property name="defaultDestinationName" value="${activemq.queueName}"></property>
 	</bean> 
 	
 	
 	<bean id="mailQueue" class="org.apache.activemq.command.ActiveMQQueue">
 		<constructor-arg>
 			<value>${activemq.queueName}</value>
 		</constructor-arg>
 	</bean>
 
 	
 	<bean id="mailQueueMessageListener" class="bhz.mq.MailQueueMessageListener"></bean>
 
 	
 	<bean id="sessionAwareListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
 		<property name="connectionFactory" ref="connectionFactory" />
 		<property name="destination" ref="mailQueue" />
 		<property name="messageListener" ref="mailQueueMessageListener" />
 	</bean>
 
 
 
 </beans>

spring-context.xml

spring-mail.xml

Mail.java

package bhz.entity;
    
    public class Mail {
    
    	/** 发件人 **/
    	private String from;
    	/** 收件人 **/
    	private String to;
    	/** 主题 **/
    	private String subject;
    	/** 邮件内容 **/
    	private String content;
    	
    	public Mail(){}
    	
    	public Mail(String from, String to, String subject, String content) {
    		super();
    		this.from = from;
    		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;
    	}
    }

MailQueueMessageListener.java

package bhz.mq;
 
 import javax.jms.Destination;
 import javax.jms.Message;
 import javax.jms.Session;
 import javax.jms.TextMessage;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.jms.listener.SessionAwareMessageListener;
 import org.springframework.stereotype.Component;
 
 import bhz.entity.Mail;
 import bhz.service.MailService;
 
 import com.alibaba.fastjson.JSONObject;
 
 /**
 * 系统名称： 
 * 模块名称： 
 * 中文类名： 
 * 概要说明： 
 */
 @Component
 public class MailQueueMessageListener implements SessionAwareMessageListener<Message> {
 
 	@Autowired
 	private JmsTemplate jmsTemplate;
 	@Autowired
 	private Destination mailQueue;
 	@Autowired
 	private MailService mailService;
 
 	public synchronized void onMessage(Message message, Session session) {
 		try {
 			TextMessage msg = (TextMessage) message;
 			final String ms = msg.getText();
 			System.out.println("收到消息：" + ms);
 			//转换成相应的对象
 			Mail mail = JSONObject.parseObject(ms, Mail.class);
 			if (mail == null) {
 				return;
 			}
 			try {
 				//执行发送业务
 				mailService.mailSend(mail);
 				
 			} catch (Exception e) {
 				// 发送异常，重新放回队列
 //				jmsTemplate.send(mailQueue, new MessageCreator() {
 //					@Override
 //					public Message createMessage(Session session) throws JMSException {
 //						return session.createTextMessage(ms);
 //					}
 //				});
 				e.printStackTrace();
 			}
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 	}
 }

MailService.java

package bhz.service;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.mail.MailException;
 import org.springframework.mail.SimpleMailMessage;
 import org.springframework.mail.javamail.JavaMailSender;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Service;
 
 import bhz.entity.Mail;
 
 @Service("mailService")
 public class MailService {
 
 	@Autowired
 	private JavaMailSender mailSender;
 	@Autowired
 	private SimpleMailMessage simpleMailMessage;
 	@Autowired
 	private ThreadPoolTaskExecutor threadPool;
 
 	/**
 	 * 方法名称：发送邮件 
 	 * 概要说明：发送邮件 
 	 * @param mail
 	 */
 	public void mailSend(final Mail mail) {
 		threadPool.execute(new Runnable() {
 			public void run() {
 				try {
 					simpleMailMessage.setFrom(simpleMailMessage.getFrom()); 
 					simpleMailMessage.setTo(mail.getTo()); 
 					simpleMailMessage.setSubject(mail.getSubject());
 					simpleMailMessage.setText(mail.getContent());
 					mailSender.send(simpleMailMessage);
 				} catch (MailException e) {
 					e.printStackTrace();
 					throw e;
 				}
 			}
 		});
 	}
 }

TestConsumer.java

package bhz.test;
 
 import org.springframework.context.support.ClassPathXmlApplicationContext;
 
 /**
 * 系统名称： 
 * 模块名称： 
 * 中文类名： 
 * 概要说明： 
 */
 public class TestConsumer {
 
 	public static void main(String[] args) {
 		try {
 			ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[] { "spring-context.xml" });
 			context.start();
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
 	}
 }

activemq-provider

![70 2][]

以上配置文件里面的内容和activemq-consumer一样。

MQProducer.java

package bhz.mq;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.Session;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.jms.core.MessageCreator;
 import org.springframework.stereotype.Service;
 
 import bhz.entity.Mail;
 
 import com.alibaba.fastjson.JSONObject;
 
 
 /**
 * 系统名称： 
 * 模块名称： 
 * 中文类名： 
 * 概要说明： 
 */
 @Service("mqProducer")
 public class MQProducer {
 	
 	private JmsTemplate jmsTemplate;
 	
 	public JmsTemplate getJmsTemplate() {
 		return jmsTemplate;
 	}
 	
 	@Autowired
 	public void setJmsTemplate(JmsTemplate jmsTemplate) {
 		this.jmsTemplate = jmsTemplate;
 	}
 
 	/**
 	 * 方法名称：发送邮件信息对象 
 	 * 概要说明：发送邮件信息对象 
 	 * @param mail
 	 */
 	public void sendMessage(final Mail mail) {
 		jmsTemplate.send(new MessageCreator() {
 			public Message createMessage(Session session) throws JMSException {
 				return session.createTextMessage(JSONObject.toJSONString(mail));
 			}
 		});
 	}
 
 }

TestProducer.java

package bhz.test;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 import bhz.entity.Mail;
 import bhz.mq.MQProducer;
 
 
 /**
 * 系统名称： 
 * 模块名称： 
 * 中文类名： 
 * 概要说明： 
 */
 @ContextConfiguration(locations = {"classpath:spring-context.xml" })
 @RunWith(SpringJUnit4ClassRunner.class)
 public class TestProducer {
 	
 	@Autowired
 	private MQProducer mqProducer;
 	
 	@Test
 	public void send(){
 		Mail mail = new Mail();
 		mail.setTo("174754613@qq.com");
 		mail.setSubject("异步发送邮件");
 		mail.setContent("Hi,This is a message!");
 														
 		this.mqProducer.sendMessage(mail);
 		System.out.println("发送成功..");		
 		
 	}
 
 }

[70]: /images/20210923/02bfd35903774b509bc4cb24311d42b6.png
[70 1]: /images/20210923/502201cab793436bbba45d17809bdc23.png
[70 2]: /images/20210923/55799c999b3543fd85f097f8900a080d.png