深圳建伟业公司商城_建大网络_上海网优化seo公司_最常见企业网站公司有哪些

一、前言

本篇主要是围绕着 Spring Boot 3.x 与 RabbitMQ 的动态配置集成，对比上一篇文章，进一步集成RabbitMQ动态操作，比如动态新增 RabbitMQ 实例，以及动态实例中的交换机、队列等操作。

二、动态RabbitMQ实例，创建、删除

1、RabbitMQ动态实例配置

DynamicRabbitMQConfig.java

mport com.chain.air.rpp.exchange.properties.RabbitInstance;
import com.chain.air.rpp.exchange.properties.RabbitProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.HeadersExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;/*** @className: DynamicRabbitMQConfig* @program: chain* @description: 动态RabbitMQ配置* @author: kenny* @create: 2024-10-09 22:40* @version: 1.0.0*/
@Slf4j
@Configuration
public class DynamicRabbitMQConfig {/*** RabbitMQ实例配置信息集合*/private final RabbitProperties rabbitProperties;/*** RabbitMQ动态实例列表*/private Map<String, RabbitInstance> rabbitInstanceMap = new HashMap<>();/*** RabbitTemplate集合*/private Map<String, RabbitTemplate> rabbitTemplateMap = new HashMap<>();/*** RabbitAdmin集合*/private Map<String, RabbitAdmin> rabbitAdminMap = new HashMap<>();/*** RabbitAdmin集合*/private Map<String, ConnectionFactory> connectionFactoryMap = new HashMap<>();@Autowiredpublic DynamicRabbitMQConfig(RabbitProperties rabbitProperties) {this.rabbitProperties = rabbitProperties;}/*** 初始化 RabbitMQ 实例** @PostConstruct 注解，在 Spring 容器初始化后执行*/@PostConstructpublic void init() {rabbitProperties.getInstances().forEach(this::createRabbitInstance);}/*** 创建一个 RabbitMQ 实例** @param instance 实例配置信息*/public void createRabbitInstance(RabbitInstance instance) {// 创建连接工厂CachingConnectionFactory connectionFactory = new CachingConnectionFactory();connectionFactory.setHost(instance.getHost());connectionFactory.setPort(instance.getPort());connectionFactory.setUsername(instance.getUsername());connectionFactory.setPassword(instance.getPassword());connectionFactory.setVirtualHost(instance.getVirtualHost());connectionFactoryMap.put(instance.getName(), connectionFactory);// 创建 RabbitTemplateRabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);// 设置消息转换器rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter(new ObjectMapper()));rabbitTemplateMap.put(instance.getName(), rabbitTemplate);// 创建 RabbitAdminRabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);rabbitAdminMap.put(instance.getName(), rabbitAdmin);// 存储实例信息rabbitInstanceMap.put(instance.getName(), instance);}/*** 根据名称获取 RabbitTemplate** @param name 名称* @return RabbitTemplate*/public RabbitTemplate getRabbitTemplate(String name) {return rabbitTemplateMap.get(name);}/*** 根据名称获取 RabbitAdmin** @param name 名称* @return RabbitAdmin*/public RabbitAdmin getRabbitAdmin(String name) {return rabbitAdminMap.get(name);}/*** 根据名称获取 ConnectionFactory** @param name 名称* @return ConnectionFactory*/public ConnectionFactory getConnectionFactory(String name) {return connectionFactoryMap.get(name);}/*** 根据名称获取 RabbitTemplate 和 RabbitAdmin** @param name 名称* @return Map*/public Map<String, Object> getRabbitInstance(String name) {Map<String, Object> result = new HashMap<>();result.put("instance", rabbitInstanceMap.get(name));result.put("rabbitAdmin", rabbitAdminMap.get(name));result.put("rabbitTemplate", rabbitTemplateMap.get(name));result.put("connectionFactory", connectionFactoryMap.get(name));return result;}/*** 检查 RabbitMQ 实例是否存在** @param name 名称* @return Boolean*/public Boolean checkInstanceExist(String name) {return rabbitInstanceMap.containsKey(name);}/*** 获取RabbitMQ动态实例名称集合** @return RabbitMQ动态实例名称集合*/public List<String> getDynamicInstanceNames() {return new ArrayList<>(rabbitInstanceMap.keySet());}/*** 根据名称删除RabbitMQ实例** @param name 名称*/public void removeRabbitInstance(String name) {rabbitInstanceMap.remove(name);rabbitAdminMap.remove(name);rabbitTemplateMap.remove(name);connectionFactoryMap.remove(name);}
}

2、RabbitMQ动态实例操作Service组件

RabbitDynamicInstanceService.java

import com.chain.air.rpp.exchange.config.rabbit.DynamicRabbitMQConfig;
import com.chain.air.rpp.exchange.properties.RabbitInstance;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import java.util.List;/*** @className: RabbitDynamicInstanceService* @program: chain* @description: RabbitMQ 动态实例 Service组件* @author: kenny* @create: 2024-10-10 21:56* @version: 1.0.0*/
@Slf4j
@Service
public class RabbitDynamicInstanceService {/*** RabbitMQ动态实例配置*/private final DynamicRabbitMQConfig dynamicRabbitMQConfig;@Autowiredpublic RabbitDynamicInstanceService(DynamicRabbitMQConfig dynamicRabbitMQConfig) {this.dynamicRabbitMQConfig = dynamicRabbitMQConfig;}/*** 创建RabbitMQ动态实例** @param rabbitInstance 实例配置信息* @return 处理结果*/public Boolean createRabbitInstance(RabbitInstance rabbitInstance) {try {boolean instanceExist = dynamicRabbitMQConfig.checkInstanceExist(rabbitInstance.getName());if (instanceExist) {log.warn("实例【{}】已存在，无需重复创建", rabbitInstance.getName());return true;}dynamicRabbitMQConfig.createRabbitInstance(rabbitInstance);return true;} catch (Exception e) {log.error("创建RabbitMQ实例失败，失败原因：{}", e.getMessage());return false;}}/*** 删除RabbitMQ动态实例** @param name 实例名称* @return 处理结果*/public Boolean removeRabbitInstance(String name) {try {boolean instanceExist = dynamicRabbitMQConfig.checkInstanceExist(name);if (!instanceExist) {log.warn("实例【{}】不存在，无需删除", name);return true;}dynamicRabbitMQConfig.removeRabbitInstance(name);return true;} catch (Exception e) {log.error("删除RabbitMQ实例失败，失败原因：{}", e.getMessage());return false;}}/*** 获取RabbitMQ动态实例名称列表** @return 实例名称列表*/public List<String> getDynamicInstanceNames() {return dynamicRabbitMQConfig.getDynamicInstanceNames();}/*** 指定动态实例，通过exchange发送消息** @param name       实例名称* @param exchange   交换机* @param routingKey 路由键* @param message    消息内容* @return 处理结果*/public Boolean appointRabbitWithExchangeSendMessage(String name, String exchange, String routingKey, String message) {try {Boolean instanceExist = dynamicRabbitMQConfig.checkInstanceExist(name);if (!instanceExist) {log.warn("实例【{}】不存在，无法发送消息", name);return false;}dynamicRabbitMQConfig.sendMessage(name, exchange, routingKey, message);return true;} catch (Exception e) {log.error("指定RabbitMQ实例发送消息失败，失败原因：{}", e.getMessage());return false;}}/*** 指定动态实例，通过queue发送消息** @param name    实例名称* @param queue   消息队列* @param message 消息内容* @return 处理结果*/public Boolean appointRabbitWithQueueSendMessage(String name, String queue, String message) {try {Boolean instanceExist = dynamicRabbitMQConfig.checkInstanceExist(name);if (!instanceExist) {log.warn("实例【{}】不存在，无法发送消息", name);return false;}dynamicRabbitMQConfig.sendMessage(name, queue, message);return true;} catch (Exception e) {log.error("指定RabbitMQ实例发送消息失败，失败原因：{}", e.getMessage());return false;}}}

二、动态RabbitMQ实例中的exchange、queue动态新增及监听

1、DynamicRabbitMQConfig.java 新增操作代码

import com.chain.air.rpp.exchange.properties.RabbitInstance;
import com.chain.air.rpp.exchange.properties.RabbitProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.HeadersExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;/*** @className: DynamicRabbitMQConfig* @program: chain* @description: 动态RabbitMQ配置* @author: kenny* @create: 2024-10-09 22:40* @version: 1.0.0*/
@Slf4j
@Configuration
public class DynamicRabbitMQConfig {/*** RabbitMQ实例配置信息集合*/private final RabbitProperties rabbitProperties;/*** RabbitMQ动态实例列表*/private Map<String, RabbitInstance> rabbitInstanceMap = new HashMap<>();/*** RabbitTemplate集合*/private Map<String, RabbitTemplate> rabbitTemplateMap = new HashMap<>();/*** RabbitAdmin集合*/private Map<String, RabbitAdmin> rabbitAdminMap = new HashMap<>();/*** RabbitAdmin集合*/private Map<String, ConnectionFactory> connectionFactoryMap = new HashMap<>();@Autowiredpublic DynamicRabbitMQConfig(RabbitProperties rabbitProperties) {this.rabbitProperties = rabbitProperties;}/*** 初始化 RabbitMQ 实例** @PostConstruct 注解，在 Spring 容器初始化后执行*/@PostConstructpublic void init() {rabbitProperties.getInstances().forEach(this::createRabbitInstance);}/*** 创建一个 RabbitMQ 实例** @param instance 实例配置信息*/public void createRabbitInstance(RabbitInstance instance) {// 创建连接工厂CachingConnectionFactory connectionFactory = new CachingConnectionFactory();connectionFactory.setHost(instance.getHost());connectionFactory.setPort(instance.getPort());connectionFactory.setUsername(instance.getUsername());connectionFactory.setPassword(instance.getPassword());connectionFactory.setVirtualHost(instance.getVirtualHost());connectionFactoryMap.put(instance.getName(), connectionFactory);// 创建 RabbitTemplateRabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);// 设置消息转换器rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter(new ObjectMapper()));rabbitTemplateMap.put(instance.getName(), rabbitTemplate);// 创建 RabbitAdminRabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate);rabbitAdminMap.put(instance.getName(), rabbitAdmin);// 存储实例信息rabbitInstanceMap.put(instance.getName(), instance);}/*** 根据名称获取 RabbitTemplate** @param name 名称* @return RabbitTemplate*/public RabbitTemplate getRabbitTemplate(String name) {return rabbitTemplateMap.get(name);}/*** 根据名称获取 RabbitAdmin** @param name 名称* @return RabbitAdmin*/public RabbitAdmin getRabbitAdmin(String name) {return rabbitAdminMap.get(name);}/*** 根据名称获取 ConnectionFactory** @param name 名称* @return ConnectionFactory*/public ConnectionFactory getConnectionFactory(String name) {return connectionFactoryMap.get(name);}/*** 根据名称获取 RabbitTemplate 和 RabbitAdmin** @param name 名称* @return Map*/public Map<String, Object> getRabbitInstance(String name) {Map<String, Object> result = new HashMap<>();result.put("instance", rabbitInstanceMap.get(name));result.put("rabbitAdmin", rabbitAdminMap.get(name));result.put("rabbitTemplate", rabbitTemplateMap.get(name));result.put("connectionFactory", connectionFactoryMap.get(name));return result;}/*** 检查 RabbitMQ 实例是否存在** @param name 名称* @return Boolean*/public Boolean checkInstanceExist(String name) {return rabbitInstanceMap.containsKey(name);}/*** 获取RabbitMQ动态实例名称集合** @return RabbitMQ动态实例名称集合*/public List<String> getDynamicInstanceNames() {return new ArrayList<>(rabbitInstanceMap.keySet());}/*** 根据名称删除RabbitMQ实例** @param name 名称*/public void removeRabbitInstance(String name) {rabbitInstanceMap.remove(name);rabbitAdminMap.remove(name);rabbitTemplateMap.remove(name);connectionFactoryMap.remove(name);}/*** 根据name名称往指定 exchange 发送消息** @param name       名称* @param exchange   交换机名称* @param routingKey 路由键* @param message    消息内容*/public void sendMessage(String name, String exchange, String routingKey, String message) {try {RabbitTemplate rabbitTemplate = rabbitTemplateMap.get(name);rabbitTemplate.convertAndSend(exchange, routingKey, message);} catch (Exception e) {log.error("RabbitMQ实例对象：{}，发送消息失败，失败原因：{}", name, e.getMessage());log.error("异常堆栈信息：{}", e);}}/*** 根据name名称往指定 queue 发送消息** @param name    名称* @param queue   队列名称* @param message 消息内容*/public void sendMessage(String name, String queue, String message) {RabbitTemplate rabbitTemplate = rabbitTemplateMap.get(name);rabbitTemplate.convertAndSend(queue, message);}// 根据name名称获取rabbitAdmin，创建Direct交换机public void createDirectExchange(String name, String exchangeName) {RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);DirectExchange directExchange = new DirectExchange(exchangeName, true, false);rabbitAdmin.declareExchange(directExchange);log.info("RabbitMQ实例对象：{}，Direct 交换机创建成功: {}", name, exchangeName);}// 根据name名称获取rabbitAdmin，创建Fount交换机public void createFountExchange(String name, String exchangeName) {RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);DirectExchange fanoutExchange = new DirectExchange(exchangeName, true, false);rabbitAdmin.declareExchange(fanoutExchange);log.info("RabbitMQ实例对象：{}，Fanout 交换机创建成功: {}", name, exchangeName);}// 根据name名称获取rabbitAdmin，创建Topic交换机public void createTopicExchange(String name, String exchangeName) {RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);TopicExchange topicExchange = new TopicExchange(exchangeName, true, false);rabbitAdmin.declareExchange(topicExchange);log.info("RabbitMQ实例对象：{}，Topic 交换机创建成功: {}", name, exchangeName);}// 根据name名称获取rabbitAdmin，创建Headers交换机public void createHeadersExchange(String name, String exchangeName) {RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);HeadersExchange headersExchange = new HeadersExchange(exchangeName, true, false);rabbitAdmin.declareExchange(headersExchange);log.info("RabbitMQ实例对象：{}，Headers 交换机创建成功: {}", name, exchangeName);}// 根据name名称获取rabbitAdmin，创建队列public void createQueue(String name, String queueName) {RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);Queue queue = new Queue(queueName, true, false, false);rabbitAdmin.declareQueue(queue);log.info("RabbitMQ实例对象：{}，队列创建成功: {}", name, queueName);}// 根据name名称获取rabbitAdmin，删除队列public void deleteQueue(String name, String queueName) {RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);rabbitAdmin.deleteQueue(queueName);log.info("RabbitMQ实例对象：{}，队列删除成功: {}", name, queueName);}// 根据name名称获取rabbitAdmin，创建队列并监听public void createQueue(String name, String queueName, Boolean isListener) {RabbitAdmin rabbitAdmin = rabbitAdminMap.get(name);ConnectionFactory connectionFactory = connectionFactoryMap.get(name);Queue queue = new Queue(queueName, true, false, false);rabbitAdmin.declareQueue(queue);log.info("RabbitMQ实例对象：{}，队列创建成功: {}", name, queueName);if (isListener) {createListener(connectionFactory, queueName);}}/*** 创建监听器容器并启动监听** @param queueName 队列名称*/public void createListener(ConnectionFactory connectionFactory, String queueName) {// 创建并启动监听器容器SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(queueName);// 监听逻辑处理container.setMessageListener(new MessageListenerAdapter(new Object() {public void handleMessage(String message) {System.out.println("收到来自RabbitMQ中队列：" + queueName + " 队列的消息：" + message);}}));// 启动监听器容器container.start();System.out.println("RabbitMQ队列监听器已启动：" + queueName);}
}