feat(all): 修改mq发送消息根据租户切换的问题

blade-service/logpm-distribution/src/main/java/com/logpm/distribution/config/CustomMessageConverter.java

@@ -1,4 +1,4 @@
-package com.logpm.factorydata.olo.config;
+package com.logpm.distribution.config;
 
 import cn.hutool.json.JSONUtil;
 import org.springblade.core.secure.BladeUser;

blade-service/logpm-factory-data/logpm-factory-data-base/src/main/java/com/logpm/factorydata/base/config/CustomMessageConverter.java

@@ -6,24 +6,100 @@ import org.springblade.core.secure.utils.AuthUtil;
 import org.springblade.core.tool.utils.ThreadLocalUtil;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.AllowedListDeserializingMessageConverter;
 import org.springframework.amqp.support.converter.MessageConversionException;
-import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.amqp.utils.SerializationUtils;
+import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.http.HttpHeaders;
 import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.remoting.rmi.CodebaseAwareObjectInputStream;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ClassUtils;
 import org.springframework.web.context.request.RequestContextHolder;
 import org.springframework.web.context.request.ServletRequestAttributes;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectStreamClass;
+import java.io.Serializable;
+import java.io.UnsupportedEncodingException;
 import java.util.Map;
 
 /**
  * @author zhaoqiaobo
  * @create 2024-05-08
  */
