mirror of https://github.com/jeecgboot/jeecg-boot
消息队列中报微服务异常 issues/I4977W
parent
537cc05601
commit
4b830b37c9
|
@ -0,0 +1,43 @@
|
|||
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() {
|
||||
}
|
||||
}
|
|
@ -0,0 +1,22 @@
|
|||
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) {
|
||||
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);
|
||||
log.info("创建队列:" + queueName);
|
||||
}else{
|
||||
log.info("已有队列:" + queueName);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -1,10 +1,11 @@
|
|||
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) {
|
||||
|
|
Loading…
Reference in New Issue