SpringBoot整合JMS

文章目录

SpringBoot整合JMS
前言
一、JMS、AMQP概念以及区别
二、ActiveMQ安装
三、SpringBoot整合ActiveMQ：队列模式实例
- 1、创建demo项目
- 2、添加依赖
- 3、添加application配置
- 4、添加配置文件ActiveMQConfig
- 5、创建JMS组件文件
- 6、发送消息控制器
- 7、测试验证
四、SpringBoot整合ActiveMQ：广播模式实例
- 1、创建demo项目
- 2、添加依赖
- 3、添加application配置
- 4、添加配置文件ActiveMQConfig
- 5、创建JMS组件文件
- 6、发送消息控制器
- 7、测试验证
- 8、配置同时支持队列模式和广播模式

前言

消息队列（Message Queue）是一种进程间或者线程间的异步通信方式。使用消息队列，消息生产者会将消息保存在消息队列中，知道消息消费者来取走它。实现服务的解耦合，并提高系统的可靠性和扩展性。
目前常用的开源消息队列有很多，RabbitMQ、ActiveMQ、Redis、Kafka等，也就是常说的消息中间件。
本篇文章以Apache ActiveMQ为例，实战整合ActiveMQ的队列模式和广播模式两种以及兼容两种的实战。

一、JMS、AMQP概念以及区别

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

JMS包括两种消息模型：点对点、发布者/订阅者；
点对点式： – 消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获取消息内容，消息读取后被移出队列 – 消息只有唯一的发送者和接受者，但并不是说只能有一个接收者
发布订阅式： – 发送者（发布者）发送消息到主题，多个接收者（订阅者）监听（订阅）这个主题，那么就会在消息到达时同时收到消息

JMS仅支持JAVA平台，不支持跨平台、跨语言。

AMQP

AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。例如RabbitMQ消息中间件。

两者区别

JMS是定义了统一的接口，来对消息操作进行统一；
AMQP是通过规定协议来统一数据交互的格式，
JMS限定了必须使用Java语言；
AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模型；而AMQP的消息模型更加丰富。

二、ActiveMQ安装

wget https://archive.apache.org/dist/activemq/5.16.0/apache-activemq-5.16.0-bin.tar.gz
mv apache-activemq-5.16.0-bin.tar.gz /usr/local/
cd /usr/local/
tar -zxvf apache-activemq-5.16.0-bin.tar.gz apache-activemq-5.16.0
chmod 777 apache-activemq-5.16.0 <==为目录设置权限
cd apache-activemq-5.16.0/bin/
chmod 755 activemq <==为命令设置权限
./activemq start

默认端口8161,访问http://192.168.138.135:8161/

1、遇到问题
http://192.168.138.135:8161/无法访问
现象描述
远程telnet 192.168.138.135 8161不通
本机telnet 192.168.138.135 8161不通
本机telnet 127.0.0.1 8161通
解决办法
/usr/local/apache-activemq-5.16.0/conf中jetty.xml中，将下图中127.0.0.1–》改成0.0.0.0.，允许所有ip访问，重启服务即可。

ActiveMQ启动成功后访问http://192.168.138.135:8161/时，默认用户密码都是admin

单机Manage ActiveMQ broker进入控制台

Linux知识点
linux 命令chmod 755的意思
chmod是Linux下设置文件权限的命令，后面的数字表示不同用户或用户组的权限。
一般是三个数字：
第一个数字表示文件所有者的权限
第二个数字表示与文件所有者同属一个用户组的其他用户的权限
第三个数字表示其它用户组的权限。
权限分为三种：读（r=4），写（w=2），执行（x=1）。综合起来还有可读可执行（rx=5=4+1）、可读可写（rw=6=4+2）、可读可写可执行(rwx=7=4+2+1)。
所以，chmod 755 设置用户的权限为：
1.文件所有者可读可写可执行 –7
2.与文件所有者同属一个用户组的其他用户可读可执行 –5
3.其它用户组可读可执行

三、SpringBoot整合ActiveMQ：队列模式实例

1、创建demo项目

https://start.spring.io/无法访问，通过https://start.springboot.io/生成一个demo项目

2、添加依赖

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

3、添加application配置

server.port=8001
spring.activemq.broker-url=tcp://192.168.138.135:61616
##配置信赖所有包，支持发送对象消息
spring.activemq.packages.trust-all=true
spring.activemq.user=admin
spring.activemq.password=admin
queueName=publish.queue

4、添加配置文件ActiveMQConfig

@Configuration
public class ActiveMQConfig {@Value("${queueName}")private String queueName;@Beanpublic Queue queue() {return new ActiveMQQueue(queueName);}}

5、创建JMS组件文件

