今天我们踩到一个collect_list的坑，collect_list的结果不包含null值

对city作group by后collect_list(name)得到的结果中city='广州’为List(‘张三’)，没有null值。跟踪源码：

  def collect_list(e: Column): Column = withAggregateFunction { CollectList(e.expr) }

collect_list使用CollectList计算

@ExpressionDescription(
  usage = "_FUNC_(expr) - Collects and returns a list of non-unique elements.")
case class CollectList(
    child: Expression,
    mutableAggBufferOffset: Int = 0,
    inputAggBufferOffset: Int = 0) extends Collect[mutable.ArrayBuffer[Any]] {

  def this(child: Expression) = this(child, 0, 0)

  override lazy val bufferElementType = child.dataType

  override def convertToBufferElement(value: Any): Any = InternalRow.copyValue(value)

  override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
    copy(mutableAggBufferOffset = newMutableAggBufferOffset)

  override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
    copy(inputAggBufferOffset = newInputAggBufferOffset)

  override def createAggregationBuffer(): mutable.ArrayBuffer[Any] = mutable.ArrayBuffer.empty

  override def prettyName: String = "collect_list"

  override def eval(buffer: mutable.ArrayBuffer[Any]): Any = {
    new GenericArrayData(buffer.toArray)
  }
}

CollectList继承于Collect[mutable.ArrayBuffer[Any]]

/**
 * A base class for collect_list and collect_set aggregate functions.
 *
 * We have to store all the collected elements in memory, and so notice that too many elements
 * can cause GC paused and eventually OutOfMemory Errors.
 */
abstract class Collect[T <: Growable[Any] with Iterable[Any]] extends TypedImperativeAggregate[T] {

  val child: Expression

  override def children: Seq[Expression] = child :: Nil

  override def nullable: Boolean = true

  override def dataType: DataType = ArrayType(child.dataType)

  // Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
  // actual order of input rows.
  override lazy val deterministic: Boolean = false

  override def defaultResult: Option[Literal] = Option(Literal.create(Array(), dataType))

  protected def convertToBufferElement(value: Any): Any

  override def update(buffer: T, input: InternalRow): T = {
    val value = child.eval(input)

    // Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
    // See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
    if (value != null) {
      buffer += convertToBufferElement(value)
    }
    buffer
  }

  override def merge(buffer: T, other: T): T = {
    buffer ++= other
  }

  protected val bufferElementType: DataType

  private lazy val projection = UnsafeProjection.create(
    Array[DataType](ArrayType(elementType = bufferElementType, containsNull = false)))
  private lazy val row = new UnsafeRow(1)

  override def serialize(obj: T): Array[Byte] = {
    val array = new GenericArrayData(obj.toArray)
    projection.apply(InternalRow.apply(array)).getBytes()
  }

  override def deserialize(bytes: Array[Byte]): T = {
    val buffer = createAggregationBuffer()
    row.pointTo(bytes, bytes.length)
    row.getArray(0).foreach(bufferElementType, (_, x: Any) => buffer += x)
    buffer
  }
}

在update方法中可以看到判空的逻辑，注释内容说要与Hive的collect_list/collect_set方法保持一致。
如果collect_list再支持一个可选参数，用于控制是否过滤null就好了，于是我们自定义了collect_list函数

object CollectListWithNullUDAF extends UserDefinedAggregateFunction {

  // Data types of input arguments of this aggregate function
  override def inputSchema: StructType = StructType(StructField("element", StringType) :: Nil)

  // Data types of values in the aggregation buffer
  override def bufferSchema: StructType = StructType(StructField("buffer", ArrayType(StringType, containsNull = true)) :: Nil)

  // The data type of the returned value
  override def dataType: DataType = ArrayType(StringType, containsNull = true)

  // Whether this function always returns the same output on the identical input
  override def deterministic: Boolean = true

