package com.xxl.job.core.server;
|
|
import com.xxl.job.core.biz.ExecutorBiz;
|
import com.xxl.job.core.biz.impl.ExecutorBizImpl;
|
import com.xxl.job.core.biz.model.*;
|
import com.xxl.job.core.thread.ExecutorRegistryThread;
|
import com.xxl.job.core.util.GsonTool;
|
import com.xxl.job.core.util.ThrowableUtil;
|
import com.xxl.job.core.util.XxlJobRemotingUtil;
|
import io.netty.bootstrap.ServerBootstrap;
|
import io.netty.buffer.Unpooled;
|
import io.netty.channel.*;
|
import io.netty.channel.nio.NioEventLoopGroup;
|
import io.netty.channel.socket.SocketChannel;
|
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
import io.netty.handler.codec.http.*;
|
import io.netty.handler.timeout.IdleStateEvent;
|
import io.netty.handler.timeout.IdleStateHandler;
|
import io.netty.util.CharsetUtil;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
|
import java.util.concurrent.*;
|
|
/**
|
* Copy from : https://github.com/xuxueli/xxl-rpc
|
*
|
* @author xuxueli 2020-04-11 21:25
|
*/
|
public class EmbedServer {
|
private static final Logger logger = LoggerFactory.getLogger(EmbedServer.class);
|
|
private ExecutorBiz executorBiz;
|
private Thread thread;
|
|
public void start(final String address, final int port, final String appname, final String accessToken) {
|
executorBiz = new ExecutorBizImpl();
|
thread = new Thread(new Runnable() {
|
|
@Override
|
public void run() {
|
|
// param
|
EventLoopGroup bossGroup = new NioEventLoopGroup();
|
EventLoopGroup workerGroup = new NioEventLoopGroup();
|
ThreadPoolExecutor bizThreadPool = new ThreadPoolExecutor(
|
0,
|
200,
|
60L,
|
TimeUnit.SECONDS,
|
new LinkedBlockingQueue<Runnable>(2000),
|
new ThreadFactory() {
|
@Override
|
public Thread newThread(Runnable r) {
|
return new Thread(r, "xxl-rpc, EmbedServer bizThreadPool-" + r.hashCode());
|
}
|
},
|
new RejectedExecutionHandler() {
|
@Override
|
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
|
throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!");
|
}
|
});
|
|
|
try {
|
// start server
|
ServerBootstrap bootstrap = new ServerBootstrap();
|
bootstrap.group(bossGroup, workerGroup)
|
.channel(NioServerSocketChannel.class)
|
.childHandler(new ChannelInitializer<SocketChannel>() {
|
@Override
|
public void initChannel(SocketChannel channel) throws Exception {
|
channel.pipeline()
|
.addLast(new IdleStateHandler(0, 0, 30 * 3, TimeUnit.SECONDS)) // beat 3N, close if idle
|
.addLast(new HttpServerCodec())
|
.addLast(new HttpObjectAggregator(5 * 1024 * 1024)) // merge request & reponse to FULL
|
.addLast(new EmbedHttpServerHandler(executorBiz, accessToken, bizThreadPool));
|
}
|
})
|
.childOption(ChannelOption.SO_KEEPALIVE, true);
|
|
// bind
|
ChannelFuture future = bootstrap.bind(port).sync();
|
|
logger.info(">>>>>>>>>>> xxl-job remoting server start success, nettype = {}, port = {}", EmbedServer.class, port);
|
|
// start registry
|
startRegistry(appname, address);
|
|
// wait util stop
|
future.channel().closeFuture().sync();
|
|
} catch (InterruptedException e) {
|
if (e instanceof InterruptedException) {
|
logger.info(">>>>>>>>>>> xxl-job remoting server stop.");
|
} else {
|
logger.error(">>>>>>>>>>> xxl-job remoting server error.", e);
|
}
|
} finally {
|
// stop
|
try {
|
workerGroup.shutdownGracefully();
|
bossGroup.shutdownGracefully();
|
} catch (Exception e) {
|
logger.error(e.getMessage(), e);
|
}
|
}
|
|
}
|
|
});
|
thread.setDaemon(true); // daemon, service jvm, user thread leave >>> daemon leave >>> jvm leave
|
thread.start();
|
}
|
|
public void stop() throws Exception {
|
// destroy server thread
|
if (thread!=null && thread.isAlive()) {
|
thread.interrupt();
|
}
|
|
// stop registry
|
stopRegistry();
|
logger.info(">>>>>>>>>>> xxl-job remoting server destroy success.");
|
}
|
|
|
// ---------------------- registry ----------------------
|
|
/**
|
* netty_http
|
*
|
* Copy from : https://github.com/xuxueli/xxl-rpc
|
*
|
* @author xuxueli 2015-11-24 22:25:15
|
*/
|
public static class EmbedHttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
|
private static final Logger logger = LoggerFactory.getLogger(EmbedHttpServerHandler.class);
|
|
private ExecutorBiz executorBiz;
|
private String accessToken;
|
private ThreadPoolExecutor bizThreadPool;
|
public EmbedHttpServerHandler(ExecutorBiz executorBiz, String accessToken, ThreadPoolExecutor bizThreadPool) {
|
this.executorBiz = executorBiz;
|
this.accessToken = accessToken;
|
this.bizThreadPool = bizThreadPool;
|
}
|
|
@Override
|
protected void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
|
|
// request parse
|
//final byte[] requestBytes = ByteBufUtil.getBytes(msg.content()); // byteBuf.toString(io.netty.util.CharsetUtil.UTF_8);
|
String requestData = msg.content().toString(CharsetUtil.UTF_8);
|
String uri = msg.uri();
|
HttpMethod httpMethod = msg.method();
|
boolean keepAlive = HttpUtil.isKeepAlive(msg);
|
String accessTokenReq = msg.headers().get(XxlJobRemotingUtil.XXL_JOB_ACCESS_TOKEN);
|
|
// invoke
|
bizThreadPool.execute(new Runnable() {
|
@Override
|
public void run() {
|
// do invoke
|
Object responseObj = process(httpMethod, uri, requestData, accessTokenReq);
|
|
// to json
|
String responseJson = GsonTool.toJson(responseObj);
|
|
// write response
|
writeResponse(ctx, keepAlive, responseJson);
|
}
|
});
|
}
|
|
private Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq) {
|
|
// valid
|
if (HttpMethod.POST != httpMethod) {
|
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, HttpMethod not support.");
|
}
|
if (uri==null || uri.trim().length()==0) {
|
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping empty.");
|
}
|
if (accessToken!=null
|
&& accessToken.trim().length()>0
|
&& !accessToken.equals(accessTokenReq)) {
|
return new ReturnT<String>(ReturnT.FAIL_CODE, "The access token is wrong.");
|
}
|
|
// services mapping
|
try {
|
if ("/beat".equals(uri)) {
|
return executorBiz.beat();
|
} else if ("/idleBeat".equals(uri)) {
|
IdleBeatParam idleBeatParam = GsonTool.fromJson(requestData, IdleBeatParam.class);
|
return executorBiz.idleBeat(idleBeatParam);
|
} else if ("/run".equals(uri)) {
|
TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class);
|
return executorBiz.run(triggerParam);
|
} else if ("/kill".equals(uri)) {
|
KillParam killParam = GsonTool.fromJson(requestData, KillParam.class);
|
return executorBiz.kill(killParam);
|
} else if ("/log".equals(uri)) {
|
LogParam logParam = GsonTool.fromJson(requestData, LogParam.class);
|
return executorBiz.log(logParam);
|
} else {
|
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping("+ uri +") not found.");
|
}
|
} catch (Exception e) {
|
logger.error(e.getMessage(), e);
|
return new ReturnT<String>(ReturnT.FAIL_CODE, "request error:" + ThrowableUtil.toString(e));
|
}
|
}
|
|
/**
|
* write response
|
*/
|
private void writeResponse(ChannelHandlerContext ctx, boolean keepAlive, String responseJson) {
|
// write response
|
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.copiedBuffer(responseJson, CharsetUtil.UTF_8)); // Unpooled.wrappedBuffer(responseJson)
|
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html;charset=UTF-8"); // HttpHeaderValues.TEXT_PLAIN.toString()
|
response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
|
if (keepAlive) {
|
response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
|
}
|
ctx.writeAndFlush(response);
|
}
|
|
@Override
|
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
|
ctx.flush();
|
}
|
|
@Override
|
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
|
logger.error(">>>>>>>>>>> xxl-job provider netty_http server caught exception", cause);
|
ctx.close();
|
}
|
|
@Override
|
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
|
if (evt instanceof IdleStateEvent) {
|
ctx.channel().close(); // beat 3N, close if idle
|
logger.debug(">>>>>>>>>>> xxl-job provider netty_http server close an idle channel.");
|
} else {
|
super.userEventTriggered(ctx, evt);
|
}
|
}
|
}
|
|
// ---------------------- registry ----------------------
|
|
public void startRegistry(final String appname, final String address) {
|
// start registry
|
ExecutorRegistryThread.getInstance().start(appname, address);
|
}
|
|
public void stopRegistry() {
|
// stop registry
|
ExecutorRegistryThread.getInstance().toStop();
|
}
|
|
|
}
|
