专栏首页张泽旭的专栏springcloud gateway 自定义协议转化实现

springcloud gateway 自定义协议转化实现

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。

本文链接:https://blog.csdn.net/qq_17655941/article/details/103362115

这几天公司要求实现 springcloud gateway 自定义协议 需要对外统一提供http/https 的接口,但是后端有很多服务提供了不同的方式,包括dubbo 协议,和dubbo 上提供的各种访问协议等(dubbo服务上协议的支持),需要从网关直接代理去访问。仔细分析后,发现gateway 只有自带的几种协议 ,http/https 和 ws 这个几种 ps:这就很难受,网上也百度了许多没有找到具体的实现方案,或许没有人这样干吧,在需求的推动下,决定自己研究一把.

1、源码分析

在没有头绪的情况下,还是得从阅读源码开始,源码下载地址,然后配置官方提供的私服地址,因为源码引用的有些包是未发布版本,在项目下找到对应 .flattened-pom.xml 文件下,上面的地址就是私服地址。(我这个说几个需要用到的类,其他的自己看百度查看 推荐学习地址)

org.springframework.cloud.gateway.config.GatewayAutoConfiguration

这个类是对gateway GatewayFilter 的实例类注入,在源代码调试过程中要在这里添加自己自定义的GatewayFilter,不然用注解注入不能起作用。

org.springframework.cloud.gateway.filter.GlobalFilter

全局网关过滤器,是一个接口,自定义需要实现此接口(下面说明的类都实现此接口,并在同一个包中)

NettyRoutingFilter

@Override
	@SuppressWarnings("Duplicates")
	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
		URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);

		/**
		 * 获取访问的url地址
		 */
		String scheme = requestUrl.getScheme();
		/**
		 * 判断这次请求是否被处理过,请求协议是否是http/https
		 * 如果满足则不处理
		 */
		if (isAlreadyRouted(exchange)
				|| (!"http".equals(scheme) && !"https".equals(scheme))) {
			return chain.filter(exchange);
		}
		/**
		 * 开始处理,首先将状态标志为已处理
		 */
		setAlreadyRouted(exchange);


		/**
		 * 获取ServerHttpRequest,相当于HttpServletRequest请求,在gateway
		 * 中使用的是webflux 做的web服务,所有处理有点变化
		 */
		ServerHttpRequest request = exchange.getRequest();

		/**
		 * 获取 请求方式
		 */
		final HttpMethod method = HttpMethod.valueOf(request.getMethodValue());
		final String url = requestUrl.toASCIIString();
		System.out.println(url);
		/**
		 * 获取 HttpHeaders
		 */
		HttpHeaders filtered = filterRequest(getHeadersFilters(), exchange);

		final DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders();
		filtered.forEach(httpHeaders::set);

		boolean preserveHost = exchange
				.getAttributeOrDefault(PRESERVE_HOST_HEADER_ATTRIBUTE, false);


		/**
		 * httpClient 这次采用httpclient 去代理访问要去代理的地址,相当于网关去
		 * 帮你请求了一次
		 */
		Flux<HttpClientResponse> responseFlux = this.httpClient.headers(headers -> {
			headers.add(httpHeaders);
			/**
			 * 判断请求主机是否一致,一起是访问地址和路由地址是否是一个
			 * ip 不是则需要替换
			 */
			if (preserveHost) {
				String host = request.getHeaders().getFirst(HttpHeaders.HOST);
				headers.add(HttpHeaders.HOST, host);
			}
			else {
				// let Netty set it based on hostname
				headers.remove(HttpHeaders.HOST);
			}
		}).request(method).uri(url).send((req, nettyOutbound) -> {
			if (log.isTraceEnabled()) {
				nettyOutbound.withConnection(connection -> log.trace(
						"outbound route: " + connection.channel().id().asShortText()
								+ ", inbound: " + exchange.getLogPrefix()));
			}
			NettyOutbound outbound = nettyOutbound.send(request.getBody()
					.map(dataBuffer -> ((NettyDataBuffer) dataBuffer).getNativeBuffer()));

			return outbound;
		}).responseConnection((res, connection) -> {

			// Defer committing the response until all route filters have run
			// Put client response as ServerWebExchange attribute and write
			// response later NettyWriteResponseFilter
			exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
			exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);
			/**
			 * 异步获取 httpClient 访问的响应结果,下面就是对响应结果的封装
			 */

			ServerHttpResponse response = exchange.getResponse();
			// put headers and status so filters can modify the response
			HttpHeaders headers = new HttpHeaders();

			res.responseHeaders()
					.forEach(entry -> headers.add(entry.getKey(), entry.getValue()));

			String contentTypeValue = headers.getFirst(HttpHeaders.CONTENT_TYPE);
			if (StringUtils.hasLength(contentTypeValue)) {
				exchange.getAttributes().put(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR,
						contentTypeValue);
			}

			HttpStatus status = HttpStatus.resolve(res.status().code());
			if (status != null) {
				response.setStatusCode(status);
			}
			else if (response instanceof AbstractServerHttpResponse) {
				// https://jira.spring.io/browse/SPR-16748
				((AbstractServerHttpResponse) response)
						.setStatusCodeValue(res.status().code());
			}
			else {
				// TODO: log warning here, not throw error?
				throw new IllegalStateException("Unable to set status code on response: "
						+ res.status().code() + ", " + response.getClass());
			}

			// make sure headers filters run after setting status so it is
			// available in response
			HttpHeaders filteredResponseHeaders = HttpHeadersFilter
					.filter(getHeadersFilters(), headers, exchange, Type.RESPONSE);

			if (!filteredResponseHeaders.containsKey(HttpHeaders.TRANSFER_ENCODING)
					&& filteredResponseHeaders.containsKey(HttpHeaders.CONTENT_LENGTH)) {
				// It is not valid to have both the transfer-encoding header and
				// the content-length header.
				// Remove the transfer-encoding header in the response if the
				// content-length header is present.
				response.getHeaders().remove(HttpHeaders.TRANSFER_ENCODING);
			}

			exchange.getAttributes().put(CLIENT_RESPONSE_HEADER_NAMES,
					filteredResponseHeaders.keySet());

			response.getHeaders().putAll(filteredResponseHeaders);

			System.out.println(res+"///////");
			return Mono.just(res);
		});

		if (properties.getResponseTimeout() != null) {
			responseFlux = responseFlux.timeout(properties.getResponseTimeout(),
					Mono.error(new TimeoutException("Response took longer than timeout: "
							+ properties.getResponseTimeout())))
					.onErrorMap(TimeoutException.class,
							th -> new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT,
									th.getMessage(), th));
		}


		return responseFlux.then(chain.filter(exchange));
	}