  // Initializes the given aggregation buffer. The buffer itself is a `Row` that in addition to
  // standard methods like retrieving a value at an index (e.g., get(), getBoolean()), provides
  // the opportunity to update its values. Note that arrays and maps inside the buffer are still
  // immutable.  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = List.empty[String]
  }

  // Updates the given aggregation buffer `buffer` with new input data from `input`
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    implicit val defaultFormats: DefaultFormats.type = org.json4s.DefaultFormats
    var list = buffer.get(0).asInstanceOf[mutable.WrappedArray[String]].toList

    val value = input.get(0)
    if (value == null) {
      list = list.+:(null)
    } else {
      list = list.+:(value.toString)
    }

    buffer(0) = list
  }

  // Merges two aggregation buffers and stores the updated buffer values back to `buffer1`
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    val listResult = buffer1.get(0).asInstanceOf[mutable.WrappedArray[String]].toList
    val listTemp = buffer2.get(0).asInstanceOf[mutable.WrappedArray[String]].toList

    buffer1(0) = listResult ++ listTemp
  }

  // Calculates the final result
  override def evaluate(buffer: Row): List[String] = {
    val list = buffer.get(0).asInstanceOf[mutable.WrappedArray[String]].reverse.toList
    list
  }
}

// 方法注册
spark.udf.register("collect_list_with_null", CollectListWithNullUDAF)

TcpClient tcpClient；  

byte[] byteBuffer= new byte[1024*4];

tcpClient.Client.BeginReceive(byteBuffer, 0, byteBuffer.Length, SocketFlags.None, ReceiveCallBack, null);

    // 参数:
        //   buffer:
        //     类型的数组 System.Byte ，它是接收到的数据的存储位置。

如上代码片段所示：当发送端以10ms的间隔频率密集发送数据时，接收方利用Thread.Sleep(10000) 刻意等待10s 后观察测试结果：

发送方发了382次，此时接收方数据先写入socket系统缓冲区累加了，但我们程序读取延时，所以只用了6次就全部读取完了。

如果把上面伪代码的 byteBuffer 设置为发送数据长度（测试发送的数据长度为48:）的一半：即 byte[] byteBuffer=new byte[24];

再次测试发现接收次数为发送次数的两倍，这个好理解：接收容量小了一半，次数就要增加一倍，此时和是否有Sleep无关。Sleep等待只是使得接收数据变慢了。

由上可知：这个buffer参数 太小，则接收太慢，如果参数过大，则可以最大化利用网络I/O性能，但相应速度可能下降，通常要保持一个合理的值，如1024*2 或1024*4 等；

附加一个Socket异步接收模型：参考了这个Blog 的封装，但实际测试有些bug,做了些改进：

比如连接时，不小心触发了多次连接就可能报异常：

System.InvalidOperationException:“同一个套接字上正在进行另一个异步操作时，不能调用 BeginConnect

另外测试发现一个比较严重的bug，如果客户端Socket异步发送频繁：比如：

  for(int i=0;i<(int)this.numericUpDown1.Value;i++)
            {
                PipeReader l100 = new PipeReader { PipeCode ="#"+ (i + 1).ToString("0000"), ProductType= ProductType.S45, IsNormalCode=true };
                client.SendAsync(l100.ToBytes());
            }        

此Socket异步模型会导致数据漏收，比如发了50个，实际只收到了2,3个，问题发生的原因在线程同步锁ManualResetEvent的使用导致，去掉这个锁的逻辑，可以接受完整；或者在客户端发送消息时加上延时，则也可接收完整的回调；