-public class CustomMessageConverter implements MessageConverter {
-	private String name;
+@Component
+public class CustomMessageConverter extends AllowedListDeserializingMessageConverter implements BeanClassLoaderAware {
+
+	private volatile String defaultCharset = "UTF-8";
+	private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
+	private String codebaseUrl;
+
+	@Override
+	public Object fromMessage(Message message) throws MessageConversionException {
+		Object content = null;
+		MessageProperties properties = message.getMessageProperties();
+		if (properties != null) {
+			String contentType = properties.getContentType();
+			if (contentType != null && contentType.startsWith("text")) {
+				String encoding = properties.getContentEncoding();
+				if (encoding == null) {
+					encoding = "UTF-8";
+				}
+
+				try {
+					content = new String(message.getBody(), encoding);
+				} catch (UnsupportedEncodingException var8) {
+					throw new MessageConversionException("failed to convert text-based Message content", var8);
+				}
+			} else if (contentType != null && contentType.equals("application/x-java-serialized-object")) {
+				try {
+					content = SerializationUtils.deserialize(this.createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
+				} catch (IllegalArgumentException | IllegalStateException | IOException var7) {
+					throw new MessageConversionException("failed to convert serialized Message content", var7);
+				}
+			}
+		}
+		Map<String, Object> headers = properties.getHeaders();
+		ThreadLocalUtil.put("bladeContext", headers);
+		Object bladeUser = headers.get("bladeUser");
+		MockHttpServletRequest request = new MockHttpServletRequest();
+		BladeUser bladeUser1 = JSONUtil.toBean(bladeUser.toString(), BladeUser.class);
+		request.setAttribute("_BLADE_USER_REQUEST_ATTR_", bladeUser1);
+		RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
+		if (content == null) {
+			content = message.getBody();
+		}
+		return content;
+	}
 
 	@Override
-	public Message toMessage(Object o, MessageProperties messageProperties) throws MessageConversionException {
+	protected Message createMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
+		byte[] bytes = null;
+		if (object instanceof byte[]) {
+			bytes = (byte[]) object;
+			messageProperties.setContentType("application/octet-stream");
+		} else if (object instanceof String) {
+			try {
+				bytes = ((String) object).getBytes(this.defaultCharset);
+			} catch (UnsupportedEncodingException var6) {
+				throw new MessageConversionException("failed to convert to Message content", var6);
+			}
+
+			messageProperties.setContentType("text/plain");
+			messageProperties.setContentEncoding(this.defaultCharset);
+		} else if (object instanceof Serializable) {
+			try {
+				bytes = SerializationUtils.serialize(object);
+			} catch (IllegalArgumentException var5) {
+				throw new MessageConversionException("failed to convert to serialized Message content", var5);
+			}
+
+			messageProperties.setContentType("application/x-java-serialized-object");
+		}
 		HttpHeaders headers = (HttpHeaders) ThreadLocalUtil.get("bladeContext");
 		if (headers != null && !headers.isEmpty()) {
 			headers.forEach((key, values) -> {
@@ -39,27 +115,28 @@ public class CustomMessageConverter implements MessageConverter {
 		bladeUser.setAccount(user.getAccount());
 		bladeUser.setRoleId(user.getRoleId());
 		messageProperties.setHeader("bladeUser", JSONUtil.toJsonStr(bladeUser));
-		return new Message(o.toString().getBytes(), messageProperties);
-	}
 
-	@Override
-	public Object fromMessage(Message message) throws MessageConversionException {
-		MessageProperties messageProperties = message.getMessageProperties();
-		Map<String, Object> headers = messageProperties.getHeaders();
-		ThreadLocalUtil.put("bladeContext", headers);
-		Object bladeUser = headers.get("bladeUser");
-		MockHttpServletRequest request = new MockHttpServletRequest();
-		BladeUser bladeUser1 = JSONUtil.toBean(bladeUser.toString(), BladeUser.class);
-		request.setAttribute("_BLADE_USER_REQUEST_ATTR_", bladeUser1);
-		RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
-		return new String(message.getBody());
+		if (bytes != null) {
+			messageProperties.setContentLength((long) bytes.length);
+			return new Message(bytes, messageProperties);
+		} else {
+			throw new IllegalArgumentException(this.getClass().getSimpleName() + " only supports String, byte[] and Serializable payloads, received: " + object.getClass().getName());
+		}
 	}
 
-	public String getName() {
-		return name;
+	@Override
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.beanClassLoader = beanClassLoader;
 	}
 
-	public void setName(String name) {
-		this.name = name;
+	protected ObjectInputStream createObjectInputStream(InputStream is, String codebaseUrl) throws IOException {
+		return new CodebaseAwareObjectInputStream(is, this.beanClassLoader, codebaseUrl) {
+			@Override
+			protected Class<?> resolveClass(ObjectStreamClass classDesc) throws IOException, ClassNotFoundException {
+				Class<?> clazz = super.resolveClass(classDesc);
+				CustomMessageConverter.this.checkAllowedList(clazz);
+				return clazz;
+			}
+		};
 	}
 }

blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/aspect/AnnotationAspect.java

@@ -0,0 +1,45 @@
+package com.logpm.factorydata.jinpai.aspect;
+
+import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.springblade.core.secure.utils.AuthUtil;
+import org.springframework.stereotype.Component;
+
+@Aspect
+@Component
+public class AnnotationAspect {
+
+	/**
+	 * 该方法为一个切面，用于对标记了@RabbitListener注解的方法进行增强，主要增加了日志记录和数据源上下文处理的功能。
+	 *
+	 * @param joinPoint 切面的连接点，表示当前被拦截的方法。
+	 * @return 返回被拦截方法的执行结果。
+	 * @throws Throwable 如果执行过程中发生异常，则抛出。
+	 */
+	@Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitListener)")
+	public Object logAroundAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable {
+		// 在方法执行前，获取当前租户ID并设置到数据源上下文中
+		String tenantId = AuthUtil.getTenantId();
+		DynamicDataSourceContextHolder.push(tenantId);
+		// 执行原方法
+		Object result = joinPoint.proceed();
+		// 在方法执行后，从数据源上下文中移除租户ID
+		DynamicDataSourceContextHolder.poll();
+		return result;
+	}
+
+	@Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitHandler)")
+	public Object rabbitHandlerAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable {
+		// 在方法执行前，获取当前租户ID并设置到数据源上下文中
+		String tenantId = AuthUtil.getTenantId();
+		DynamicDataSourceContextHolder.push(tenantId);
+		// 执行原方法
+		Object result = joinPoint.proceed();
+		// 在方法执行后，从数据源上下文中移除租户ID
+		DynamicDataSourceContextHolder.poll();
+		return result;
+	}
+
+}

blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/CustomMessageConverter.java

@@ -0,0 +1,142 @@
+package com.logpm.factorydata.jinpai.config;
+
+import cn.hutool.json.JSONUtil;
+import org.springblade.core.secure.BladeUser;
+import org.springblade.core.secure.utils.AuthUtil;
+import org.springblade.core.tool.utils.ThreadLocalUtil;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.AllowedListDeserializingMessageConverter;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.utils.SerializationUtils;
+import org.springframework.beans.factory.BeanClassLoaderAware;
+import org.springframework.http.HttpHeaders;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.remoting.rmi.CodebaseAwareObjectInputStream;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ClassUtils;
+import org.springframework.web.context.request.RequestContextHolder;
+import org.springframework.web.context.request.ServletRequestAttributes;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectStreamClass;
+import java.io.Serializable;
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+/**
+ * @author zhaoqiaobo
+ * @create 2024-05-08
+ */
+@Component
+public class CustomMessageConverter extends AllowedListDeserializingMessageConverter implements BeanClassLoaderAware {
+
+	private volatile String defaultCharset = "UTF-8";
+	private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
+	private String codebaseUrl;
+
+	@Override
+	public Object fromMessage(Message message) throws MessageConversionException {
+		Object content = null;
+		MessageProperties properties = message.getMessageProperties();
+		if (properties != null) {
+			String contentType = properties.getContentType();
+			if (contentType != null && contentType.startsWith("text")) {
+				String encoding = properties.getContentEncoding();
+				if (encoding == null) {
+					encoding = "UTF-8";
+				}
+
+				try {
+					content = new String(message.getBody(), encoding);
+				} catch (UnsupportedEncodingException var8) {
+					throw new MessageConversionException("failed to convert text-based Message content", var8);
+				}
+			} else if (contentType != null && contentType.equals("application/x-java-serialized-object")) {
+				try {
+					content = SerializationUtils.deserialize(this.createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
+				} catch (IllegalArgumentException | IllegalStateException | IOException var7) {
+					throw new MessageConversionException("failed to convert serialized Message content", var7);
+				}
+			}
+		}
+		Map<String, Object> headers = properties.getHeaders();
+		ThreadLocalUtil.put("bladeContext", headers);
+		Object bladeUser = headers.get("bladeUser");
+		MockHttpServletRequest request = new MockHttpServletRequest();
+		BladeUser bladeUser1 = JSONUtil.toBean(bladeUser.toString(), BladeUser.class);
+		request.setAttribute("_BLADE_USER_REQUEST_ATTR_", bladeUser1);
+		RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
+		if (content == null) {
+			content = message.getBody();
+		}
+		return content;
+	}
+
+	@Override
+	protected Message createMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
+		byte[] bytes = null;
+		if (object instanceof byte[]) {
+			bytes = (byte[]) object;
+			messageProperties.setContentType("application/octet-stream");
+		} else if (object instanceof String) {
+			try {
+				bytes = ((String) object).getBytes(this.defaultCharset);
+			} catch (UnsupportedEncodingException var6) {
+				throw new MessageConversionException("failed to convert to Message content", var6);
+			}
+
+			messageProperties.setContentType("text/plain");
+			messageProperties.setContentEncoding(this.defaultCharset);
+		} else if (object instanceof Serializable) {
+			try {
+				bytes = SerializationUtils.serialize(object);
+			} catch (IllegalArgumentException var5) {
+				throw new MessageConversionException("failed to convert to serialized Message content", var5);
+			}
+
+			messageProperties.setContentType("application/x-java-serialized-object");
+		}
+		HttpHeaders headers = (HttpHeaders) ThreadLocalUtil.get("bladeContext");
+		if (headers != null && !headers.isEmpty()) {
+			headers.forEach((key, values) -> {
+				values.forEach((value) -> {
+					messageProperties.setHeader(key, new String[]{value});
+				});
+			});
+		}
+		BladeUser user = AuthUtil.getUser();
+		BladeUser bladeUser = new BladeUser();
+		bladeUser.setTenantId(user.getTenantId());
+		bladeUser.setUserId(user.getUserId());
+		bladeUser.setAccount(user.getAccount());
+		bladeUser.setRoleId(user.getRoleId());
+		messageProperties.setHeader("bladeUser", JSONUtil.toJsonStr(bladeUser));
+
+		if (bytes != null) {
+			messageProperties.setContentLength((long) bytes.length);
+			return new Message(bytes, messageProperties);
+		} else {
+			throw new IllegalArgumentException(this.getClass().getSimpleName() + " only supports String, byte[] and Serializable payloads, received: " + object.getClass().getName());
+		}
+	}
+
+	@Override
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.beanClassLoader = beanClassLoader;
+	}
+
+	protected ObjectInputStream createObjectInputStream(InputStream is, String codebaseUrl) throws IOException {
+		return new CodebaseAwareObjectInputStream(is, this.beanClassLoader, codebaseUrl) {
+			@Override
+			protected Class<?> resolveClass(ObjectStreamClass classDesc) throws IOException, ClassNotFoundException {
+				Class<?> clazz = super.resolveClass(classDesc);
+				CustomMessageConverter.this.checkAllowedList(clazz);
+				return clazz;
+			}
+		};
+	}
+}

blade-service/logpm-factory-data/logpm-factory-data-jinpai/src/main/java/com/logpm/factorydata/jinpai/config/RabbitConfig.java

@@ -0,0 +1,17 @@
+package com.logpm.factorydata.jinpai.config;
+
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RabbitConfig {
+
+	@Bean
+	public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
+		RabbitTemplate template = new RabbitTemplate(factory);
+		template.setMessageConverter(new CustomMessageConverter());
+		return template;
+	}
+}

blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/aspect/AnnotationAspect.java

@@ -30,4 +30,16 @@ public class AnnotationAspect {
 		return result;
 	}
 
+	@Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitHandler)")
+	public Object rabbitHandlerAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable {
+		// 在方法执行前，获取当前租户ID并设置到数据源上下文中
+		String tenantId = AuthUtil.getTenantId();
+		DynamicDataSourceContextHolder.push(tenantId);
+		// 执行原方法
+		Object result = joinPoint.proceed();
+		// 在方法执行后，从数据源上下文中移除租户ID
+		DynamicDataSourceContextHolder.poll();
+		return result;
+	}
+
 }

blade-service/logpm-factory-data/logpm-factory-data-olo/src/main/java/com/logpm/factorydata/olo/config/RabbitConfig.java

@@ -1,6 +1,5 @@
 package com.logpm.factorydata.olo.config;
 
-import com.logpm.factorydata.olo.config.CustomMessageConverter;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.context.annotation.Bean;
@@ -10,9 +9,9 @@ import org.springframework.context.annotation.Configuration;
 public class RabbitConfig {
 
 	@Bean
-	public RabbitTemplate rabbitTemplate(ConnectionFactory factory, CustomMessageConverter customMessageConverter) {
+	public RabbitTemplate rabbitTemplate(ConnectionFactory factory, CustomMessageConverter converter) {
 		RabbitTemplate template = new RabbitTemplate(factory);
-		template.setMessageConverter(customMessageConverter);
+		template.setMessageConverter(converter);
 		return template;
 	}
 }

blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/aspect/AnnotationAspect.java

@@ -0,0 +1,45 @@
+package com.logpm.factorydata.suofeiya.aspect;
+
+import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.springblade.core.secure.utils.AuthUtil;
+import org.springframework.stereotype.Component;
+
+@Aspect
+@Component
+public class AnnotationAspect {
+
+	/**
+	 * 该方法为一个切面，用于对标记了@RabbitListener注解的方法进行增强，主要增加了日志记录和数据源上下文处理的功能。
+	 *
+	 * @param joinPoint 切面的连接点，表示当前被拦截的方法。
+	 * @return 返回被拦截方法的执行结果。
+	 * @throws Throwable 如果执行过程中发生异常，则抛出。
+	 */
+	@Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitListener)")
+	public Object logAroundAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable {
+		// 在方法执行前，获取当前租户ID并设置到数据源上下文中
+		String tenantId = AuthUtil.getTenantId();
+		DynamicDataSourceContextHolder.push(tenantId);
+		// 执行原方法
+		Object result = joinPoint.proceed();
+		// 在方法执行后，从数据源上下文中移除租户ID
+		DynamicDataSourceContextHolder.poll();
+		return result;
+	}
+
+	@Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitHandler)")
+	public Object rabbitHandlerAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable {
+		// 在方法执行前，获取当前租户ID并设置到数据源上下文中
+		String tenantId = AuthUtil.getTenantId();
+		DynamicDataSourceContextHolder.push(tenantId);
+		// 执行原方法
+		Object result = joinPoint.proceed();
+		// 在方法执行后，从数据源上下文中移除租户ID
+		DynamicDataSourceContextHolder.poll();
+		return result;
+	}
+
+}

blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/CustomMessageConverter.java

@@ -0,0 +1,142 @@
+package com.logpm.factorydata.suofeiya.config;
+
+import cn.hutool.json.JSONUtil;
+import org.springblade.core.secure.BladeUser;
+import org.springblade.core.secure.utils.AuthUtil;
+import org.springblade.core.tool.utils.ThreadLocalUtil;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.AllowedListDeserializingMessageConverter;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.utils.SerializationUtils;
+import org.springframework.beans.factory.BeanClassLoaderAware;
+import org.springframework.http.HttpHeaders;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.remoting.rmi.CodebaseAwareObjectInputStream;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ClassUtils;
+import org.springframework.web.context.request.RequestContextHolder;
+import org.springframework.web.context.request.ServletRequestAttributes;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectStreamClass;
+import java.io.Serializable;
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+/**
+ * @author zhaoqiaobo
+ * @create 2024-05-08
+ */
+@Component
+public class CustomMessageConverter extends AllowedListDeserializingMessageConverter implements BeanClassLoaderAware {
+
+	private volatile String defaultCharset = "UTF-8";
+	private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
+	private String codebaseUrl;
+
+	@Override
+	public Object fromMessage(Message message) throws MessageConversionException {
+		Object content = null;
+		MessageProperties properties = message.getMessageProperties();
+		if (properties != null) {
+			String contentType = properties.getContentType();
+			if (contentType != null && contentType.startsWith("text")) {
+				String encoding = properties.getContentEncoding();
+				if (encoding == null) {
+					encoding = "UTF-8";
+				}
+
+				try {
+					content = new String(message.getBody(), encoding);
+				} catch (UnsupportedEncodingException var8) {
+					throw new MessageConversionException("failed to convert text-based Message content", var8);
+				}
+			} else if (contentType != null && contentType.equals("application/x-java-serialized-object")) {
+				try {
+					content = SerializationUtils.deserialize(this.createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
+				} catch (IllegalArgumentException | IllegalStateException | IOException var7) {
+					throw new MessageConversionException("failed to convert serialized Message content", var7);
+				}
+			}
+		}
+		Map<String, Object> headers = properties.getHeaders();
+		ThreadLocalUtil.put("bladeContext", headers);
+		Object bladeUser = headers.get("bladeUser");
+		MockHttpServletRequest request = new MockHttpServletRequest();
+		BladeUser bladeUser1 = JSONUtil.toBean(bladeUser.toString(), BladeUser.class);
+		request.setAttribute("_BLADE_USER_REQUEST_ATTR_", bladeUser1);
+		RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
+		if (content == null) {
+			content = message.getBody();
+		}
+		return content;
+	}
+
+	@Override
+	protected Message createMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
+		byte[] bytes = null;
+		if (object instanceof byte[]) {
+			bytes = (byte[]) object;
+			messageProperties.setContentType("application/octet-stream");
+		} else if (object instanceof String) {
+			try {
+				bytes = ((String) object).getBytes(this.defaultCharset);
+			} catch (UnsupportedEncodingException var6) {
+				throw new MessageConversionException("failed to convert to Message content", var6);
+			}
+
+			messageProperties.setContentType("text/plain");
+			messageProperties.setContentEncoding(this.defaultCharset);
+		} else if (object instanceof Serializable) {
+			try {
+				bytes = SerializationUtils.serialize(object);
+			} catch (IllegalArgumentException var5) {
+				throw new MessageConversionException("failed to convert to serialized Message content", var5);
+			}
+
+			messageProperties.setContentType("application/x-java-serialized-object");
+		}
+		HttpHeaders headers = (HttpHeaders) ThreadLocalUtil.get("bladeContext");
+		if (headers != null && !headers.isEmpty()) {
+			headers.forEach((key, values) -> {
+				values.forEach((value) -> {
+					messageProperties.setHeader(key, new String[]{value});
+				});
+			});
+		}
+		BladeUser user = AuthUtil.getUser();
+		BladeUser bladeUser = new BladeUser();
+		bladeUser.setTenantId(user.getTenantId());
+		bladeUser.setUserId(user.getUserId());
+		bladeUser.setAccount(user.getAccount());
+		bladeUser.setRoleId(user.getRoleId());
+		messageProperties.setHeader("bladeUser", JSONUtil.toJsonStr(bladeUser));
+
+		if (bytes != null) {
+			messageProperties.setContentLength((long) bytes.length);
+			return new Message(bytes, messageProperties);
+		} else {
+			throw new IllegalArgumentException(this.getClass().getSimpleName() + " only supports String, byte[] and Serializable payloads, received: " + object.getClass().getName());
+		}
+	}
+
+	@Override
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.beanClassLoader = beanClassLoader;
+	}
+
+	protected ObjectInputStream createObjectInputStream(InputStream is, String codebaseUrl) throws IOException {
+		return new CodebaseAwareObjectInputStream(is, this.beanClassLoader, codebaseUrl) {
+			@Override
+			protected Class<?> resolveClass(ObjectStreamClass classDesc) throws IOException, ClassNotFoundException {
+				Class<?> clazz = super.resolveClass(classDesc);
+				CustomMessageConverter.this.checkAllowedList(clazz);
+				return clazz;
+			}
+		};
+	}
+}