NettyWriteResponseFilter(此类接收和处理上面的代理响应结果)

@Override
	public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
		// NOTICE: nothing in "pre" filter stage as CLIENT_RESPONSE_CONN_ATTR is not added
		// until the NettyRoutingFilter is run
		// @formatter:off

		return chain.filter(exchange)
				.doOnError(throwable -> cleanup(exchange))
				.then(Mono.defer(() -> {
					Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);

					if (connection == null) {
						return Mono.empty();
					}
					if (log.isTraceEnabled()) {
						log.trace("NettyWriteResponseFilter start inbound: "
								+ connection.channel().id().asShortText() + ", outbound: "
								+ exchange.getLogPrefix());
					}
					ServerHttpResponse response = exchange.getResponse();

					// TODO: what if it's not netty
					NettyDataBufferFactory factory = (NettyDataBufferFactory) response
							.bufferFactory();


					// TODO: needed? 对响应结果进一个解析,获取到的数据是一个NettyDataBuffer,可以百度下
					// 是netty的数据
					final Flux<NettyDataBuffer> body = connection
							.inbound()
							.receive()
							.retain()
							.map(factory::wrap);

					MediaType contentType = null;
					try {
						contentType = response.getHeaders().getContentType();
					}
					catch (Exception e) {
						if (log.isTraceEnabled()) {
							log.trace("invalid media type", e);
						}
					}

					return (isStreamingMediaType(contentType)
							? response.writeAndFlushWith(body.map(Flux::just))
							: response.writeWith(body));
				})).doOnCancel(() -> cleanup(exchange));
		// @formatter:on
	}

上面的代码变量和编写方式都和框架使用了webflux 有关 大家可以学习一下就很容易理解了阿里JAVA架构师详解Spring5新特性之WebFlux 我也是看了这个视频学会的 2个小时,感觉讲的挺好的

2、自定义代码实现

package com.neo.config;

import com.alibaba.fastjson.JSONObject;
import io.netty.buffer.UnpooledByteBufAllocator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.cloud.gateway.filter.headers.HttpHeadersFilter;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.util.StringUtils;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Method;
import java.net.URI;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;

import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.*;

/**
 * @ClassName DubboResponseGlobalFilter
 * @Desription 协议转换的过滤器类
 * @Author zhangzexu
 * @Date 2019/11/28 17:14
 * @Version V1.0
 */
@Configuration
public class DubboResponseGlobalFilter implements GlobalFilter, Ordered {

    @Value("${plugin.calssName}")
    private String className;

    private static Logger LOGGER = LoggerFactory.getLogger(DubboResponseGlobalFilter.class);
    private volatile List<HttpHeadersFilter> headersFilters;

    @Override
    public int getOrder() {
        return Ordered.LOWEST_PRECEDENCE;
    }

    public DubboResponseGlobalFilter() {

    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);


        final String scheme = requestUrl.getScheme();
        if (isAlreadyRouted(exchange) || "http".equals(scheme) || "https".equals(scheme) || "lb".equals(scheme) || "ws".equals(scheme)) {
            return chain.filter(exchange);
        }
        LOGGER.info("请求的url为{},协议为{}",requestUrl,scheme);
        setAlreadyRouted(exchange);
        /**
         * 获取请求的url 对路径进行重新编码
         */
        final String url = requestUrl.toASCIIString();

