消息中间件JMS介绍、入门demo与spring整合

本文涉及的产品
注册配置 MSE Nacos/ZooKeeper,182元/月
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
任务调度 XXL-JOB 版免费试用,400 元额度,开发版规格
简介: 消息中间件JMS介绍、入门demo与spring整合

为啥使用消息中间件


看下品优购的结构图

1dc618a0ed9580ce8bfa6facb208c08f.png


我们已经完成了5个web模块和4个服务模块。其中运营商后台的调用关系最多,用到了商家商品服务、广告内容服务、搜索服务和页面生成服务。这种模块之间的依赖也称之为耦合。而耦合越多,之后的维护工作就越困难。那么如何改善系统模块调用关系、减少模块之间的耦合呢?一种解决方案就是----消息中间件。


使用了消息中间件改造后的调用关系:

5d4c6812c8535adbb050f4ddf2e1bce8.png

运营商系统与搜索服务、页面生成服务解除了耦合.


注意:


使用消息中间件的通常前提条件是: 调用的业务执行时间比较长,而且不需要结果返回值的时候。


什么是消息中间件


消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型,它可以在分布式环境下扩展进程间的通信。对于消息中间件,常见的角色大致也就有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 Messaging Service)是Java平台上有关面向消息中间件的技术规范,它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。


JMS本身只定义了一系列的接口规范,是一种与厂商无关的 API,用来访问消息收发系统。它类似于 JDBC(java Database Connectivity):这里,JDBC 是可以用来访问许多不同关系数据库的 API,而 JMS 则提供同样与厂商无关的访问方法,以访问消息收发服务。许多厂商目前都支持 JMS,包括 IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ,这只是几个例子。 JMS 使您能够通过消息收发服务(有时称为消息中介程序或路由器)从一个 JMS 客户机向另一个 JML 客户机发送消息。消息是 JMS 中的一种类型对象,由两部分组成:报头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据或有效负载。


JMS五种消息正文格式


JMS 定义了五种不同的消息正文格式,以及调用的消息类型,允许你发送并接收以一

些不同形式的数据,提供现有消息格式的一些级别的兼容性。

· TextMessage–一个字符串对象

· MapMessage–一套名称-值对

· ObjectMessage–一个序列化的 Java 对象

· BytesMessage–一个字节的数据流

· StreamMessage – Java 原始值的数据流


JMS消息传递类型


对于消息的传递有两种类型:


一种是点对点的,即一个生产者和一个消费者一一对应;

1dc618a0ed9580ce8bfa6facb208c08f.png

另一种是发布/ 订阅模式,即一个生产者产生消息并进行发送后,可以由多个消费者进行接收。

5d4c6812c8535adbb050f4ddf2e1bce8.png


入门小demo(点对点 ,发布/订阅模式)


创建工程


46a9d80a6e05e4e3b19d57a0ee70bcdf.png


点对点


点对点的模式主要建立在一个队列上面,当连接一个列队的时候,发送端不需要知道接收端是否正在接收,可以直接向ActiveMQ发送消息,发送的消息,将会先进入队列中,如果有接收端在监听,则会发向接收端,如果没有接收端接收,则会保存在activemq服务器,直到接收端接收消息,点对点的消息模式可以有多个发送端,多个接收端,但是一条消息,只会被一个接收端给接收到,哪个接收端先连上ActiveMQ,则会先接收到,而后来的接收端则接收不到那条消息。


pom
 <dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-client</artifactId>
  <version>5.13.4</version>
  </dependency>


QueueProducer


//1.创建连接工厂
  ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.157.111:61616");
  //2.获取连接
  Connection connection = connectionFactory.createConnection();
  //3.启动连接
  connection.start();
  //4.获取session  (参数1:是否启动事务,参数2:消息确认模式)
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
  //5.创建队列对象
  Queue queue = session.createQueue("test-queue");
  //6.创建消息生产者
  MessageProducer producer = session.createProducer(queue);
  //7.创建消息
  TextMessage textMessage = session.createTextMessage("欢迎来到神奇的消息队列世界");
  //8.发送消息
  producer.send(textMessage);
  //9.关闭资源
  producer.close();
  session.close();
  connection.close();


上述代码中第4步创建session 的两个参数:

第1个参数 是否使用事务

第2个参数 消息的确认模式

• AUTO_ACKNOWLEDGE = 1 自动确认

• CLIENT_ACKNOWLEDGE = 2 客户端手动确认

• DUPS_OK_ACKNOWLEDGE = 3 自动批量确认

• SESSION_TRANSACTED = 0 事务提交并确认

运行后通过ActiveMQ管理界面查询

5d4c6812c8535adbb050f4ddf2e1bce8.png

消息消费者QueueConsumer