blade-service/logpm-factory-data/logpm-factory-data-suofeiya/src/main/java/com/logpm/factorydata/suofeiya/config/RabbitConfig.java

@@ -0,0 +1,17 @@
+package com.logpm.factorydata.suofeiya.config;
+
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RabbitConfig {
+
+	@Bean
+	public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
+		RabbitTemplate template = new RabbitTemplate(factory);
+		template.setMessageConverter(new CustomMessageConverter());
+		return template;
+	}
+}

blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/aspect/AnnotationAspect.java

@@ -0,0 +1,45 @@
+package com.logpm.factorydata.zbom.aspect;
+
+import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.springblade.core.secure.utils.AuthUtil;
+import org.springframework.stereotype.Component;
+
+@Aspect
+@Component
+public class AnnotationAspect {
+
+	/**
+	 * 该方法为一个切面，用于对标记了@RabbitListener注解的方法进行增强，主要增加了日志记录和数据源上下文处理的功能。
+	 *
+	 * @param joinPoint 切面的连接点，表示当前被拦截的方法。
+	 * @return 返回被拦截方法的执行结果。
+	 * @throws Throwable 如果执行过程中发生异常，则抛出。
+	 */
+	@Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitListener)")
+	public Object logAroundAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable {
+		// 在方法执行前，获取当前租户ID并设置到数据源上下文中
+		String tenantId = AuthUtil.getTenantId();
+		DynamicDataSourceContextHolder.push(tenantId);
+		// 执行原方法
+		Object result = joinPoint.proceed();
+		// 在方法执行后，从数据源上下文中移除租户ID
+		DynamicDataSourceContextHolder.poll();
+		return result;
+	}
+
+	@Around("@annotation(org.springframework.amqp.rabbit.annotation.RabbitHandler)")
+	public Object rabbitHandlerAsyncMethods(ProceedingJoinPoint joinPoint) throws Throwable {
+		// 在方法执行前，获取当前租户ID并设置到数据源上下文中
+		String tenantId = AuthUtil.getTenantId();
+		DynamicDataSourceContextHolder.push(tenantId);
+		// 执行原方法
+		Object result = joinPoint.proceed();
+		// 在方法执行后，从数据源上下文中移除租户ID
+		DynamicDataSourceContextHolder.poll();
+		return result;
+	}
+
+}

blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/CustomMessageConverter.java

@@ -0,0 +1,142 @@
+package com.logpm.factorydata.zbom.config;
+
+import cn.hutool.json.JSONUtil;
+import org.springblade.core.secure.BladeUser;
+import org.springblade.core.secure.utils.AuthUtil;
+import org.springblade.core.tool.utils.ThreadLocalUtil;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.AllowedListDeserializingMessageConverter;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.utils.SerializationUtils;
+import org.springframework.beans.factory.BeanClassLoaderAware;
+import org.springframework.http.HttpHeaders;
+import org.springframework.mock.web.MockHttpServletRequest;
+import org.springframework.remoting.rmi.CodebaseAwareObjectInputStream;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ClassUtils;
+import org.springframework.web.context.request.RequestContextHolder;
+import org.springframework.web.context.request.ServletRequestAttributes;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectStreamClass;
+import java.io.Serializable;
+import java.io.UnsupportedEncodingException;
+import java.util.Map;
+
+/**
+ * @author zhaoqiaobo
+ * @create 2024-05-08
+ */
+@Component
+public class CustomMessageConverter extends AllowedListDeserializingMessageConverter implements BeanClassLoaderAware {
+
+	private volatile String defaultCharset = "UTF-8";
+	private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
+	private String codebaseUrl;
+
+	@Override
+	public Object fromMessage(Message message) throws MessageConversionException {
+		Object content = null;
+		MessageProperties properties = message.getMessageProperties();
+		if (properties != null) {
+			String contentType = properties.getContentType();
+			if (contentType != null && contentType.startsWith("text")) {
+				String encoding = properties.getContentEncoding();
+				if (encoding == null) {
+					encoding = "UTF-8";
+				}
+
+				try {
+					content = new String(message.getBody(), encoding);
+				} catch (UnsupportedEncodingException var8) {
+					throw new MessageConversionException("failed to convert text-based Message content", var8);
+				}
+			} else if (contentType != null && contentType.equals("application/x-java-serialized-object")) {
+				try {
+					content = SerializationUtils.deserialize(this.createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
+				} catch (IllegalArgumentException | IllegalStateException | IOException var7) {
+					throw new MessageConversionException("failed to convert serialized Message content", var7);
+				}
+			}
+		}
+		Map<String, Object> headers = properties.getHeaders();
+		ThreadLocalUtil.put("bladeContext", headers);
+		Object bladeUser = headers.get("bladeUser");
+		MockHttpServletRequest request = new MockHttpServletRequest();
+		BladeUser bladeUser1 = JSONUtil.toBean(bladeUser.toString(), BladeUser.class);
+		request.setAttribute("_BLADE_USER_REQUEST_ATTR_", bladeUser1);
+		RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request));
+		if (content == null) {
+			content = message.getBody();
+		}
+		return content;
+	}
+
+	@Override
+	protected Message createMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
+		byte[] bytes = null;
+		if (object instanceof byte[]) {
+			bytes = (byte[]) object;
+			messageProperties.setContentType("application/octet-stream");
+		} else if (object instanceof String) {
+			try {
+				bytes = ((String) object).getBytes(this.defaultCharset);
+			} catch (UnsupportedEncodingException var6) {
+				throw new MessageConversionException("failed to convert to Message content", var6);
+			}
+
+			messageProperties.setContentType("text/plain");
+			messageProperties.setContentEncoding(this.defaultCharset);
+		} else if (object instanceof Serializable) {
+			try {
+				bytes = SerializationUtils.serialize(object);
+			} catch (IllegalArgumentException var5) {
+				throw new MessageConversionException("failed to convert to serialized Message content", var5);
+			}
+
+			messageProperties.setContentType("application/x-java-serialized-object");
+		}
+		HttpHeaders headers = (HttpHeaders) ThreadLocalUtil.get("bladeContext");
+		if (headers != null && !headers.isEmpty()) {
+			headers.forEach((key, values) -> {
+				values.forEach((value) -> {
+					messageProperties.setHeader(key, new String[]{value});
+				});
+			});
+		}
+		BladeUser user = AuthUtil.getUser();
+		BladeUser bladeUser = new BladeUser();
+		bladeUser.setTenantId(user.getTenantId());
+		bladeUser.setUserId(user.getUserId());
+		bladeUser.setAccount(user.getAccount());
+		bladeUser.setRoleId(user.getRoleId());
+		messageProperties.setHeader("bladeUser", JSONUtil.toJsonStr(bladeUser));
+
+		if (bytes != null) {
+			messageProperties.setContentLength((long) bytes.length);
+			return new Message(bytes, messageProperties);
+		} else {
+			throw new IllegalArgumentException(this.getClass().getSimpleName() + " only supports String, byte[] and Serializable payloads, received: " + object.getClass().getName());
+		}
+	}
+
+	@Override
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.beanClassLoader = beanClassLoader;
+	}
+
+	protected ObjectInputStream createObjectInputStream(InputStream is, String codebaseUrl) throws IOException {
+		return new CodebaseAwareObjectInputStream(is, this.beanClassLoader, codebaseUrl) {
+			@Override
+			protected Class<?> resolveClass(ObjectStreamClass classDesc) throws IOException, ClassNotFoundException {
+				Class<?> clazz = super.resolveClass(classDesc);
+				CustomMessageConverter.this.checkAllowedList(clazz);
+				return clazz;
+			}
+		};
+	}
+}