//友情提示：下面的封装异步Tcp客户端,可以拿来做测试用，如果实际项目利用需要谨慎使用
public class AsynSocketClient
    {
        #region private
        /// <summary>
        /// 用于控制异步接收消息
        /// </summary>
        // private ManualResetEvent doReceive = new ManualResetEvent(false);
        private TcpClient tcpClient;

        //标识客户端是否接收数据
        private bool isStopWork = false;
        #endregion

        /// <summary>
        /// 是否释放对象了
        /// </summary>
        public bool IsClosed { get => isStopWork; set => isStopWork = value; }

        #region 事件
        /// <summary>
        /// 客户端连接完成、发送完成、连接异常或者服务端关闭触发的事件
        /// </summary>
        public event Action<TcpClient, SocketAction> Completed;
        
        /// <summary>
        /// 客户端接收消息触发的事件
        /// </summary>
        public event EventHandler<DataEventArgs> Received;

        #endregion
        
        public AsynSocketClient()
        {
            tcpClient = new TcpClient();    
          
        }

        #region 连接
        /// <summary>
        /// 异步连接
        /// </summary>
        /// <param name="ip">要连接的服务器的ip地址</param>
        /// <param name="port">要连接的服务器的端口</param>
        public void ConnectAsync(string ip, int port)
        {
            IPAddress ipAddress = null;
            try
            {
                ipAddress = IPAddress.Parse(ip);
            }
            catch (Exception)
            {
                throw new Exception("ip地址格式不正确，请使用正确的ip地址！");
            }
            try
            {
                if (!tcpClient.Connected)
                {
                    tcpClient.BeginConnect(ipAddress, port, ConnectCallBack, tcpClient);
                }
                else if (isStopWork)
                {
                    isStopWork = false;
                    OnComplete(tcpClient, SocketAction.Connect);
                }
            }
            catch
            {
            }            
        }

        /// <summary>
        /// 异步连接的回调函数
        /// </summary>
        /// <param name="ar"></param>
        private void ConnectCallBack(IAsyncResult ar)
        {
            try
            {
                TcpClient client = ar.AsyncState as TcpClient;
                client.EndConnect(ar);
                OnComplete(client, SocketAction.Connect);
            }
            catch
            { }           
        }
        #endregion

        #region 收发数据
        /// <summary>
        /// 异步接收消息
        /// </summary>
        private void ReceiveAsync()
        {
           // doReceive.Reset();
            StateObject obj = new StateObject();
            obj.TcpClient = tcpClient;

            tcpClient.Client.BeginReceive(obj.ListData, 0, obj.ListData.Length, SocketFlags.None, ReceiveCallBack, obj);
           // doReceive.WaitOne();
        }
        /// <summary>
        /// 异步发送消息
        /// </summary>
        /// <param name="msg"></param>
        public void SendAsync(string msg)
        {
            byte[] listData = Encoding.UTF8.GetBytes(msg);
            SendAsync(listData);
        }

        public void SendAsync(byte[] bytes)
        {
            try
            {
                if (tcpClient.Client == null)
                {
                    throw new Exception("连接已经断开");
                }
                if (isStopWork)
                {
                    return;
                }
                tcpClient.Client.BeginSend(bytes, 0, bytes.Length, SocketFlags.None, SendCallBack, tcpClient);
            }
            catch (SocketException ex)
            {
                if (ex.ErrorCode == (int)SocketError.ConnectionAborted)
                {
                    throw new Exception("连接已经断开");
                }
            }
          
        }


        private void ReceiveCallBack(IAsyncResult ar)
        {
            StateObject state = ar.AsyncState as StateObject;
            int count = -1;
            try
            {
                if (isStopWork)
                {
                    return;
                }
                count = state.TcpClient.Client.EndReceive(ar);
                //doReceive.Set();
            }
            catch (Exception ex)
            {
                //如果发生异常，说明客户端失去连接，触发关闭事件
                Stop();
                OnComplete(state.TcpClient, SocketAction.Close);
            }
            if (count > 0)
            {
                Received?.Invoke(this, new DataEventArgs() { Data = state.ListData.Take<byte>(count).ToArray() });
            }
        }

        private void SendCallBack(IAsyncResult ar)
        {
            TcpClient client = ar.AsyncState as TcpClient;
            try
            {
                client.Client.EndSend(ar);
                OnComplete(client, SocketAction.SendMsg);
            }
            catch (Exception)
            {
                //如果发生异常，说明客户端失去连接，触发关闭事件
                Stop();
                OnComplete(client, SocketAction.Close);
            }
        }
        #endregion

        public virtual void OnComplete(TcpClient client, SocketAction action)
        {
            if (Completed != null)
                Completed(client, action);
            if (action == SocketAction.Connect)
            {
                // 接收数据
                ThreadPool.QueueUserWorkItem(x =>
                {
                    while (!isStopWork)
                    {
                        try
                        {
                            ReceiveAsync();
                            Thread.Sleep(20);
                        }
                        catch (Exception)
                        {
                            Stop();
                            OnComplete(client, SocketAction.Close);
                        }
                    }
                });
            }
            else if (action == SocketAction.Close)
            {
                try
                {
                    Thread.Sleep(50);
                    this.Received = null;
                    tcpClient.Close();
                }
                catch
                {
                }
            }
        }

        void Stop()
        {
            isStopWork = true;
        }

        public void Close()
        {
            if (tcpClient != null)
            {
                try
                {
                    Stop();
                    IsClosed = true;
                    Completed = null;
                    Received = null;
                    tcpClient.GetStream().Close();
                 
                }
                catch(Exception ex)
                {

                }
            }
        }

    }

    /// <summary>
    /// 接收socket的行为
    /// </summary>
    public enum SocketAction
    {
        /// <summary>
        /// socket发生连接
        /// </summary>
        Connect = 1,
        /// <summary>
        /// socket发送数据
        /// </summary>
        SendMsg = 2,
        /// <summary>
        /// socket关闭
        /// </summary>
        Close = 4
    }
    
    public class StateObject
    {
        public TcpClient TcpClient { get; set; }
        private byte[] listData = new byte[1024*4];
        /// <summary>
        /// 接收的数据
        /// </summary>
        public byte[] ListData
        {
            get
            {
                return listData;
            }
            set
            {
                listData = value;
            }
        }
    }

    public class DataEventArgs : EventArgs
    {
        public byte[] Data
        {
            get;
            set;
        }

        public int Offset
        {
            get;
            set;
        }

        public int Length
        {
            get;
            set;
        }
    }

