消息队列中报微服务异常 issues/I4977W

zhangdaiscott
1 parent ae917438
Showing 5 changed files with 96 additions and 10 deletions
jeecg-boot/jeecg-boot-base/jeecg-boot-base-tools/src/main/java/org/jeecg/common/config/mqtoken/TransmitUserTokenFilter.java
jeecg-boot/jeecg-boot-base/jeecg-boot-base-tools/src/main/java/org/jeecg/common/config/mqtoken/UserTokenContext.java
jeecg-boot/jeecg-boot-starter/jeecg-boot-starter-rabbitmq/src/main/java/org/jeecg/boot/starter/rabbitmq/client/RabbitMqClient.java
jeecg-boot/jeecg-boot-starter/jeecg-boot-starter-rabbitmq/src/main/java/org/jeecg/boot/starter/rabbitmq/config/RabbitMqConfig.java
jeecg-boot/jeecg-boot-starter/jeecg-boot-starter-rabbitmq/src/main/java/org/jeecg/boot/starter/rabbitmq/core/BaseRabbiMqHandler.java
+package org.jeecg.common.config.mqtoken;
+
+import javax.servlet.*;
+import javax.servlet.http.HttpServletRequest;
+import java.io.IOException;
+
+/**
+ * 存放token到上下文供队列调用feign使用
+ * @author zyf
+ */
+public class TransmitUserTokenFilter implements Filter {
+
+    private static String X_ACCESS_TOKEN="X-Access-Token";
+
+    public TransmitUserTokenFilter() {
+    }
+
+    @Override
+    public void init(FilterConfig filterConfig) throws ServletException {
+    }
+
+    @Override
+    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
+        this.initUserInfo((HttpServletRequest) request);
+        chain.doFilter(request, response);
+    }
+
+    private void initUserInfo(HttpServletRequest request) {
+        String token = request.getHeader(X_ACCESS_TOKEN);
+        if (token!=null) {
+            try {
+                //将token放入上下文中
+                UserTokenContext.setToken(token);
+            } catch (Exception e) {
+
+            }
+        }
+    }
+
+    @Override
+    public void destroy() {
+    }
+}
 \ No newline at end of file
+package org.jeecg.common.config.mqtoken;
+
+
+/**
+ * 用户token上下文
+ * @author zyf
+ */
+public class UserTokenContext {
+
+    private static ThreadLocal<String> userToken = new ThreadLocal<String>();
+
+    public UserTokenContext() {
+    }
+
+    public static String getToken(){
+        return userToken.get();
+    }
+
+    public static void setToken(String token){
+        userToken.set(token);
+    }
+}
@@ -28,6 +28,7 @@ import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
  
 /**
  * 消息队列客户端
@@ -91,11 +92,16 @@ public class RabbitMqClient {
         rabbitAdmin.declareExchange(directExchange);
         if (ObjectUtil.isNotEmpty(queues)) {
             for (String queueName : queues) {
-                Queue queue = new Queue(queueName);
-                addQueue(queue);
-                Binding binding = BindingBuilder.bind(queue).to(directExchange).with(queueName);
-                rabbitAdmin.declareBinding(binding);
-                log.info("队列创建成功:" + queueName);
+                Properties result = rabbitAdmin.getQueueProperties(queueName);
+                if (ObjectUtil.isEmpty(result)) {
+                    Queue queue = new Queue(queueName);
+                    addQueue(queue);
+                    Binding binding = BindingBuilder.bind(queue).to(directExchange).with(queueName);
+                    rabbitAdmin.declareBinding(binding);
+                    log.info("创建队列:" + queueName);
+                }else{
+                    log.info("已有队列:" + queueName);
+                }
             }
         }
     }
 package org.jeecg.boot.starter.rabbitmq.config;
  
  
+import java.util.UUID;
+
 import org.jeecg.boot.starter.rabbitmq.event.JeecgRemoteApplicationEvent;
-import org.jeecg.boot.starter.rabbitmq.exchange.DelayExchangeBuilder;
+import org.jeecg.common.config.mqtoken.TransmitUserTokenFilter;
 import org.springframework.amqp.core.AcknowledgeMode;
-import org.springframework.amqp.core.CustomExchange;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
@@ -13,8 +14,6 @@ import org.springframework.cloud.bus.jackson.RemoteApplicationEventScan;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
  
-import java.util.UUID;
-
 /**
  * 消息队列配置类
  *
@@ -33,7 +32,14 @@ public class RabbitMqConfig {
         return rabbitAdmin;
     }
  
-
+    /**
+     * 注入获取token过滤器
+     * @return
+     */
+    @Bean
+    public TransmitUserTokenFilter transmitUserInfoFromHttpHeader(){
+        return new TransmitUserTokenFilter();
+    }
  
     @Bean
     public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory) {
@@ -2,15 +2,24 @@ package org.jeecg.boot.starter.rabbitmq.core;
  
 import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
+
 import org.jeecg.boot.starter.rabbitmq.listenter.MqListener;
+import org.jeecg.common.config.mqtoken.UserTokenContext;
  
 import java.io.IOException;
  
+/**
+ *
+ * @author zyf
+ */
 @Slf4j
 public class BaseRabbiMqHandler<T> {
  
+    private String token= UserTokenContext.getToken();
+
     public void onMessage(T t, Long deliveryTag, Channel channel, MqListener mqListener) {
         try {
+            UserTokenContext.setToken(token);
             mqListener.handler(t, channel);
             channel.basicAck(deliveryTag, false);
         } catch (Exception e) {