blade-service/logpm-factory-data/logpm-factory-data-zbom/src/main/java/com/logpm/factorydata/zbom/config/RabbitConfig.java

@@ -0,0 +1,17 @@
+package com.logpm.factorydata.zbom.config;
+
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class RabbitConfig {
+
+	@Bean
+	public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
+		RabbitTemplate template = new RabbitTemplate(factory);
+		template.setMessageConverter(new CustomMessageConverter());
+		return template;
+	}
+}

blade-service/logpm-factory/src/main/java/com/logpm/factory/config/CustomMessageConverter.java

@@ -1,4 +1,4 @@
-package com.logpm.factorydata.olo.config;
+package com.logpm.factory.config;
 
 import cn.hutool.json.JSONUtil;
 import org.springblade.core.secure.BladeUser;

blade-service/logpm-trunkline/src/main/java/com/logpm/trunkline/config/CustomMessageConverter.java

@@ -1,4 +1,4 @@
-package com.logpm.factorydata.olo.config;
+package com.logpm.trunkline.config;
 
 import cn.hutool.json.JSONUtil;
 import org.springblade.core.secure.BladeUser;

blade-service/logpm-warehouse/src/main/java/com/logpm/warehouse/config/CustomMessageConverter.java

@@ -1,4 +1,4 @@
-package com.logpm.factorydata.olo.config;
+package com.logpm.warehouse.config;
 
 import cn.hutool.json.JSONUtil;
 import org.springblade.core.secure.BladeUser;