转载大佬的文章: http://www.itmuch.com/spring-cloud-sum/hystrix-threadlocal/

自定义feign拦截器
 

package com.wm.commonfeign.config;

import feign.RequestInterceptor;
import feign.RequestTemplate;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.ObjectUtils;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.servlet.http.HttpServletRequest;
import java.util.Enumeration;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * @class_name: CustomFeignConfig
 * @description: 服务间调用携带authorization请求头,拦截器统一处理
 * @author: wm_yu
 * @create: 2019/07/26
 **/
@Configuration
@Slf4j
public class CustomFeignConfig implements RequestInterceptor {
    private static final String TOKEN_HEADER = "authorization";

    @Override
    public void apply(RequestTemplate requestTemplate) {
        HttpServletRequest request = getHttpServletRequest();
        //获取请求参数 TODO 获取请求参数失败  RequestInterceptor是在请求前给request添加参数,实际上参数并没有同步过来,待解决,建议使用feign官方的输出日志
        String jsonString = this.getRequestJsonString(request);
        //输出调用feign日志
        log.info("feign调用请求方法日志打印,methodName={},Parameter={}",request.getMethod(),jsonString);
        if (ObjectUtils.isEmpty(request)) {
            requestTemplate.header(TOKEN_HEADER, getHeaders(request).get(TOKEN_HEADER));
        }
    }


    /**
     * 获取请求的request
     * @return
     */
    private HttpServletRequest getHttpServletRequest() {
        try {
            // hystrix隔离策略会导致RequestContextHolder.getRequestAttributes()返回null
            // 解决方案：http://www.itmuch.com/spring-cloud-sum/hystrix-threadlocal/
            ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
            if (!ObjectUtils.isEmpty(attributes)){
                return attributes.getRequest();
            }
            return null;
        } catch (Exception e) {
            log.error("feign拦截器中获取HttpServletRequest error...",e.getMessage(),e);
            return null;
        }
    }

    /**
     * 获取所有的header中的参数
     * @param request
     * @return
     */
    private Map<String, String> getHeaders(HttpServletRequest request) {
        Map<String, String> map = new LinkedHashMap<>();
        Enumeration<String> enumeration = request.getHeaderNames();
        while (enumeration.hasMoreElements()) {
            String key = enumeration.nextElement();
            String value = request.getHeader(key);
            map.put(key, value);
        }
        return map;
    }


