SpringBoot入门建站全系列（十七）整合ActiveMq（JMS类消息队列）

￡神魔★判官ぃ 2021-12-09 03:17 249阅读 0赞

## SpringBoot入门建站全系列（十七）整合ActiveMq（JMS类消息队列） ##

#### 一、概述 ####

消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型，它可以在分布式环境下扩展进程间的通信。对于消息中间件，常见的角色大致也就有Producer（生产者）、Consumer（消费者）

常见的消息中间件产品：

（1）ActiveMQ

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。这里介绍的是ActiveMQ的使用。

（2）RabbitMQ

AMQP协议的领导实现，支持多种场景。淘宝的MySQL集群内部有使用它进行通讯，OpenStack开源云平台的通信组件，最先在金融行业得到运用。

（3）ZeroMQ

史上最快的消息队列系统

（4）Kafka

Apache下的一个子项目 。特点：高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统。适合处理海量数据。

**Jms**

JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。

**AMQP**

AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。意味着我们可以使用Java的AMQP provider，同时使用一个python的producer加一个rubby的consumer。从这一点看，AQMP可以用http来进行类比，不关心实现的语言，只要大家都按照相应的数据格式去发送报文请求，不同语言的client均可以和不同语言的server链接。

首发地址：  
[品茗IT-同步发布][IT-]

[品茗IT][IT] 提供在线支持：

[一键快速构建Spring项目工具][Spring]

[一键快速构建SpringBoot项目工具][SpringBoot]

[一键快速构建SpringCloud项目工具][SpringCloud]

[一站式Springboot项目生成][Springboot]

[Mysql一键生成Mybatis注解Mapper][Mysql_Mybatis_Mapper]