//1.创建连接工厂
  ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.157.111:61616");
  //2.获取连接
  Connection connection = connectionFactory.createConnection();
  //3.启动连接
  connection.start();
  //4.获取session  (参数1:是否启动事务,参数2:消息确认模式)
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  //5.创建队列对象
  Queue queue = session.createQueue("test-queue");
  //6.创建消息消费
  MessageConsumer consumer = session.createConsumer(queue);
  //7.监听消息
  consumer.setMessageListener(new MessageListener() {
  public void onMessage(Message message) {
    TextMessage textMessage=(TextMessage)message;
    try {
    System.out.println("接收到消息:"+textMessage.getText());
    } catch (JMSException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    }
  }
  }); 
  //8.等待键盘输入
  System.in.read(); 
  //9.关闭资源
  consumer.close();
  session.close();
  connection.close();


执行后看到控制台输出


运行测试


同时开启2个以上的消费者,再次运行生产者,观察每个消费者控制台的输出,会发现只有一个消费者会接收到消息。


发布/订阅模式


消息生产者TopicProducer

//1.创建连接工厂
  ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");
  //2.获取连接
  Connection connection = connectionFactory.createConnection();
  //3.启动连接
  connection.start();
  //4.获取session  (参数1:是否启动事务,参数2:消息确认模式)
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  //5.创建主题对象
  Topic topic = session.createTopic("test-topic");
  //6.创建消息生产者
  MessageProducer producer = session.createProducer(topic);
  //7.创建消息
  TextMessage textMessage = session.createTextMessage("欢迎来到申请的消息队列广播世界");
  //8.发送消息
  producer.send(textMessage);
  //9.关闭资源
  producer.close();
  session.close();
  connection.close();


运行效果如下:

46a9d80a6e05e4e3b19d57a0ee70bcdf.png


消息消费者TopicConsumer


//1.创建连接工厂
  ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.25.135:61616");
  //2.获取连接
  Connection connection = connectionFactory.createConnection();
  //3.启动连接
  connection.start();
  //4.获取session  (参数1:是否启动事务,参数2:消息确认模式)
  Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  //5.创建主题对象
  //Queue queue = session.createQueue("test-queue");
  Topic topic = session.createTopic("test-topic");
  //6.创建消息消费
  MessageConsumer consumer = session.createConsumer(topic);
  //7.监听消息
  consumer.setMessageListener(new MessageListener() {
  public void onMessage(Message message) {
    TextMessage textMessage=(TextMessage)message;
    try {
    System.out.println("接收到消息:"+textMessage.getText());
    } catch (JMSException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    }
  }
  });
  //8.等待键盘输入
  System.in.read();
  //9.关闭资源
  consumer.close();
  session.close();
  connection.close();

1dc618a0ed9580ce8bfa6facb208c08f.png


运行测试


同时开启2个以上的消费者,再次运行生产者,观察每个消费者控制台的输出,会发现每个消费者会接收到消息。


代码地址


https://github.com/hufanglei/pinyou/tree/jms-demo


spring整合JMS之activemq


pom