    /**
     * 获取request中的json数据
     * @param request
     * @return
     */
    public  String getRequestJsonString(HttpServletRequest request){
        try {

            int contentLength = request.getContentLength();
            if (contentLength < 0) {
                return null;
            }
            byte buffer[] = new byte[contentLength];
            for (int i = 0; i < contentLength;) {
                int readlen = request.getInputStream().read(buffer, i, contentLength - i);
                if (readlen == -1) {
                    break;
                }
                i += readlen;
            }

            String charEncoding = request.getCharacterEncoding();
            if (charEncoding == null) {
                charEncoding = "UTF-8";
            }
            return new String(buffer, charEncoding);
        } catch (Exception e) {
            log.error("解析request中的流异常....",e.getMessage(),e);
        }
        return null;
    }


}

途中代码处返回null

解决方法:自定义熔断策略

package com.wm.commonfeign.strategy;

import com.netflix.hystrix.HystrixThreadPoolKey;
import com.netflix.hystrix.HystrixThreadPoolProperties;
import com.netflix.hystrix.strategy.HystrixPlugins;
import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy;
import com.netflix.hystrix.strategy.concurrency.HystrixRequestVariable;
import com.netflix.hystrix.strategy.concurrency.HystrixRequestVariableLifecycle;
import com.netflix.hystrix.strategy.eventnotifier.HystrixEventNotifier;
import com.netflix.hystrix.strategy.executionhook.HystrixCommandExecutionHook;
import com.netflix.hystrix.strategy.metrics.HystrixMetricsPublisher;
import com.netflix.hystrix.strategy.properties.HystrixPropertiesStrategy;
import com.netflix.hystrix.strategy.properties.HystrixProperty;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @class_name: CustomFeignHystrixConcurrencyStrategy
 * @description: 自定义隔离策略 解决RequestContextHolder.getRequestAttributes()问题 为null
 * @author: wm_yu
 * @create: 2019/07/26
 **/
@Component
@Primary
@Slf4j
public class CustomFeignHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy{

    private HystrixConcurrencyStrategy hystrixConcurrencyStrategy;

    public CustomFeignHystrixConcurrencyStrategy() {
        try {
            this.hystrixConcurrencyStrategy = HystrixPlugins.getInstance().getConcurrencyStrategy();
            if (this.hystrixConcurrencyStrategy instanceof CustomFeignHystrixConcurrencyStrategy) {
                // Welcome to singleton hell...
                return;
            }
            HystrixCommandExecutionHook commandExecutionHook =
                    HystrixPlugins.getInstance().getCommandExecutionHook();
            HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance().getEventNotifier();
            HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance().getMetricsPublisher();
            HystrixPropertiesStrategy propertiesStrategy =
                    HystrixPlugins.getInstance().getPropertiesStrategy();
            this.logCurrentStateOfHystrixPlugins(eventNotifier, metricsPublisher, propertiesStrategy);
            HystrixPlugins.reset();
            HystrixPlugins.getInstance().registerConcurrencyStrategy(this);
            HystrixPlugins.getInstance().registerCommandExecutionHook(commandExecutionHook);
            HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
            HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
            HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
        } catch (Exception e) {
            log.error("Failed to register Sleuth Hystrix Concurrency Strategy", e);
        }
    }

    private void logCurrentStateOfHystrixPlugins(HystrixEventNotifier eventNotifier,
                                                 HystrixMetricsPublisher metricsPublisher, HystrixPropertiesStrategy propertiesStrategy) {
        if (log.isDebugEnabled()) {
            log.debug("Current Hystrix plugins configuration is [" + "concurrencyStrategy ["
                    + this.hystrixConcurrencyStrategy + "]," + "eventNotifier [" + eventNotifier + "]," + "metricPublisher ["
                    + metricsPublisher + "]," + "propertiesStrategy [" + propertiesStrategy + "]," + "]");
            log.debug("Registering Sleuth Hystrix Concurrency Strategy.");
        }
    }

    @Override
    public <T> Callable<T> wrapCallable(Callable<T> callable) {
        RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
        return new WrappedCallable<>(callable, requestAttributes);
    }