package com.yangxf.demoActiveMQ.component;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Component;import javax.jms.Queue;/*** 〈一句话功能简述〉<br> * 〈JMS组件发送和接收消息〉** @author linwd* @create 2021/4/11* @since 1.0.0*/
@Component
public class JmsComponent {@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;@Autowiredprivate Queue queue;/*** 发送消息*/public void send(Message message){jmsMessagingTemplate.convertAndSend(this.queue,message.toString());}/*** 接收消息*/@JmsListener(destination= "${queueName}")    //用这个注解去监听 监听的队列public void receive(String messageString){System.out.println("receive:"+messageString);}}/*** 〈一句话功能简述〉<br>* 〈封装类〉** @author linwd* @create 2021/4/11* @since 1.0.0*/
public class Message implements Serializable {private String content;private Date data;...set,get方法@Overridepublic String toString() {return "Message{" +"content='" + content + '\'' +", data=" + data +'}';}
}

6、发送消息控制器

/*** 〈一句话功能简述〉<br> * 〈发送信息控制器〉** @author linwd* @create 2021/4/11* @since 1.0.0*/
@RestController
public class JmsController {private final Logger logger=LoggerFactory.getLogger(JmsController.class);@Autowiredprivate JmsComponent jmsComponent;@GetMapping("/pushMessage")public void send(){logger.info("pushMessage推送消息");Message message=new Message();message.setContent("推送消息");message.setData(new Date());jmsComponent.send(message);}
}

7、测试验证

在这里插入图片描述

四、SpringBoot整合ActiveMQ：广播模式实例

1、创建demo项目

https://start.spring.io/无法访问，通过https://start.springboot.io/生成一个demo项目

2、添加依赖

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

3、添加application配置

server.port=8000
spring.activemq.broker-url=tcp://192.168.138.135:61616
##配置信赖所有包，支持发送对象消息
spring.activemq.packages.trust-all=true
spring.activemq.user=admin
spring.activemq.password=admin
queueName=publish.queue
##topicid
topicName=publish.topic
#Spring Boot集成ActiveMQ时默认只支持队列或者广播之一，通过配置项spring.jms.pub-sub-domain来指定，
# true 为广播模式，false为队列模式，默认情况下支持队列模式。
spring.jms.pub-sub-domain=true

4、添加配置文件ActiveMQConfig

@Configuration
public class ActiveMQConfig {@Value("${queueName}")private String queueName;@Value("${topicName}")private String topicName;/*** 队列模式实例** @return*/@Beanpublic Queue queue() {return new ActiveMQQueue(queueName);}/*** 订阅模式实例** @return*/@Beanpublic Topic topic() {return new ActiveMQTopic(topicName);}}

5、创建JMS组件文件

package com.yangxf.demoActiveMQ.component;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Component;import javax.jms.Queue;
import javax.jms.Topic;@Component
public class JmsComponent {@Autowiredprivate JmsMessagingTemplate jmsMessagingTemplate;@Autowiredprivate Queue queue;@Autowiredprivate Topic topic;/*** 发送消息*/public void send(Message message){jmsMessagingTemplate.convertAndSend(this.queue,message.toString());}/*** 接收消息*/@JmsListener(destination= "${queueName}")    //用这个注解去监听 监听的队列public void receive(String messageString){System.out.println("receive:"+messageString);}/*** 广播发送消息*/public void sendForTopic(Message message){jmsMessagingTemplate.convertAndSend(this.topic,message.toString());}/*** 接收消息*/@JmsListener(destination= "${topicName}")public void receiveForTopic1(String messageString){System.out.println("receive1:"+messageString);}/*** 接收消息*/@JmsListener(destination= "${topicName}")public void receiveForTopic2(String messageString){System.out.println("receive2:"+messageString);}}/*** 〈一句话功能简述〉<br>* 〈封装类〉** @author linwd* @create 2021/4/11* @since 1.0.0*/
public class Message implements Serializable {private String content;private Date data;...set,get方法@Overridepublic String toString() {return "Message{" +"content='" + content + '\'' +", data=" + data +'}';}
}

6、发送消息控制器

/*** 〈一句话功能简述〉<br> * 〈发送信息控制器〉** @author linwd* @create 2021/4/11* @since 1.0.0*/
@RestController
public class JmsController {private final Logger logger=LoggerFactory.getLogger(JmsController.class);@Autowiredprivate JmsComponent jmsComponent;@GetMapping("/pushMessage")public void send(){logger.info("pushMessage推送消息");Message message=new Message();message.setContent("推送消息");message.setData(new Date());jmsComponent.send(message);}@GetMapping("/pushMessageForTopic")public void pushMessageForTopic(){logger.info("pushMessageForTopic广播消息");Message message=new Message();message.setContent("广播消息");message.setData(new Date());jmsComponent.sendForTopic(message);}
}

7、测试验证

复制一份项目，改一下端口两个同时启动，访问其中一个服务的广播发送短信地址:http://localhost:8001/pushMessageForTopic

服务1
服务2

两个服务同时接收到信息。
但是这种情况下http://localhost:8001/pushMessage无法收到信息，即不能兼容支持队列模式。

8、配置同时支持队列模式和广播模式

-简单的springboot配置无法支持这个功能，通过java配置完成
将ActiveMQConfig配置改成如下

package com.yangxf.demoActiveMQ.config;import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Value;
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 javax.jms.ConnectionFactory;
import javax.jms.Queue;
import javax.jms.Topic;/*** 〈一句话功能简述〉<br>* 〈ActiveMQConfig配置〉** @author linwd* @create 2021/4/11* @since 1.0.0*/
@Configuration
public class ActiveMQConfig {@Value("${spring.activemq.user}")private String usrName;@Value("${spring.activemq.password}")private  String password;@Value("${spring.activemq.broker-url}")private  String brokerUrl;@Value("${queueName}")private String queueName;@Value("${topicName}")private String topicName;/*** 队列模式实例** @return*/@Beanpublic Queue queue() {return new ActiveMQQueue(queueName);}/*** 订阅模式实例** @return*/@Beanpublic Topic topic() {return new ActiveMQTopic(topicName);}/*** 配置以下两个bean，同时支持队列模式和广播模式，配置中spring.jms.pub-sub-domain=true将失效* @return*/@Bean("connectionFactory")public ActiveMQConnectionFactory connectionFactory() {return new ActiveMQConnectionFactory(usrName, password, brokerUrl);}@Bean("queueListenerFactory")public JmsListenerContainerFactory<?> queueListenerFactory(ConnectionFactory connectionFactory) {DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setPubSubDomain(false);return factory;}@Bean("topicListenerFactory")public JmsListenerContainerFactory <?> topicListenerFactory(ConnectionFactory connectionFactory) {DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();factory.setConnectionFactory(connectionFactory);//设置为发布订阅方式, 默认情况下使用的生产消费者方式factory.setPubSubDomain(true);return factory;}}

推送消息组件中明确factory