代码可以在SpringBoot组件化构建[https://www.pomit.cn/java/spring/springboot.html][SpringBoot]中的ActiveMQ组件中查看，并下载。

**如果大家正在寻找一个java的学习环境，或者在开发中遇到困难，可以[加入我们的java学习圈，点击即可加入][java]，共同学习，节约学习时间，减少很多在学习中遇到的难题。**

### 二、配置 ###

本文假设你已经引入spring-boot-starter-web。已经是个SpringBoot项目了,如果不会搭建，可以打开这篇文章看一看[《SpringBoot入门建站全系列（一）项目建立》][SpringBoot 1]。

#### 2.1 Maven依赖 ####

使用activemq可以使用spring-boot-starter-activemq，方便快捷，一般springboot对大多数开源项目都做了整合，提供了专用的stater。

<dependency>
    	<groupId>org.springframework.boot</groupId>
    	<artifactId>spring-boot-starter-activemq</artifactId>
    </dependency>

#### 2.2 配置文件 ####

在application.properties 中需要配置activemq的信息，也可以配置自定义的配置，如：

spring.activemq.broker-url=tcp://localhost:61616
    #spring.activemq.user=admin
    #spring.activemq.password=secret
    
    jms.destQueueName=destQueue

这里面，

*  spring.activemq.broker-url是springboot自动装配的配置，activemq的地址。
 *  spring.activemq.user是springboot自动装配的配置，activemq的用户名，一般自己测试都不会去设置用户名密码的。
 *  spring.activemq.password是springboot自动装配的配置，，activemq的密码，一般自己测试都不会去设置用户名密码的。
 *  jms.destQueueName，这是我自己定义的一个目的队列名。

### 三、ActiveMQ的使用 ###

#### 3.1 配置数据转换 ####

ActiveMQ是分为生产者和消费者的，生产者生产的消息，如何能够被消费者正常解析，需要开发者自己对数据转换做定义，如果你非要说，我用字符串也可以啊，那当这个不存在就行了。

下面这个配置是指明了ActiveMQ的数据转换是用MappingJackson2MessageConverter，将json数据转换为对象，或者将对象转换为json。

ActiveMQConfig :

package com.cff.springbootwork.activemq;
    
    import javax.jms.ConnectionFactory;
    
    import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
    import org.springframework.jms.config.JmsListenerContainerFactory;
    import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
    import org.springframework.jms.support.converter.MessageConverter;
    import org.springframework.jms.support.converter.MessageType;
    
    @Configuration
    public class ActiveMQConfig { 
    
    	@Bean
    	public JmsListenerContainerFactory<?> myFactory(ConnectionFactory connectionFactory,
    			DefaultJmsListenerContainerFactoryConfigurer configurer) { 
    		DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    		configurer.configure(factory, connectionFactory);
    		factory.setMessageConverter(jacksonJmsMessageConverter());
    		return factory;
    	}
    
    	@Bean
    	public MessageConverter jacksonJmsMessageConverter() { 
    		MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
    		converter.setTargetType(MessageType.TEXT);
    		converter.setTypeIdPropertyName("_type");
    		return converter;
    	}
    }

#### 3.2 生产者 ####

ActiveMQ当然需要生产者来生产信息，然后才发送到消息队列的。

JmsProducer：

package com.cff.springbootwork.activemq.handler;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.stereotype.Component;
    
    import com.cff.springbootwork.activemq.model.DefaultMqModel;
    
    @Component
    public class JmsProducer { 
    	@Autowired
    	JmsTemplate jmsTemplate;
    
    	@Value("${jms.destQueueName}")
    	String destQueueName;
    
    	public void send(DefaultMqModel defaultMqModel) { 
    		jmsTemplate.convertAndSend(destQueueName, defaultMqModel);
    	}
    }

这里的JmsTemplate 是Spring自动生成的bean。destQueueName注入的是配置文件中定义的目的队列。然后发送数据。

测试发送数据,ActiveMQRest :

package com.cff.springbootwork.activemq.web;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.RestController;
    
    import com.cff.springbootwork.activemq.handler.JmsProducer;
    import com.cff.springbootwork.activemq.model.DefaultMqModel;
    import com.cff.springbootwork.activemq.model.SeccondMqModel;
    
    @RestController
    @RequestMapping("/activemq")
    public class ActiveMQRest { 
    	@Autowired
    	JmsProducer jmsProducer;
    
    	@RequestMapping(value = "/test", method = {  RequestMethod.GET })
    	public DefaultMqModel test() { 
    		DefaultMqModel defaultMqModel = new DefaultMqModel();
    		defaultMqModel.setContent("hahahahahahhahaha哈啊哈哈");
    		defaultMqModel.setTitle("测试");
    		defaultMqModel.setType(1);
    		jmsProducer.send(defaultMqModel);
    		return defaultMqModel;
    	}
    
    	@RequestMapping(value = "/test2", method = {  RequestMethod.GET })
    	public SeccondMqModel test2() { 
    		SeccondMqModel seccondMqModel = new SeccondMqModel();
    		seccondMqModel.setContent("asdasdasd哈啊哈哈");
    		seccondMqModel.setTitle("测试2");
    		seccondMqModel.setRemark("第二个");
    		seccondMqModel.setType(2);
    		jmsProducer.send(seccondMqModel);
    		return seccondMqModel;
    	}
    }

#### 3.3 消费者 ####

消费者监听指定的队列，需要用@JmsListener注解标明它是一个消费者。

JmsConsumer :

package com.cff.springbootwork.activemq.handler;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.stereotype.Component;
    
    import com.cff.springbootwork.activemq.model.DefaultMqModel;
    import com.cff.springbootwork.activemq.service.BusinessSerivce;
    
    @Component
    public class JmsConsumer { 
    
    	@Autowired
    	BusinessSerivce businessSerivce;
    
    	@JmsListener(destination = "${jms.destQueueName}")
    	public void processMessage(DefaultMqModel defaultMqModel) { 
    		businessSerivce.doBusiness(defaultMqModel);
    	}
    }

这里，@JmsListener注解表明它在监听我们配置文件中配置的队列，方法的参数可以直接传入对象，因为我们在3.1中配置了统一的MessageConverter，它可以自动解析成对象。

BusinessSerivce是我自定义的一个业务逻辑处理service.

BusinessSerivce:

package com.cff.springbootwork.activemq.service;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Service;
    
    import com.cff.springbootwork.activemq.model.DefaultMqModel;
    import com.cff.springbootwork.activemq.model.SeccondMqModel;
    
    @Service
    public class BusinessSerivce { 
    	private final Logger log = LoggerFactory.getLogger(this.getClass());
    
    	public void doBusiness(DefaultMqModel defaultMqModel) { 
    		log.info(defaultMqModel.toString());
    		
    		if(defaultMqModel.getType() == 2 && defaultMqModel instanceof SeccondMqModel){ 
    			SeccondMqModel seccondMqModel = (SeccondMqModel) defaultMqModel;
    			log.info(seccondMqModel.remark);
    		}
    	}
    }

### 四、过程中用到的实体 ###

DefaultMqModel：

SeccondMqModel:

详细完整的实体，可以访问[品茗IT-博客《SpringBoot入门建站全系列（十七）整合ActiveMq（JMS类消息队列）》][IT-]进行查看

### 快速构建项目 ###

[Spring组件化构建][Spring]

[SpringBoot组件化构建][SpringBoot]

[SpringCloud服务化构建][SpringCloud]

喜欢这篇文章么，喜欢就加入我们一起讨论SpringBoot使用吧！  
![品茗IT交流群][IT 1]

[IT-]: https://www.pomit.cn/p/1986810777889281
[IT]: https://www.pomit.cn
[Spring]: https://www.pomit.cn/java/spring/spring.html
[SpringBoot]: https://www.pomit.cn/java/spring/springboot.html
[SpringCloud]: https://www.pomit.cn/java/spring/springcloud.html
[Springboot]: https://www.pomit.cn/java/java/sqlToSpringboot.html
[Mysql_Mybatis_Mapper]: https://www.pomit.cn/java/java/sqlToMybatis.html
[java]: https://jq.qq.com/?_wv=1027&k=52sgH1J
[SpringBoot 1]: https://www.pomit.cn/p/169774771175680
[IT 1]: /images/20211209/71cff07afe2948158c6a2e05d47467e4.png