wordpress 仿站 主题,网站建设拍金手指排名贰拾,网站是广西住房和城乡建设厅,百度公司注册地址目录 前言步骤引入相关maven依赖添加相关配置 使用方法配置消息序列化创建第一个消息队列和交换机使用方法 总结 前言
在当今的微服务架构盛行的时代#xff0c;消息队列作为一种重要的通信机制#xff0c;在分布式系统中扮演着不可或缺的角色。RabbitMQ#xff0c;作为一款… 目录 前言步骤引入相关maven依赖添加相关配置 使用方法配置消息序列化创建第一个消息队列和交换机使用方法 总结 前言
在当今的微服务架构盛行的时代消息队列作为一种重要的通信机制在分布式系统中扮演着不可或缺的角色。RabbitMQ作为一款开源的消息代理和队列服务器以其高可用性、易扩展性、灵活的路由机制以及多协议支持等特点深受开发者们的青睐。而Spring Cloud作为Spring生态中针对微服务架构的一套集成解决方案也提供了与RabbitMQ的集成支持使得在Spring Cloud环境下使用RabbitMQ变得更加简单高效。 Spring Cloud集成RabbitMQ不仅继承了RabbitMQ本身的诸多优点还充分利用了Spring Cloud的自动配置和声明式编程特性极大地简化了消息队列的配置和使用过程。开发者可以通过简单的配置和注解轻松实现消息的发布、订阅、路由和持久化等功能从而构建出稳定可靠、高性能的分布式系统。 此外Spring Cloud集成RabbitMQ还提供了丰富的消息处理机制如消息确认、死信队列、延迟队列等这些机制可以帮助开发者更好地处理消息丢失、重复消费、消息堆积等问题提升系统的健壮性和可靠性。
步骤
引入相关maven依赖
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency添加相关配置
spring: rabbitmq: # RabbitMQ服务器的地址 host: 127.0.0.1 # RabbitMQ服务器的端口号 port: 5672 # RabbitMQ服务器的用户名 username: guest # RabbitMQ服务器的密码 password: guest # 消息监听器的配置 listener: simple: # 确认模式这里设置为手动意味着需要手动确认消息处理成功 acknowledge-mode: manual # 消息重试的配置 retry: # 是否启用重试机制 enabled: true # 最大重试次数 max-attempts: 5 # 最大重试间隔单位是毫秒 max-interval: 20000ms # 初始重试间隔单位是毫秒 initial-interval: 2000ms # 重试间隔的倍增系数 multiplier: 2使用方法
配置消息序列化
// 定义一个Bean返回一个MessageConverter实例用于消息的序列化和反序列化
Bean
public MessageConverter messageConverter() { // 使用Jackson2JsonMessageConverter作为消息转换器它基于Jackson库进行JSON格式转换 return new Jackson2JsonMessageConverter();
} // 定义一个Bean返回一个RabbitTemplate实例用于发送和接收消息
Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) { // 创建一个RabbitTemplate实例传入连接工厂用于建立与RabbitMQ的连接 RabbitTemplate rabbitTemplate new RabbitTemplate(connectionFactory); // 设置RabbitTemplate的消息转换器用于将Java对象转换为消息以及将消息转换为Java对象 rabbitTemplate.setMessageConverter(messageConverter); // 返回配置好的RabbitTemplate实例供其他组件使用 return rabbitTemplate;
}创建第一个消息队列和交换机
// 定义actuator队列的名称常量
public static final String ACTUATOR_QUEUE_NAME actuator_queue;
// 定义actuator交换机的名称常量
public static final String ACTUATOR_EXCHANGE_NAME actuator_exchange;
// 定义actuator路由键的常量
public static final String ACTUATOR_ROUTING_KEY actuator_routing_key; // 定义一个Bean用于创建actuator队列
Bean
public Queue actuatorQueue() { // 创建一个持久化队列队列名称为ACTUATOR_QUEUE_NAME常量定义的值 return new Queue(ACTUATOR_QUEUE_NAME, true);
} // 定义一个Bean用于创建actuator交换机
Bean
public DirectExchange actuatorExchange() { // 创建一个Direct类型的交换机交换机名称为ACTUATOR_EXCHANGE_NAME常量定义的值 return new DirectExchange(ACTUATOR_EXCHANGE_NAME);
} // 定义一个Bean用于绑定actuator队列和交换机
Bean
public Binding actuatorBinding(Qualifier(actuatorQueue) Queue queue, Qualifier(actuatorExchange) DirectExchange exchange) { // 使用BindingBuilder来绑定actuator队列和actuator交换机 // 并指定ACTUATOR_ROUTING_KEY作为路由键 return BindingBuilder.bind(queue).to(exchange).with(ACTUATOR_ROUTING_KEY);
}使用方法
// 使用RabbitListener注解指定监听ACTUATOR_QUEUE_NAME队列中的消息
RabbitListener(queues ActuatorBind.ACTUATOR_QUEUE_NAME)
public void onMessage(TaskInfoBo taskInfoBo, Message message, Channel channel) throws IOException { // 获取消息的属性并从中提取消息的deliveryTag long deliveryTag message.getMessageProperties().getDeliveryTag(); try { // 执行一些业务逻辑代码...... // ...业务代码省略 // 如果业务代码执行成功则手动确认消息处理成功 // deliveryTag用于标识需要确认的消息false表示是否进行批量确认这里不进行批量确认 channel.basicAck(deliveryTag, false); } catch (Exception e) { // 如果在业务代码执行过程中出现异常则手动拒绝该消息 // deliveryTag用于标识需要拒绝的消息false表示是否将消息重新放回队列这里不重新放回队列 channel.basicReject(deliveryTag, false); }
}在这个方法中我们使用了RabbitListener注解来声明这个方法是一个消息监听器它监听ActuatorBind.ACTUATOR_QUEUE_NAME队列中的消息。当消息到达这个队列时Spring会调用这个方法来处理消息。 方法接受三个参数TaskInfoBo taskInfoBo用于接收消息体中的信息并自动转换为TaskInfoBo对象、Message message代表原始的RabbitMQ消息和Channel channel用于与RabbitMQ进行交互的通道。 在方法体中我们首先通过message对象获取消息的deliveryTag这个deliveryTag用于后续的消息确认或拒绝操作。 然后我们尝试执行一些业务逻辑代码。如果业务代码执行成功我们使用channel.basicAck方法手动确认消息已被成功处理从而从队列中移除该消息。如果业务代码执行过程中出现异常我们使用channel.basicReject方法手动拒绝该消息这里拒绝时不将消息重新放回队列第二个参数为false。 这种手动确认消息的方式确保了消息处理的可靠性和健壮性只有当消息被成功处理后才从队列中移除否则会被拒绝或重新尝试处理。
总结
完成上诉步骤我们完成了Spring Cloud 集成RabbitMQ。还有很多其他的模式等到业务用到后再进行补充。
