【AKKA干货】AKKA-HTTP（JAVA版）踩坑记
2024-10-09 11:05:20
因为不会屎克拉，所以只能使用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);
}
private Flow<Message, Message, NotUsed> socketFlow(ActorRef actor) {
// 背压支撑
Source<Message, NotUsed> source = Source.<String>actorRefWithBackpressure("ack", o -> {
if (o == "complete")
return Optional.of(CompletionStrategy.draining());
else
return Optional.empty();
}, o -> Optional.empty()).map(message -> (Message) TextMessage.create(message))
.mapMaterializedValue(textMessage -> {
actor.tell(textMessage, ActorRef.noSender());
return NotUsed.getInstance();
})
// .keepAlive(Duration.ofSeconds(10), () -> TextMessage.create("ping")) //这段代码可让服务端自动向客户端发送ping
;
Sink<Message, NotUsed> sink = Flow.<Message>create().to(Sink.actorRef(actor, PoisonPill.getInstance()));
return Flow.fromSinkAndSource(sink, source);
}
@Override
protected Route routes() {
// TODO Auto-generated method stub
return path("gateway", () -> get(() -> handleWebSocketMessages(websocketFlow))));
支持HTTPS
第一步:在HttpApp中实现useHttps
public HttpsConnectionContext useHttps(ActorSystem system) {
HttpsConnectionContext https = null;
try {
// initialise the keystore
// !!! never put passwords into code !!!
final char[] password = "123456789".toCharArray();
final KeyStore ks = KeyStore.getInstance("PKCS12");
final InputStream keystore = KutaHttpApp.class.getClassLoader().getResourceAsStream("xxx-xxx-com-akka-0709085814.pfx");
if (keystore == null) {
throw new RuntimeException("Keystore required!");
}
ks.load(keystore, password);
Enumeration<String> aliases = ks.aliases();
while(aliases.hasMoreElements()) {
String next = aliases.nextElement();
logger.info(next);
java.security.Key key = ks.getKey(next, password);
logger.info("alg:{},format:{}",key.getAlgorithm(),key.getFormat());
}
final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance("SunX509");
keyManagerFactory.init(ks, password);
final TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
tmf.init(ks);
final SSLContext sslContext = SSLContext.getInstance("TLS");
sslContext.init(keyManagerFactory.getKeyManagers(), tmf.getTrustManagers(), new SecureRandom());
https = ConnectionContext.https(sslContext);
} catch (NoSuchAlgorithmException | KeyManagementException e) {
system.log().error("Exception while configuring HTTPS.", e);
} catch (CertificateException | KeyStoreException | UnrecoverableKeyException | IOException e) {
system.log().error("Exception while ", e);
}
return https;
}
第二步:在main函数中注册
final Http http = Http.get(system);
HttpsConnectionContext https = app.useHttps(system);
http.setDefaultServerHttpContext(https);
Integer sslPort = PropertyUtils.getInteger("app", "gateway.ssl.port");
http.bindAndHandle(flow, ConnectHttp.toHost(host, sslPort), materializer);
RESTful实现
public Route RESTfulAsyncRouter() {
return path(PathMatchers.segment("RESTful"),()->{
complete("hello");
});
}
跨域支持
public Route RESTfulRouter() {
return path(PathMatchers.segment("RESTful"),
()-> concat(
post(()->{
return entity(Jackson.unmarshaller(JSONObject.class), json -> {
final JSONObject data = json;
final MessageDispatcher dispatcher = system.dispatchers().lookup(KSFConstants.BLOCKING_DISPATCHER);
try {
return
CompletableFuture.<Route>supplyAsync(()->{
final ActorRef RESTfull = system.actorOf(RESTfulActor.props(Globals.getHallRouter())
.withDispatcher(KSFConstants.BLOCKING_DISPATCHER));
// logger.info("当前运行线程:{}",Thread.currentThread());
CompletionStage<Optional<KutaJsonResponse>> rsp = Patterns
.ask(RESTfull, data, timeout)
.thenApply(a -> {
return Optional.of((KutaJsonResponse) a);
});
return onSuccess(() -> rsp, performed -> {
RESTfull.tell(PoisonPill.getInstance(), ActorRef.noSender());
List<HttpHeader> list = new ArrayList<>();
list.add(HttpHeader.parse("Access-Control-Allow-Origin", "*"));
list.add(HttpHeader.parse("Access-Control-Allow-Credentials", "true"));
list.add(HttpHeader.parse("Access-Control-Allow-Methods", "POST,OPTIONS"));
list.add(HttpHeader.parse("Access-Control-Expose-Headers", "Content-Type, Access-Control-Allow-Origin, Access-Control-Allow-Credentials"));
list.add(HttpHeader.parse("Access-Control-Allow-Headers", "Content-Type,Access-Token,Authorization"));
if (performed.isPresent()) {
if(performed.get().getSize() > (1024 * 50)) {
return encodeResponseWith(
Collections.singletonList(Coder.Gzip),
()->complete(StatusCodes.OK, list, performed.get(), Jackson.marshaller())
);
}
else {
return complete(StatusCodes.OK, list, performed.get(), Jackson.marshaller());
}
}
else {
return complete(StatusCodes.NOT_FOUND);
}
}).orElse(complete(StatusCodes.INTERNAL_SERVER_ERROR));
}, dispatcher).get();
} catch (InterruptedException | ExecutionException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return null;
});
})
, options(()->{
List<HttpHeader> list = new ArrayList<>();
list.add(HttpHeader.parse("Access-Control-Allow-Origin", "*"));
list.add(HttpHeader.parse("Access-Control-Allow-Credentials", "true"));
list.add(HttpHeader.parse("Access-Control-Allow-Methods", "POST,OPTIONS"));
list.add(HttpHeader.parse("Access-Control-Expose-Headers", "Content-Type, Access-Control-Allow-Origin, Access-Control-Allow-Credentials,Vary"));
list.add(HttpHeader.parse("Access-Control-Allow-Headers", "Content-Type,Access-Token,Authorization"));
return respondWithHeaders(list,()-> complete(""));
})
)
);
}
支持静态资源（屏蔽各种配置文件）
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
//我自己实现的HttpApp
KutaHttpApp app = new KutaHttpApp(system);
final Http http = Http.get(system);
final ActorMaterializer materializer = ActorMaterializer.create(system);
final Flow<HttpRequest, HttpResponse, NotUsed> flow = app.routes().flow(system, materializer);
//先绑定http
http.bindAndHandle(flow, ConnectHttp.toHost(host, port), materializer);
boolean useHttps = Boolean.parseBoolean(PropertyUtils.getProperty("app", "gateway.usessl"));
if (useHttps) {
HttpsConnectionContext https = app.useHttps(system);
http.setDefaultServerHttpContext(https);
Integer sslPort = PropertyUtils.getInteger("app", "gateway.ssl.port");
http.bindAndHandle(flow, ConnectHttp.toHost(host, sslPort), materializer);
logger.info("启动ssl服务.host:{},port:{}",host,sslPort);
}
将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的同学们吧。
最新文章
- MzBlog分析
- D3.js 做一个简单的图表（条形图）
- WCF 内存入口检查失败
- 查看memcached中最大生存时间
- C#_DBHelper_SQL数据库操作类.
- 解决easyui datagrid加载数据时，checkbox列没有根据checkbox的值来确定是否选中
- (转)asp.net(C#)手记之Repeater与两级菜单
- Niagara技术文档汇总
- hdu 1536 S-Nim （简单sg函数）
- 新发现的一些C函数
- Action3D
- 从源码解析TreeMap
- APP H5 混合自动化使用说明 [基于 Appium+Python 系列]
- 读Zepto源码之fx_methods模块
- jQuery中的DOM操作------复制及包裹节点
- 浅谈js中的this关键字
- Fine-tuning Convolutional Neural Networks for Biomedical Image Analysis: Actively and Incrementally如何使用尽可能少的标注数据来训练一个效果有潜力的分类器
- 实战三种方式部署 MySQL5.7
- 网站实时信息采集和统计graphite
- 在电脑中配置adb