一个简单可参考的API网关架构设计_系统架构图 api

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

int sobacklog = Integer.parseInt(AppConfigUtil.getValue("netty.sobacklog"));

ServerBootstrap b = new ServerBootstrap();

b.group(bossGroup, workerGroup)

.channel(NioServerSocketChannel.class)

.localAddress(new InetSocketAddress(this.portHTTP))

.option(ChannelOption.SO_BACKLOG, sobacklog)

.childHandler(new ChannelHandlerInitializer(null));

// 绑定端口

ChannelFuture f = b.bind(this.portHTTP).sync();

logger.info("HttpServer name is " + HttpServer.class.getName() + " started and listen on " + f.channel().localAddress());

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

@Override

protected void initChannel(SocketChannel ch) throws Exception {

ChannelPipeline p = ch.pipeline();

p.addLast(new HttpRequestDecoder());

p.addLast(new HttpResponseEncoder());

int maxContentLength = 2000;

try {

maxContentLength = Integer.parseInt(AppConfigUtil.getValue("netty.maxContentLength"));

} catch (Exception e) {

logger.warn("netty.maxContentLength 配置异常，系统默认为：2000KB");

}

p.addLast(new HttpObjectAggregator(maxContentLength * 1024));// HTTP 消息的合并处理

p.addLast(new HttpServerInboundHandler());

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

@Override

public void channelRead0(ChannelHandlerContext ctx, Object msg)

throws Exception {

try {

if (msg instanceof HttpRequest && msg instanceof HttpContent) {

CmptRequest cmptRequest = CmptRequestUtil.convert(ctx, msg);

CmptResult cmptResult = this.gatewayExecutor.execute(cmptRequest);

FullHttpResponse response = encapsulateResponse(cmptResult);

ctx.write(response);

ctx.flush();

}

} catch (Exception e) {

logger.error("网关入口异常，" \+ e.getMessage());

e.printStackTrace();

}

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

/**

* 微信 token 认证，token 格式：

* {appID:'',openID:'',timestamp:132525144172,sessionKey: ''}

*

public class WeixinAuthTokenCmpt extends AbstractCmpt {

private static Logger logger = LoggerFactory.getLogger(WeixinAuthTokenCmpt.class);

private final CmptResult SUCCESS_RESULT;

public WeixinAuthTokenCmpt() {

SUCCESS_RESULT = buildSuccessResult();

}

@Override

public CmptResult execute(CmptRequest request, Map<String, FieldDTO> config) {

if (logger.isDebugEnabled()) {

logger.debug("WeixinTokenCmpt ......");

}

CmptResult cmptResult = null;

//Token 认证超时间 (传入单位: 分)

long authTokenExpireTime = getAuthTokenExpireTime(config);

WeixinTokenDTO authTokenDTO = this.getAuthTokenDTO(request);

logger.debug("Token=" + authTokenDTO);

AuthTokenState authTokenState = validateToken(authTokenDTO, authTokenExpireTime);

switch (authTokenState) {

case ACCESS: {

cmptResult = SUCCESS_RESULT;

Map<String, String> header = new HashMap<>();

header.put(HeaderKeyConstants.HEADER\_APP\_ID_KEY, authTokenDTO.getAppID());

header.put(CmptHeaderKeyConstants.HEADER\_WEIXIN\_OPENID_KEY, authTokenDTO.getOpenID());

header.put(CmptHeaderKeyConstants.HEADER\_WEIXIN\_SESSION_KEY, authTokenDTO.getSessionKey());

cmptResult.setHeader(header);

break;

}

case EXPIRED: {

cmptResult = buildCmptResult(RespErrCode.AUTH\_TOKEN\_EXPIRED, "token 过期, 请重新获取 Token！");

break;

}

case INVALID: {

cmptResult = buildCmptResult(RespErrCode.AUTH\_INVALID\_TOKEN, "Token 无效！");

break;

}

}

return cmptResult;

}

...

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

public interface ICmpt {

    /**        

     * 组件执行入口

     *

     * @param request

     * @param config，组件实例的参数配置

     * @return

     */   

    CmptResult execute(CmptRequest request, Map<String, FieldDTO> config);

    /**

     * 销毁组件持有的特殊资源，比如线程。

     */

    void destroy();

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

public class FieldDTO {

private String title;

private String name;

private FieldType fieldType = FieldType.STRING;

private String defaultValue;

private boolean required;

private String regExp;

private String description;

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

public class CmptResult {

RespErrMsg respErrMsg;// 组件返回错误信息

private boolean passed;// 组件过滤是否通过

private byte\[\] data;// 组件返回数据

private Map<String, String> header = new HashMap<String, String>();// 透传后端服务响应头信息

private MediaType mediaType;// 返回响应数据类型

private Integer statusCode = 200;// 默认返回状态码为 200

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

public static ICmpt newInstance(final CmptDef cmptDef) {

    ICmpt cmpt = null;

    try {

        final String jarPath = getJarPath(cmptDef);

        if (logger.isDebugEnabled()) {

            logger.debug("尝试载入 jar 包,jar 包路径: " + jarPath);

        }

        // 加载依赖 jar

        CmptClassLoader cmptClassLoader = CmptClassLoaderManager.loadJar(jarPath, true);

        // 创建实例

        if (null != cmptClassLoader) {

            cmpt = LoadClassUtil.newObject(cmptDef.getFullQualifiedName(), ICmpt.class, cmptClassLoader);

        } else {

            logger.error("加载组件 jar 包失败! jarPath: " + jarPath);

        }

    } catch (Exception e) {

        logger.error("组件类加载失败，请检查类名和版本是否正确。ClassName=" + cmptDef.getFullQualifiedName() + ", Version=" + cmptDef.getVersion());

        e.printStackTrace();

    }

    return cmpt;

}

1

2

3

4

5

6

7

8

9

10

11

12

13

Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(groupKey))

        .andCommandKey(HystrixCommandKey.Factory.asKey(commandKey ))

        .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()

                // 舱壁隔离策略 - 信号量

                .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)

                // 设置每组 command 可以申请的信号量最大数

                .withExecutionIsolationSemaphoreMaxConcurrentRequests(CmptInvoker.maxSemaphore)

                /* 开启超时设置 */

                .withExecutionIsolationThreadInterruptOnTimeout(true)

                /* 超时时间设置 */

                .withExecutionIsolationThreadTimeoutInMilliseconds(timeout)

                .withCircuitBreakerEnabled(true)// 开启熔断

                .withCircuitBreakerSleepWindowInMilliseconds(Constants.DEFAULT_CIRCUIT_BREAKER_SLEEP_WINDOW_IN_MILLISECONDS)// 5 秒后会尝试闭合回路

1

2

3

4

5

6

7

PoolingHttpClientConnectionManager cmOfHttp = new PoolingHttpClientConnectionManager();

cmOfHttp.setMaxTotal(maxConn);

cmOfHttp.setDefaultMaxPerRoute(maxPerRoute);

httpClient = HttpClients.custom()

        .setConnectionManager(cmOfHttp)

        .setConnectionManagerShared(true)

        .build();