上海做网站要多少钱,网站怎么做移动端适配,郑州微网站建设,温州哪里有网站建设订阅模式Publish/Subscribe 基于API的方式1.使用AmqpAdmin定制消息发送组件2.消息发送者发送消息3.消息消费者接收消息 基于配置类的方式基于注解的方式总结 SpringBoot整合RabbitMQ中间件实现消息服务#xff0c;主要围绕3个部分的工作进行展开#xff1a;定制中间件、消息发… 订阅模式Publish/Subscribe 基于API的方式1.使用AmqpAdmin定制消息发送组件2.消息发送者发送消息3.消息消费者接收消息 基于配置类的方式基于注解的方式总结 SpringBoot整合RabbitMQ中间件实现消息服务主要围绕3个部分的工作进行展开定制中间件、消息发送者发送消息、消息消费者接收消息。其中定制中间件是比较麻烦的工作且必须预先定制。
下面以用户注册成功后同时发送邮件通知和短信通知这一场景为例 分别使用基于API、基于配置类和基于注解这3种方式来实现Publish/Subscribe工作模式的整合。
基于API的方式
基于API的方式是指使用Spring框架提供的API管理类AmqpAdmin定制消息发送组件并进行消息的发送。这种定制消息发送组件的方式与在RabbitMQ可视化界面上通过对应面板进行组件操作的实现基本一样都是通过管理员的身份预先手动声明交换器、队列、路由键等然后组装消息队列供应用程序调用从而实现消息服务。下面我们就对这种基于API的方式进行讲解和演示。
1.使用AmqpAdmin定制消息发送组件
我们先打开chapter08项目的测试类Chapter08ApplicationTests在该测试类中先引入AmqpAdmin管理类定制Publish/Subscribe工作模式所需的消息组件。
package com.ytx;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;SpringBootTest
class Chapter08ApplicationTests {Autowiredprivate AmqpAdmin amqpAdmin;Testvoid contextLoads() {}/** 使用AmqpAdmin管理员API定制消息组件 */Testpublic void amqpAdmin() {// 1.定义fanout类型的交换器amqpAdmin.declareExchange(new FanoutExchange(fanout_exchange));// 2.定义两个默认持久化队列分别处理email和smsamqpAdmin.declareQueue(new Queue(fanout_queue_email));amqpAdmin.declareQueue(new Queue(fanout_queue_sms));// 3.将队列分别与交换器进行绑定amqpAdmin.declareBinding(new Binding(fanout_queue_email, Binding.DestinationType.QUEUE, fanout_exchange, , null));amqpAdmin.declareBinding(new Binding(fanout_queue_sms, Binding.DestinationType.QUEUE, fanout_exchange, , null));}
}执行上述单元测试方法amqpAdmin()验证RabbitMQ消息组件的定制效果。单元测试方法执行成功后通过RabbitMQ可视化管理页面的Exchanges面板查看效果。 从上图可以看出在RabbitMQ可视化管理页面的Exchanges面板中新出现了一个名称为fanout_exchange的交换器(其他7个交换器是RabbitMQ自带的)且其类型是我们设置的fanout类型。我们可以单击fanout_exchange交换器进入查看。 从上图可以看出在fanout_exchange交换器详情页面中展示有该交换器的具体信息还有与之绑定的两个消息队列fanout_queue_email和fanout_queue_sms并且与程序中设置的绑定规则一致。切换到Queues面板页面查看定制生成的消息队列信息。
从上图可以看出在Queues队列面板页面中展示有定制的消息队列信息这与程序中定制的消息队列一致我们可以单击消息队列名称查看每个队列的详情。
通过上述操作可以发现在管理页面中提供了消息组件交换器、队列的定制功能。在程序中使用Spring框架提供的管理员API组件AmqpAdmin定制消息组件和在管理页面上手动定制消息组件的本质是一样的。
2.消息发送者发送消息
完成消息组件的定制工作后创建消息发送者发送消息到消息队列中。发送消息时我们可以借助一个实体类传递消息需要预先创建一个实体类对象。
首先在chapter08项目中创建名为com.cy.domain的包并在该包下创建一个实体类User。
package com.ytx.domain;/** 发布消息的实体类可以通过实现Serializable序列化接口进行发布 */
public class User {private Integer id;private String username;public Integer getId() {return id;}public void setId(Integer id) {this.id id;}public String getUsername() {return username;}public void setUsername(String username) {this.username username;}Overridepublic String toString() {return User{ id id , username username \ };}
}其次我们在项目测试类Chapter08ApplicationTests中使用Spring框架提供的RabbitTemplate模板类实现消息发送。
Autowired
private RabbitTemplate rabbitTemplate;/** 1.Publish/Subscribe工作模式消息发送端 */
Test
public void subPublisher() {User user new User();user.setId(1);user.setUsername(小明);rabbitTemplate.convertAndSend(fanout_exchange, , user);
}上述代码中我们先使用Autowired注解引入消息中间件管理的RabbitTemplate组件对象然后使用该模板工具类的convertAndSend(String exchange, String routingKey, Object object)方法进行消息发布。此方法中的第1个参数表示发送消息的交换器这个参数值要与之前定制的交换器名称一致第2个参数表示路由键因为实现的是Publish/Subscribe工作模式所以不需要指定第3个参数是发送的消息内容接收Object类型。
然后执行上述消息发送的测试方法subPublisher()控制台执行效果见下图所示。
从上图可以看出发送实体类对象消息时程序发生异常从异常信息“SimpleMessageConverter only supports String, byte[] and Serializable payloads”可以看出消息发送过程中默认使用了SimpleMessageConverter转换器进行消息转换存储该转换器只支持字符串或实体类对象序列化后的消息。而测试类中发送的是User实体类对象消息所以发生异常。
如果要解决上述消息中间件发送实体类消息出现的异常我们通常可以采用两种解决方案第一种是执行JDK自带的Serializable序列化接口第二种是定制其他类型的消息转化器。两种实现方式都可行相对于第二种实现方式而言第一种方式实现后的可视化效果较差转换后的消息无法辨识所以一般使用第二种方式。
接着我们在chapter08项目中创建名为com.ytx.config的包并在该包下创建一个RabbitMQ消息配置类RabbitMQConfig。
package com.ytx.config;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/** RabbitMQ消息配置类 */
Configuration
public class RabbitMQConfig {/** 定制JSON格式的消息转换器 */Beanpublic MessageConverter messageConverter() {return new Jackson2JsonMessageConverter();}
}代码中创建一个RabbitMQ消息配置类RabbitMQConfig并在该配置类中通过Bean注解自定义一个Jackson2JsonMessageConverter类型的消息转换器组件该组件的返回值必须为MessageConverter类型。
再次执行subPublisher()方法该方法执行成功后查看RabbitMQ可视化管理页面Queues面板信息。 从上图可以看出消息发送完成后Publish/Subscribe工作模式下绑定的两个消息队列中各自拥有一条待接收的消息 由于目前尚未提供消息。
3.消息消费者接收消息
在chapter08项目中创建名为com.ytx.service的包并在该包下创建一个针对RabbitMQ消息中间件进行消息接收和处理的业务类RabbitMQService。
package com.ytx.chapter08.service;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;/** RabbitMQ消息接收处理的业务类 */
Service
public class RabbitMQService {/** Publish/Subscribe工作模式接收处理邮件业务 */RabbitListener(queues fanout_queue_email)public void subConsumerEmail(Message message) {byte[] body message.getBody();String msg new String(body);System.out.println(邮件业务接收到消息 msg);}/** Publish/Subscribe工作模式接收处理短信业务 */RabbitListener(queues fanout_queue_sms)public void subConsumerSms(Message message) {byte[] body message.getBody();String msg new String(body);System.out.println(短信业务接收到消息 msg);}
}上述代码中创建了一个接收处理RabbitMQ消息的业务处理类RabbitMQService在该类中使用Spring框架提供的RabbitListener注解我们可以监听队列名称为fanout_queue_email和fanout_queue_sms的消息监听的这两个队列是前面指定发送并存储消息的消息队列。
需要说明的是使用RabbitListener注解监听队列消息后一旦服务启动且监听到指定的队列中有消息存在(目前两个队列中各有一条相同的消息)对应注解的方法就会立即接收并消费队列中的消息。另外在接收消息的方法中参数类型可以与发送的消息类型保持一致或者使用Object类型和Message类型。如果使用与消息类型对应的参数接收消息的话只能够得到具体的消息体信息如果使用Object或者Message类型参数接收消息的话还可以获得除了消息体外的消息参数信息MessageProperties。
启动chapter08项目控制台显示的消息消费效果如下图所示。 从上图可以看出项目启动成功后消息消费者监听到消息队列中存在的两条消息并进行了各自的消费。与此同时通过RabbitMQ可视化管理页面的Queues面板查看队列消息情况会发现两个队列中存储的消息已经被消费。至此一条完整的消息发送、 消息中间件存储、消息消费的Publish/Subscribe发布订阅模式工作模式的业务案例已经实现。
注意如果没有引入Spring Web模块的依赖启动chapter08项目时消息消费者接收消息会报以下错误。
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.amqp.support.converter.MessageConverter]: Factory method messageConverter threw exception; nested exception is java.lang.NoClassDefFoundError: com/fasterxml/jackson/databind/ObjectMapperat org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:185) ~[spring-beans-5.3.25.jar:5.3.25]at org.springframework.beans.factory.support.ConstructorResolver.instantiate(ConstructorResolver.java:653) ~[spring-beans-5.3.25.jar:5.3.25]... 18 common frames omitted小提示 上述代码中使用的是开发中常用的RabbitListener注解来监听指定名称队列的消息情况这种方式会在监听到指定队列存在消息后立即进行消费处理。除此之外我们还可以使用RabbitTemplate模板类的receiveAndConvert(String queueName)方法手动消费指定队列中的消息。 基于配置类的方式
基于配置类的方式主要讲的是使用SpringBoot框架提供的Configuration注解配置定制消息发送组件并进行消息发送。下面我们来对这种基于配置类的方式进行讲解和演示。
打开RabbitMQ消息配置类RabbitMQConfig在该配置类中使用基于配置类的方式定制消息发送相关组件。
package com.ytx.config;
import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/** RabbitMQ消息配置类 */
Configuration
public class RabbitMQConfig {/** 定制JSON格式的消息转换器 */Beanpublic MessageConverter messageConverter() {return new Jackson2JsonMessageConverter();}/** 使用基于配置类的方式定制消息中间件 */// 1.定义fanout类型的交换器Beanpublic Exchange fanoutExchange() {return ExchangeBuilder.fanoutExchange(fanout_exchange).build();}// 2.定义两个不同名称的消息队列Beanpublic Queue fanoutQueueEmail() {return new Queue(fanout_queue_email);}Beanpublic Queue fanoutQueueSms() {return new Queue(fanout_queue_sms);}// 3.将两个不同名称的消息队列与交换器进行绑定Beanpublic Binding bindingEmail() {return BindingBuilder.bind(fanoutQueueEmail()).to(fanoutExchange()).with().noargs();}Beanpublic Binding bindingSms() {return BindingBuilder.bind(fanoutQueueSms()).to(fanoutExchange()).with().noargs();}
}上述代码中使用Bean注解定制了3种类型的Bean组件这3种组件分别表示交换器、消息队列和消息队列与交换器的绑定。这种基于配置类方式定制的消息组件其实现和基于API方式定制的消息组件完全一样只不过是实现方式不同而已。
按照消息服务整合实现步骤完成消息组件的定制后还需要编写消息发送者和消息消费者而在基于API的方式中已经实现了消息发送者和消息消费者并且基于配置类方式定制的消息组件名称和之前测试用的消息发送和消息消费组件名称都是一致的所以这里我们可以直接重复使用。
重新运行消息发送者测试方法subPublisher()消息消费者可以自动监听并消费消息队列中存在的消息效果与基于API的方式测试效果一样。
基于注解的方式
基于注解的方式指的是使用Spring框架的RabbitListener注解定制消息发送组件并发送消息。
在消息接收和处理的业务类RabbitMQService中将针对邮件业务和短信业务处理的消息消费者方法进行注释使用RabbitListener注解及其相关属性定制消息发送组件。
package com.ytx.chapter08.service;
import com.ytx.chapter08.domain.User;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;/** RabbitMQ消息接收处理的业务类 */
Service
public class RabbitMQService {/** Publish/Subscribe工作模式接收处理邮件业务 *//*RabbitListener(queues fanout_queue_email)public void subConsumerEmail(Message message) {byte[] body message.getBody();String msg new String(body);System.out.println(邮件业务接收到消息 msg);}*//** Publish/Subscribe工作模式接收处理短信业务 *//*RabbitListener(queues fanout_queue_sms)public void subConsumerSms(Message message) {byte[] body message.getBody();String msg new String(body);System.out.println(短信业务接收到消息 msg);}*//** 使用基于注解的方式实现消息服务 */// 1.1 Publish/Subscribe工作模式接收处理邮件业务RabbitListener(bindings QueueBinding(value Queue(fanout_queue_email),exchange Exchange(value fanout_exchange, type fanout)))public void subConsumerEmailAno(User user) {System.out.println(邮件业务接收到消息 user);}// 1.2 Publish/Subscribe工作模式接收处理短信业务RabbitListener(bindings QueueBinding(value Queue(fanout_queue_sms),exchange Exchange(value fanout_exchange, type fanout)))public void subConsumerSmsAno(User user) {System.out.println(短信业务接收到消息 user);}
}上述代码中使用RabbitListener注解及其相关属性定制了两个消息组件的消费者这两个消费者都接收实体类User并消费。在RabbitListener注解中bindings属性用于创建并绑定交换器和消息队列组件需要注意的是为了能使两个消息组件的消费者接收到实体类User需要我们在定制交换器时将交换器类型type设置为fanout。另外bindings属性的QueueBinding注解除了有value、exchange属性外还有key属性用于定制路由键routingKey当前发布订阅模式不需要。
重启测试方法subPublisher()消息消费者可以自动监听并消费消息队列中存在的消息效果与基于API的方式测试效果一样。
至此我们就在SpringBoot中完成了基于API、基于配置类和基于注解这3种方式来实现Publish/Subscribe工作模式的整合讲解。在这3种实现消息服务的方式中基于API的方式相对简单、直观但容易与业务代码产生耦合基于配置类的方式相对隔离、容易统一管理、符合Spring Boot框架思想基于注解的方式清晰明了、方便各自管理但是也容易与业务代码产生耦合。
在实际开发中使用基于配置类的方式和基于注解的方式较为常见基于API的方式则偶尔使用当然大家要根据实际情况进行具体选择。
总结
今天介绍了基于API方式、配置类方式和注解的3种消息队列并展示了实现发布订阅Publish/Subscribe模式的整合及代码实现基于注解方式的实现需要重点掌握。有关RabbitMQ的其他内容袁老后续更新
