【AKKA干貨】AKKA-HTTP（JAVA版）踩坑記-有解無憂

Kuta Service Framework(KSF) 基于akka的分布式集群框架

https://github.com/kutasms/kuta-service-framework

===================================================================================

因為不會scala，所以只能使用java版本，

國內AKKA的中文資料實在太少，想要找解決方案真心頭大，特別是對我這種英文差的小白來說實在痛苦，

===================================================================================

先公布一下我的AKKA-HTTP的性能測驗資料吧，

測驗環境：華為云 2核4G 云耀云服務器

單機簡單GET回傳資料（hello world）并發可達 30000+

請求轉到AKKA集群環境輸出并發可達 23000-28000 （因為我在轉發之前使用了AES解密，可能有影響）

AKKA-HTTP的幾個難點主要集中在以下幾點：

1. 支持websocket

2. 支持https(TSL）

3. RESTful實作

4. 跨域支持

5. 支持靜態資源（屏蔽各種組態檔）

6. 一個行程同時系結HTTP和HTTPS

7. 將http請求異步接管到akka集群

===================================================================================

好了，直接上干貨吧，

支持websocket

private Flow<Message, Message, NotUsed> websocketFlow() {
    ActorRef actor =     
        system.actorOf(GatewayActor.props(Globals.getGameRouter())); //gameRouter是集群路由Group
    return socketFlow(actor);
}

 1 private Flow<Message, Message, NotUsed> socketFlow(ActorRef actor) {
 2         // 背壓支撐
 3     Source<Message, NotUsed> source = Source.<String>actorRefWithBackpressure("ack", o -> {
 4         if (o == "complete")
 5         return Optional.of(CompletionStrategy.draining());
 6             else
 7         return Optional.empty();
 8         }, o -> Optional.empty()).map(message -> (Message) TextMessage.create(message))
 9                 .mapMaterializedValue(textMessage -> {
10                     actor.tell(textMessage, ActorRef.noSender());
11                     return NotUsed.getInstance();
12                 })
13         // .keepAlive(Duration.ofSeconds(10), () -> TextMessage.create("ping")) //這段代碼可讓服務端自動向客戶端發送ping
14         ;
15 
16         Sink<Message, NotUsed> sink = Flow.<Message>create().to(Sink.actorRef(actor, PoisonPill.getInstance()));
17 
18         return Flow.fromSinkAndSource(sink, source);
19     }

@Override
protected Route routes() {
    // TODO Auto-generated method stub
    return path("gateway", () -> get(() -> handleWebSocketMessages(websocketFlow))));

支持HTTPS

第一步:在HttpApp中實作useHttps

 1 public HttpsConnectionContext useHttps(ActorSystem system) {
 2         HttpsConnectionContext https = null;
 3         try {
 4           // initialise the keystore
 5           // !!! never put passwords into code !!!
 6           final char[] password = "123456789".toCharArray();
 7 
 8           final KeyStore ks = KeyStore.getInstance("PKCS12");
 9           final InputStream keystore = KutaHttpApp.class.getClassLoader().getResourceAsStream("xxx-xxx-com-akka-0709085814.pfx");
10           if (keystore == null) {
11             throw new RuntimeException("Keystore required!");
12           }
13           ks.load(keystore, password);
14           Enumeration<String> aliases = ks.aliases();
15             while(aliases.hasMoreElements()) {
16                 String next = aliases.nextElement();
17                 logger.info(next);
18                 java.security.Key key = ks.getKey(next, password);
19                 logger.info("alg:{},format:{}",key.getAlgorithm(),key.getFormat());
20             }
21           
22           final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance("SunX509");
23           keyManagerFactory.init(ks, password);
24 
25           final TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
26           tmf.init(ks);
27 
28           final SSLContext sslContext = SSLContext.getInstance("TLS");
29           sslContext.init(keyManagerFactory.getKeyManagers(), tmf.getTrustManagers(), new SecureRandom());
30 
31           https = ConnectionContext.https(sslContext);
32 
33         } catch (NoSuchAlgorithmException | KeyManagementException e) {
34           system.log().error("Exception while configuring HTTPS.", e);
35         } catch (CertificateException | KeyStoreException | UnrecoverableKeyException | IOException e) {
36           system.log().error("Exception while ", e);
37         }
38 
39         return https;
40     }

第二步:在main函式中注冊

1 final Http http = Http.get(system);
2 HttpsConnectionContext https = app.useHttps(system);
3 http.setDefaultServerHttpContext(https);
4 Integer sslPort = PropertyUtils.getInteger("app", "gateway.ssl.port");
5 http.bindAndHandle(flow, ConnectHttp.toHost(host, sslPort), materializer);

RESTful實作

public Route RESTfulAsyncRouter() {
    return path(PathMatchers.segment("RESTful"),()->{
          complete("hello");
    });
}

跨域支持

 1 public Route RESTfulRouter() {
 2         return path(PathMatchers.segment("RESTful"), 
 3                 ()-> concat(
 4                         post(()->{
 5                             return entity(Jackson.unmarshaller(JSONObject.class), json -> {
 6                                  final JSONObject data =https://www.cnblogs.com/kutasms/p/ json;
 7                                  final MessageDispatcher dispatcher = system.dispatchers().lookup(KSFConstants.BLOCKING_DISPATCHER);
 8                                  try {
 9                                     return 
10                                              CompletableFuture.<Route>supplyAsync(()->{
11                                                     final ActorRef RESTfull = system.actorOf(RESTfulActor.props(Globals.getHallRouter())
12                                                             .withDispatcher(KSFConstants.BLOCKING_DISPATCHER));
13 //                                                    logger.info("當前運行執行緒:{}",Thread.currentThread());
14                                                     CompletionStage<Optional<KutaJsonResponse>> rsp = Patterns
15                                                             .ask(RESTfull, data, timeout)
16                                                             .thenApply(a -> {
17                                                                 return Optional.of((KutaJsonResponse) a);
18                                                             });
19                                                     return onSuccess(() -> rsp, performed -> {
20                                                         RESTfull.tell(PoisonPill.getInstance(), ActorRef.noSender());
21                                                         List<HttpHeader> list = new ArrayList<>();
22                                                         list.add(HttpHeader.parse("Access-Control-Allow-Origin", "*"));
23                                                         list.add(HttpHeader.parse("Access-Control-Allow-Credentials", "true"));
24                                                         list.add(HttpHeader.parse("Access-Control-Allow-Methods", "POST,OPTIONS"));
25                                                         list.add(HttpHeader.parse("Access-Control-Expose-Headers", "Content-Type, Access-Control-Allow-Origin, Access-Control-Allow-Credentials"));
26                                                         list.add(HttpHeader.parse("Access-Control-Allow-Headers", "Content-Type,Access-Token,Authorization"));
27                                                         if (performed.isPresent()) {
28                                                             if(performed.get().getSize() > (1024 * 50)) {
29                                                                 return encodeResponseWith(
30                                                                         Collections.singletonList(Coder.Gzip),
31                                                                         ()->complete(StatusCodes.OK, list, performed.get(), Jackson.marshaller())
32                                                                         );
33                                                             }
34                                                             else {
35                                                                 return complete(StatusCodes.OK, list, performed.get(), Jackson.marshaller());
36                                                             }
37                                                         }
38                                                         else {
39                                                             return complete(StatusCodes.NOT_FOUND);
40                                                         }
41                                                 }).orElse(complete(StatusCodes.INTERNAL_SERVER_ERROR));
42                                                 }, dispatcher).get();
43                                 } catch (InterruptedException | ExecutionException e) {
44                                     // TODO Auto-generated catch block
45                                     e.printStackTrace();
46                                 }
47                                 return null;
48                             });
49                         })
50                         , options(()->{
51                             List<HttpHeader> list = new ArrayList<>();
52                             list.add(HttpHeader.parse("Access-Control-Allow-Origin", "*"));
53                             list.add(HttpHeader.parse("Access-Control-Allow-Credentials", "true"));
54                             list.add(HttpHeader.parse("Access-Control-Allow-Methods", "POST,OPTIONS"));
55                             list.add(HttpHeader.parse("Access-Control-Expose-Headers", "Content-Type, Access-Control-Allow-Origin, Access-Control-Allow-Credentials,Vary"));
56                             list.add(HttpHeader.parse("Access-Control-Allow-Headers", "Content-Type,Access-Token,Authorization"));
57                             return respondWithHeaders(list,()-> complete(""));
58                         })
59                         )
60                 );
61     }

支持靜態資源（屏蔽各種組態檔）

public Route staticResourceRouter() {
        return path(PathMatchers.remaining(), remain -> get(()-> {
            if(remain.endsWith(".properties") || remain.endsWith(".xml") || remain.endsWith(".conf")) {
                return complete(StatusCodes.UNAUTHORIZED);
            }
            return getFromResource(remain);
            }));
    }

將資源檔案放到你的classPath目錄下即可

一個行程同時系結HTTP和HTTPS

 1 //我自己實作的HttpApp
 2 KutaHttpApp app = new KutaHttpApp(system);
 3 final Http http = Http.get(system);
 4 final ActorMaterializer materializer = ActorMaterializer.create(system);
 5 final Flow<HttpRequest, HttpResponse, NotUsed> flow = app.routes().flow(system, materializer);
 6 //先系結http
 7 http.bindAndHandle(flow, ConnectHttp.toHost(host, port), materializer);
 8 boolean useHttps = Boolean.parseBoolean(PropertyUtils.getProperty("app", "gateway.usessl")); 
 9 if (useHttps) {
10     HttpsConnectionContext https = app.useHttps(system);
11     http.setDefaultServerHttpContext(https);
12     Integer sslPort = PropertyUtils.getInteger("app", "gateway.ssl.port");
13     http.bindAndHandle(flow, ConnectHttp.toHost(host, sslPort), materializer);
14     logger.info("啟動ssl服務.host:{},port:{}",host,sslPort);
15 }

將http請求異步接管到akka集群

public Route RESTfulAsyncRouter() {
        return path(PathMatchers.segment("RESTful"),()->{
            return entity(Jackson.unmarshaller(JSONObject.class), json -> {
                final Set<HttpHeader> headers = new HashSet<>();
                return completeWith(Marshaller.entityToOKResponse(headers, Jackson.<KutaJsonResponse>marshaller()), f->{
                    system.actorOf(RESTfulAsyncActor.props(json, f));
                });
            });
        });
    }

研究了好久才研究出來，希望能幫助到正在使用akka-http的同學們吧，

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/100212.html

標籤：Java

上一篇：輕松實作java攔截器+自定義注解

下一篇：shiro+jwt 實作權限控制