    @Override
    public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
                                            HystrixProperty<Integer> corePoolSize, HystrixProperty<Integer> maximumPoolSize,
                                            HystrixProperty<Integer> keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        return this.hystrixConcurrencyStrategy.getThreadPool(threadPoolKey, corePoolSize, maximumPoolSize, keepAliveTime,
                unit, workQueue);
    }

    @Override
    public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
                                            HystrixThreadPoolProperties threadPoolProperties) {
        return this.hystrixConcurrencyStrategy.getThreadPool(threadPoolKey, threadPoolProperties);
    }

    @Override
    public BlockingQueue<Runnable> getBlockingQueue(int maxQueueSize) {
        return this.hystrixConcurrencyStrategy.getBlockingQueue(maxQueueSize);
    }

    @Override
    public <T> HystrixRequestVariable<T> getRequestVariable(HystrixRequestVariableLifecycle<T> rv) {
        return this.hystrixConcurrencyStrategy.getRequestVariable(rv);
    }

    static class WrappedCallable<T> implements Callable<T> {
        private final Callable<T> target;
        private final RequestAttributes requestAttributes;

        public WrappedCallable(Callable<T> target, RequestAttributes requestAttributes) {
            this.target = target;
            this.requestAttributes = requestAttributes;
        }

        @Override
        public T call() throws Exception {
            try {
                RequestContextHolder.setRequestAttributes(requestAttributes);
                return target.call();
            } finally {
                RequestContextHolder.resetRequestAttributes();
            }
        }
    }
}

配置fegin支持hystrix,并指定熔断的超时

feign:
  hystrix:
    enabled:  true

# hystrix配置
hystrix:
  shareSecurityContext: true
  command:
    default:
      execution:
        isolation:
          thread:
            timeoutInMilliseconds: 60000

feign提供给其他服务调用
 

package com.wm.examapi.feign;

import com.wm.commoncore.constant.ServiceConstant;
import com.wm.commoncore.model.Response;
import com.wm.commonfeign.config.CustomFeignConfig;
import com.wm.examapi.feign.fallback.ExaminationServiceClientFallbackImpl;
import com.wm.examapi.model.Examination;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;

/**
 * @class_name: ExaminationServiceClient
 * @description: 考试接口 .... FeignClient需要指定configuration为CustomFeignConfig，用于服务间调用携带token
 * @author: wm_yu
 * @create: 2019/07/26
 **/
@FeignClient(name = ServiceConstant.EXAM_SERVICE, configuration = CustomFeignConfig.class, fallback = ExaminationServiceClientFallbackImpl.class)
public interface ExaminationServiceClient {
    /**
     * 查询考试数量
     *
     * @param examination 租户标识
     * @return ResponseBean
     */
    /**
     * 注意这里只能用RequestMapping指定请求方法为post,不能使用@PostMapping
     * @param examination
     * @return
     */
    @RequestMapping(method = RequestMethod.POST,value = "/api/exam/getExaminationCount")
    Response<Integer> findExaminationCount(@RequestBody Examination examination);
}

配置熔断类:

package com.wm.examapi.feign.fallback;

import com.wm.commoncore.constant.BussiConstant;
import com.wm.commoncore.model.Response;
import com.wm.examapi.feign.ExaminationServiceClient;
import com.wm.examapi.model.Examination;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestBody;


/**
 * @class_name: ExaminationServiceClientFallbackImpl
 * @description:  考试服务熔断
 * @author: wm_yu
 * @create: 2019/07/26
 **/
@Slf4j
@Component
public class ExaminationServiceClientFallbackImpl implements ExaminationServiceClient {

    private Throwable throwable;

    @Override
    public Response<Integer> findExaminationCount(@RequestBody Examination examination) {
        log.error("调用{}异常, {}， {}", "findExaminationCount", examination.getTenantCode(),throwable);
        return  Response.createError(BussiConstant.EXAM_SERVICE_FEIGN_ERROR,0);
    }

    public Throwable getThrowable() {
        return throwable;
    }

    public void setThrowable(Throwable throwable) {
        this.throwable = throwable;
    }
}