<dependencies>
          <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>
            <version>${spring.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>${spring.version}</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.9</version>
        </dependency>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-client</artifactId>
            <version>5.13.4</version>
        </dependency>


创建2个项目,一个是消息生产者项目 springjms-producer。该项目上有2个类分别发送点对点消息,发布/订阅消息, 在测试类中test中调用改项目。


1dc618a0ed9580ce8bfa6facb208c08f.png


一个是消息消费者项目 springjms-consumer,该项目上有一个监听器,2个配置文件,分别配置点对点,发布/订阅消息。并在测试类test中监控控制台,看消息接受情况。

5d4c6812c8535adbb050f4ddf2e1bce8.png


点对点模式


消息生产者


(1)创建工程springjms_producer,在POM文件中引入SpringJms 、activeMQ以及单元测试相关依赖


(2)在src/main/resources下创建spring配置文件applicationContext-jms-producer.xml


<context:component-scan base-package="cn.itcast.demo"></context:component-scan>        
    <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->  
  <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
     <property name="brokerURL" value="tcp://192.168.157.111:61616"/>  
  </bean>          
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
  <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
  <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
     <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
  </bean>      
    <!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 -->  
  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
     <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
     <property name="connectionFactory" ref="connectionFactory"/>  
  </bean>      
    <!--这个是队列目的地,点对点的  文本信息-->  
  <bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">  
     <constructor-arg value="queue_text"/>  
  </bean>


(3)在cn.itcast.demo包下创建消息生产者类


@Component
public class QueueProducer {
  @Autowired
  private JmsTemplate jmsTemplate;
  @Autowired
  private Destination queueTextDestination;
  /**
  * 发送文本消息
  * @param text
  */
  public void sendTextMessage(final String text){
  jmsTemplate.send(queueTextDestination, new MessageCreator() {   
    public Message createMessage(Session session) throws JMSException {
    return session.createTextMessage(text);
    }
  });  
  }
}


(4)单元测试


在src/test/java创建测试类


@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:applicationContext-jms-producer.xml")
public class TestQueue {
  @Autowired
  private QueueProducer queueProducer;
  @Test
  public void testSend(){
  queueProducer.sendTextMessage("SpringJms-点对点");
  } 
}


消息消费者


(1)创建工程springjms_consumer,在POM文件中引入依赖 (同上一个工程)


(2)创建配置文件 applicationContext-jms-consumer-queue.xml


<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->  
  <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
     <property name="brokerURL" value="tcp://192.168.157.111:61616"/>  
  </bean>    
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
  <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
  <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
     <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
  </bean>   
    <!--这个是队列目的地,点对点的  文本信息-->  
  <bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">  
     <constructor-arg value="queue_text"/>  
  </bean>    
  <!-- 我的监听类 -->
  <bean id="myMessageListener" class="cn.itcast.demo.MyMessageListener"></bean>
  <!-- 消息监听容器 -->
  <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="queueTextDestination" />
  <property name="messageListener" ref="myMessageListener" />
  </bean>


(3)编写监听类


public class MyMessageListener implements MessageListener {
  public void onMessage(Message message) {
  TextMessage textMessage=(TextMessage)message;  
  try {
    System.out.println("接收到消息:"+textMessage.getText());
  } catch (JMSException e) {
    e.printStackTrace();
  }
  }
}


(4)创建测试类


@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:applicationContext-jms-consumer-queue.xml")
public class TestQueue {
  @Test
  public void testQueue(){
  try {
    System.in.read();
  } catch (IOException e) {
    e.printStackTrace();
  }  
  } 
}

46a9d80a6e05e4e3b19d57a0ee70bcdf.png


发布/订阅模式


1)在工程springjms_producer的applicationContext-jms-producer.xml增加配置


<!--这个是订阅模式  文本信息-->  
  <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">  
     <constructor-arg value="topic_text"/>  
  </bean>


(2)创建生产者类


@Component
public class TopicProducer {
  @Autowired
  private JmsTemplate jmsTemplate;
  @Autowired
  private Destination topicTextDestination;
  /**
  * 发送文本消息
  * @param text
  */
  public void sendTextMessage(final String text){
  jmsTemplate.send(topicTextDestination, new MessageCreator() {   
    public Message createMessage(Session session) throws JMSException {
    return session.createTextMessage(text);
    }
  });  
  }
}


(3)编写测试类


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 cn.itcast.demo.TopicProducer;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:applicationContext-activemq-producer.xml")
public class TestTopic {
  @Autowired
  private TopicProducer topicProducer;
  @Test
  public void sendTextQueue(){  
  topicProducer.sendTextMessage();
  } 
}


消息消费者


(1)在activemq-spring-consumer工程中创建配置文件applicationContext-jms-consumer-topic.xml


<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->  
  <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  
     <property name="brokerURL" value="tcp://192.168.157.111:61616"/>  
  </bean>
    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
  <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  
  <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
     <property name="targetConnectionFactory" ref="targetConnectionFactory"/>  
  </bean>  
    <!--这个是队列目的地,点对点的  文本信息-->  
  <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">  
     <constructor-arg value="topic_text"/>  
  </bean>    
  <!-- 我的监听类 -->
  <bean id="myMessageListener" class="cn.itcast.demo.MyMessageListener"></bean>
  <!-- 消息监听容器 -->
  <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="topicTextDestination" />
  <property name="messageListener" ref="myMessageListener" />
  </bean>


(2)编写测试类


@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations="classpath:applicationContext-jms-consumer-topic.xml")
public class TestTopic {
  @Test
  public void testTopic(){
  try {
    System.in.read();
  } catch (IOException e) {
    e.printStackTrace();
  }  
  }  
}


测试:同时运行三个消费者工程,在运行生产者工程,查看三个消费者工程的控制台输出

66ba272a0bfc97be54a5fa679e3d5482.png


spring整合jms源码地址


https://github.com/hufanglei/pinyou/tree/springjms-producer


https://github.com/hufanglei/pinyou/tree/springjms-consumer