        Flux<DataBuffer> flux = exchange.getRequest().getBody();
        AtomicReference<byte[]> atomicReference = new AtomicReference<>();
        /**
         * 获取客户端请求的数据,body体
         */
        flux.subscribe(buffer -> {
            byte[] bytes = new byte[buffer.readableByteCount()];
            buffer.read(bytes);
            DataBufferUtils.release(buffer);
            atomicReference.set(bytes);
        });
        return chain.filter(exchange)
                .then(Mono.defer(() -> {
                    ServerHttpResponse response = exchange.getResponse();
                    return response.writeWith(Flux.create(sink -> {
                        NettyDataBufferFactory nettyDataBufferFactory = new NettyDataBufferFactory(new UnpooledByteBufAllocator(false));
                        JSONObject json = new JSONObject();
                        Class c = null;
                        DataBuffer dataBuffer = null;
                        String charset = "UTF-8";
                        try {
                            /**
                             * 初始化反射数据,将要调用的类反射获取,反射的类的名称结构,
                             * 用 dubbo 协议举例 
                             * 则插件的类名组合为 DubboGatewayImpl
                             */
                            StringBuilder sb = new StringBuilder(className);
                            sb.append(".");
                            char[] name = scheme.toCharArray();
                            name[0] -= 32;
                            sb.append(String.valueOf(name));
                            sb.append("GatewayPluginImpl");
                            c = Class.forName(sb.toString());
                            c.getMethods();
                            Method method = c.getMethod("send", String.class, byte[].class);
                            Object obj = c.getConstructor().newInstance();
                            Object result = method.invoke(obj, url, atomicReference.get());
                            HttpStatus status = HttpStatus.resolve(500);
                            /**
                             * 判断结果是否返回,如果没有数据则直接返回
                             */
                            if (result == null) {

                            } else {
                                json = (JSONObject) result;
                                status = HttpStatus.resolve(json.getInteger("code"));
                                json.remove("code");
                                /**
                                 * 获取字符集编码格式 默认 utf-8
                                 */
                                if (!StringUtils.isEmpty(json.getString("charset"))) {
                                    charset = json.getString("charset");
                                }
                            }
                            response.setStatusCode(status);
                            try {
                                dataBuffer = nettyDataBufferFactory.wrap(json.toJSONString().getBytes(charset));
                            } catch (UnsupportedEncodingException e) {
                                dataBuffer = nettyDataBufferFactory.wrap(e.toString().getBytes(charset));
                                LOGGER.error("返回调用请求数据错误{}",e);
                                e.printStackTrace();
                            }
                        } catch (Exception e) {
                            try {
                                dataBuffer = nettyDataBufferFactory.wrap(e.toString().getBytes(charset));
                                LOGGER.error("获取远程数据错误{}",e);
                            } catch (UnsupportedEncodingException ex) {
                                ex.printStackTrace();
                                LOGGER.error("返回调用请求数据错误{}",ex);
                            }
                            e.printStackTrace();
                        }
                        /**
                         * 将数据进行发射到下一个过滤器
                         */
                        sink.next(dataBuffer);
                        sink.complete();
                    }));

                }));
    }
}

通过反射机制类完成除过gateway 自定义协议外的所有解析进行处理。

这就完了,简单把 具体协议插件实现可以下载源代码

完整项目下载 github

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 【腾讯云的1001种玩法】hadoop 伪分布式搭建

    Hadoop实现了一个分布式文件系统(简称HDFS)。HDFS有高容错性的特点,并且设计用来部署在低廉的(low-cost)硬件上;而且它提供高吞吐量(high...

    张泽旭
  • docker swarm部署SpringCloud微服务

    码云  : https://gitee.com/zhangzexu/blog-parent.git

    张泽旭
  • Spring Cloud Ribbon负载均衡策略自定义配置

    实现这个类,可以改变Consts.ruleType中的值,来每次动态选择负载均衡策略,其中倍权和tag轮询策略是我更具上述的随机轮询策略编写的,Consts类中...

    张泽旭
  • FreeMarker整合Springmvc

    用户5927264
  • Spring Cloud Zuul 综合使用

    从上图中可以看到,Zuul是我们整个系统的入口。当我们有参数校验的需求时,我们就可以利用Zuul的Pre过滤器,进行参数的校验。例如我现在希望请求都一律带上to...

    端碗吹水
  • poiAndEasyExcel学习(二)

    崔笑颜
  • Python实现的特征提取操作示例

    更多关于Python相关内容感兴趣的读者可查看本站专题:《Python数据结构与算法教程》、《Python编码操作技巧总结》、《Python函数使用技巧总结》、...

    砸漏
  • Python进行特征提取

    陆勤_数据人网
  • C++核心准则​SL.io.3:优先使用iostream进行I / O

    iostreams are safe, flexible, and extensible.

    面向对象思考
  • 6.2 Spring Boot集成jpa6.2 Spring Boot集成jpa小结

    Java持久化API(JPA,Java Persistence API)是一个将对象映射为关系数据库的标准技术。JPA通过注解或XML描述ORM(Object ...

    一个会写诗的程序员

扫码关注云+社区

领取腾讯云代金券