  /*** 接收消息*/@JmsListener(destination= "${queueName}",containerFactory = "queueListenerFactory")    //用这个注解去监听 监听的队列public void receive(String messageString){System.out.println("receive:"+messageString);}/*** 接收消息*/@JmsListener(destination= "${topicName}",containerFactory = "topicListenerFactory")public void receiveForTopic1(String messageString){System.out.println("receive1:"+messageString);}

在这里插入图片描述

到此，完成SpringBoot功能整合ActiveMQ的队列模式和广播模式实战。

如果是简单的情况，可以直接在配置中完成，否则可以通过JAVA文件配置
spring.activemq.broker-url=tcp://127.0.0.1:61616
在考虑结束之前等待的时间
spring.activemq.close-timeout=15s
默认代理URL是否应该在内存中。如果指定了显式代理，则忽略此值。
spring.activemq.in-memory=true
是否在回滚回滚消息之前停止消息传递。这意味着当启用此命令时，消息顺序不会被保留。
spring.activemq.non-blocking-redelivery=false
密码
spring.activemq.password=123456
等待消息发送响应的时间。设置为0等待永远。
spring.activemq.send-timeout=0
spring.activemq.user=haha
是否信任所有包
spring.activemq.packages.trust-all=
要信任的特定包的逗号分隔列表（当不信任所有包时）
spring.activemq.packages.trusted=
当连接请求和池满时是否阻塞。设置false会抛“JMSException异常”。
spring.activemq.pool.block-if-full=true
如果池仍然满，则在抛出异常前阻塞时间。
spring.activemq.pool.block-if-full-timeout=-1ms
是否在启动时创建连接。可以在启动时用于加热池。
spring.activemq.pool.create-connection-on-startup=true
是否用Pooledconnectionfactory代替普通的ConnectionFactory。
spring.activemq.pool.enabled=false
连接过期超时。
spring.activemq.pool.expiry-timeout=0ms
连接空闲超时
spring.activemq.pool.idle-timeout=30s
连接池最大连接数
spring.activemq.pool.max-connections=1
每个连接的有效会话的最大数目。
spring.activemq.pool.maximum-active-session-per-connection=500
当有"JMSException"时尝试重新连接
spring.activemq.pool.reconnect-on-exception=true
在空闲连接清除线程之间运行的时间。当为负数时，没有空闲连接驱逐线程运行。
spring.activemq.pool.time-between-expiration-check=-1ms
是否只使用一个MessageProducer
spring.activemq.pool.use-anonymous-producers=true