相关文章
|
4月前
|
人工智能 Java API
Spring AI 实战|Spring AI入门之DeepSeek调用
本文介绍了Spring AI框架如何帮助Java开发者轻松集成和使用大模型API。文章从Spring AI的初探开始,探讨了其核心能力及应用场景,包括手动与自动发起请求、流式响应实现打字机效果,以及兼容不同AI服务(如DeepSeek、通义千问)的方法。同时,还详细讲解了如何在生产环境中添加监控以优化性能和成本管理。通过Spring AI,开发者可以简化大模型调用流程,降低复杂度,为企业智能应用开发提供强大支持。最后,文章展望了Spring AI在未来AI时代的重要作用,鼓励开发者积极拥抱这一技术变革。
1339 71
Spring AI 实战|Spring AI入门之DeepSeek调用
|
5月前
|
安全 Java 数据库
Spring Security 实战指南:从入门到精通
本文详细介绍了Spring Security在Java Web项目中的应用,涵盖登录、权限控制与安全防护等功能。通过Filter Chain过滤器链实现请求拦截与认证授权,核心组件包括AuthenticationProvider和UserDetailsService,负责用户信息加载与密码验证。文章还解析了项目结构,如SecurityConfig配置类、User实体类及自定义登录逻辑,并探讨了Method-Level Security、CSRF防护、Remember-Me等进阶功能。最后总结了Spring Security的核心机制与常见配置,帮助开发者构建健壮的安全系统。
292 0
|
3月前
|
Java 关系型数据库 MySQL
【Spring】【事务】初学者直呼学会了的Spring事务入门
本文深入解析了Spring事务的核心概念与使用方法。Spring事务是一种数据库事务管理机制,通过确保操作的原子性、一致性、隔离性和持久性(ACID),维护数据完整性。文章详细讲解了声明式事务(@Transactional注解)和编程式事务(TransactionTemplate、PlatformTransactionManager)的区别与用法,并探讨了事务传播行为(如REQUIRED、REQUIRES_NEW等)及隔离级别(如READ_COMMITTED、REPEATABLE_READ)。
219 1
|
3月前
|
Java API 微服务
Java 21 与 Spring Boot 3.2 微服务开发从入门到精通实操指南
《Java 21与Spring Boot 3.2微服务开发实践》摘要: 本文基于Java 21和Spring Boot 3.2最新特性,通过完整代码示例展示了微服务开发全流程。主要内容包括:1) 使用Spring Initializr初始化项目,集成Web、JPA、H2等组件;2) 配置虚拟线程支持高并发;3) 采用记录类优化DTO设计;4) 实现JPA Repository与Stream API数据访问;5) 服务层整合虚拟线程异步处理和结构化并发;6) 构建RESTful API并使用Springdoc生成文档。文中特别演示了虚拟线程配置(@Async)和StructuredTaskSco
309 0
|
5月前
|
存储 安全 Java
Spring Security 入门与详解
Spring Security 是 Spring 框架中的核心安全模块,提供认证、授权及防护功能。本文详解其核心概念,包括认证(Authentication)、授权(Authorization)和过滤器链(Security Filter Chain)。同时,通过代码示例介绍基本配置,如 PasswordEncoder、UserDetailsService 和自定义登录页面等。最后总结常见问题与解决方法,助你快速掌握 Spring Security 的使用与优化。
1152 0
|
6月前
|
消息中间件 Java API
微服务——SpringBoot使用归纳——Spring Boot中集成ActiveMQ—— JMS 和 ActiveMQ 介绍
本文介绍如何在Spring Boot中集成ActiveMQ,首先阐述了JMS(Java消息服务)的概念及其作为与具体平台无关的API在异步通信中的作用。接着说明了JMS的主要对象模型,如连接工厂、会话、生产者和消费者等,并指出JMS支持点对点和发布/订阅两种消息类型。随后重点讲解了ActiveMQ,作为Apache开源的消息总线,它完全支持JMS规范,适用于异步消息处理。最后,文章探讨了在Spring Boot中使用队列(Queue)和主题(Topic)这两种消息通信形式的方法。
145 0
|
9月前
|
存储 安全 Java
Spring Security 入门
Spring Security 是 Spring 框架中的安全模块,提供强大的认证和授权功能,支持防止常见攻击(如 CSRF 和会话固定攻击)。它通过过滤器链拦截请求,核心概念包括认证、授权和自定义过滤器。配置方面,涉及密码加密、用户信息服务、认证提供者及过滤器链设置。示例代码展示了如何配置登录、注销、CSRF防护等。常见问题包括循环重定向、静态资源被拦截和登录失败未返回错误信息,解决方法需确保路径正确和添加错误提示逻辑。
512 2
Spring Security 入门
|
8月前
|
人工智能 自然语言处理 Java
Spring Cloud Alibaba AI 入门与实践
本文将介绍 Spring Cloud Alibaba AI 的基本概念、主要特性和功能,并演示如何完成一个在线聊天和在线画图的 AI 应用。
1897 7
|
9月前
|
Java 开发者 微服务
Spring Boot 入门:简化 Java Web 开发的强大工具
Spring Boot 是一个开源的 Java 基础框架,用于创建独立、生产级别的基于Spring框架的应用程序。它旨在简化Spring应用的初始搭建以及开发过程。
282 7
Spring Boot 入门:简化 Java Web 开发的强大工具