创建熔断工厂,生成熔断实例

package com.wm.examapi.feign.factory;

import com.wm.examapi.feign.ExaminationServiceClient;
import com.wm.examapi.feign.fallback.ExaminationServiceClientFallbackImpl;
import feign.hystrix.FallbackFactory;
import org.springframework.stereotype.Component;

/**
 * @class_name: UserServiceClientFallbackFactory
 * @description: 考试系统熔断器工厂
 * @author: wm_yu
 * @create: 2019/07/26
 **/
@Component
public class ExaminationServiceClientFallbackFactory implements FallbackFactory<ExaminationServiceClient> {
    @Override
    public ExaminationServiceClient create(Throwable throwable) {
        ExaminationServiceClientFallbackImpl examinationServiceClientFallback = new ExaminationServiceClientFallbackImpl();
        examinationServiceClientFallback.setThrowable(throwable);
        return examinationServiceClientFallback;
    }
}

启动类扫描feign包

package com.wm.userservice;

import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.web.cors.CorsConfiguration;
import org.springframework.web.cors.UrlBasedCorsConfigurationSource;
import org.springframework.web.filter.CorsFilter;

@SpringBootApplication
@EnableDiscoveryClient  //注册中心发现
@MapperScan("com.wm.userservice.mapper")
@EnableFeignClients(basePackages = {"com.wm.examapi.feign"}) //扫描fegin包
@ComponentScan({"com.wm.examapi.feign","com.wm.userservice","com.wm.commonfeign"})  //主要是需要扫描到内部的包（内部不同模块）,如:fegin配置等
public class UserServiceApplication {

    public static void main(String[] args) {
        SpringApplication.run(UserServiceApplication.class, args);
    }


    /**
     * 跨域过滤器
     * @return
     */
    @Bean
    public CorsFilter corsFilter() {
        UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
        source.registerCorsConfiguration("/**", buildConfig());
        return new CorsFilter(source);
    }

    private CorsConfiguration buildConfig() {
        CorsConfiguration corsConfiguration = new CorsConfiguration();
        corsConfiguration.addAllowedOrigin("*");
        corsConfiguration.addAllowedHeader("*");
        corsConfiguration.addAllowedMethod("*");
        return corsConfiguration;
    }

}

这里多配置一个vue跨域解决的

本来是想在拦截器中输出日志信息,发现不行,这个待处理,目前个人理解的原因的:

后面采用官方的feign日志输出:

很简单:

创建一个日志配置类:

如下:

package com.wm.commonfeign.config;

import feign.Logger;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @class_name: FeignLogConfig
 * @description: 配置feign日志输出
 * @author: wm_yu
 * @create: 2019/07/27
 **/

/**
 * Feign logging
 * A logger is created for each Feign client created. By default the name of the logger is the full class name of the interface used to create the Feign client. Feign logging only responds to the DEBUG level.
 *
 *   application.yml
 *
 * logging.level.project.user.UserClient: DEBUG
 *
 *
 * The Logger.Level object that you may configure per client, tells Feign how much to log. Choices are:
 *
 * 可以为每个客户端配置日志级别(Logger.Level)，选项有：
 *
 * NONE, No logging (DEFAULT).
 * BASIC, Log only the request method and URL and the response status code and execution time.
 * HEADERS, Log the basic information along with request and response headers.
 * FULL, Log the headers, body, and metadata for both requests and responses.
 * For example, the following would set the Logger.Level to FULL:
 *
 * 下例将日志级别设为全部(FULL)：
 *
 * @Configuration
 * public class FooConfiguration {
 *     @Bean
 *     Logger.Level feignLoggerLevel() {
 *         return Logger.Level.FULL;
 *     }
 * }
 */

@Configuration
public class FeignLogConfig {
        @Bean
        Logger.Level feignLoggerLevel() {
            return Logger.Level.FULL;
        }
}

注释掉的为官方文档中的解释

在yml中配置需要输出的日志级别和指明feign类

logging:
  level:
    root: info
     ####sql日志
    com.wm.examservice.mapper: debug
    #### feign日志
    com.wm.examapi.feign: debug