Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
B
byit-myth-job
Overview
Overview
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
liyuan
byit-myth-job
Commits
bf814404
Commit
bf814404
authored
May 28, 2020
by
huangfusuper
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
规范日志输出
parent
77a9cae7
Hide whitespace changes
Inline
Side-by-side
Showing
11 changed files
with
54 additions
and
58 deletions
+54
-58
MakeUpFlowThreadRunHelper.java
...ava/com/byit/thread/helper/MakeUpFlowThreadRunHelper.java
+4
-4
MythJobProcess.java
...server/src/main/java/com/byit/process/MythJobProcess.java
+7
-7
ApiController.java
...ava/com/byit/registry/admin/controller/ApiController.java
+4
-2
LoadBalance.java
...java/com/byit/rpc/remoting/invoker/route/LoadBalance.java
+0
-9
DefaultResultCallback.java
...va/com/byit/param/defaultparam/DefaultResultCallback.java
+0
-18
NettyClientHandler.java
...main/java/com/byit/client/handler/NettyClientHandler.java
+3
-1
NettyPluginConnectionClient.java
.../com/byit/client/handler/NettyPluginConnectionClient.java
+5
-4
RpcSpringPluginServerFactory.java
...n/java/com/byit/factory/RpcSpringPluginServerFactory.java
+5
-8
MainNettyPluginServer.java
...ain/java/com/byit/server/netty/MainNettyPluginServer.java
+4
-2
NettyPluginServerHandler.java
...m/byit/server/netty/handler/NettyPluginServerHandler.java
+5
-3
pom.xml
byit-plugin-core/pom.xml
+17
-0
No files found.
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/MakeUpFlowThreadRunHelper.java
View file @
bf814404
...
...
@@ -59,17 +59,17 @@ public class MakeUpFlowThreadRunHelper extends BaseThreadRunHelper {
Map
<
Integer
,
WaitingRecord
>
nextRunFlow
=
allByTriggerTime
.
stream
()
.
collect
(
Collectors
.
toMap
(
WaitingRecord:
:
getFlowId
,
Function
.
identity
(),
BinaryOperator
.
minBy
(
Comparator
.
comparingInt
(
WaitingRecord:
:
getWaitOrder
))));
log
.
info
(
"-------筛选后的数据为:{}----------"
,
JSON
.
toJSONString
(
nextRunFlow
));
log
.
debug
(
"-------筛选后的数据为:{}----------"
,
JSON
.
toJSONString
(
nextRunFlow
));
for
(
WaitingRecord
value
:
nextRunFlow
.
values
())
{
boolean
runRecordingIsRunning
=
runRecordingService
.
findRunRecordingIsRunning
(
value
.
getFlowId
());
log
.
info
(
"-------{}的运行状态为{}"
,
value
,
runRecordingIsRunning
);
log
.
info
(
"==================================================的运行状态为{}======================================"
,
runRecordingIsRunning
);
log
.
debug
(
"-------{}的运行状态为{}"
,
value
,
runRecordingIsRunning
);
log
.
debug
(
"==================================================的运行状态为{}======================================"
,
runRecordingIsRunning
);
if
(!
runRecordingIsRunning
){
runRecordingAndJobTaskService
.
updateRunRecordingAndTask
(
value
);
}
}
log
.
info
(
"============================com.byit.thread.helper.MakeUpFlowThreadRunHelper.start 执行完成====================================="
);
log
.
debug
(
"============================com.byit.thread.helper.MakeUpFlowThreadRunHelper.start 执行完成====================================="
);
return
UNIVERSAL_WAIT_TIME
;
}
...
...
byit-myth-executor/myth-executor-server/src/main/java/com/byit/process/MythJobProcess.java
View file @
bf814404
...
...
@@ -222,16 +222,16 @@ public class MythJobProcess implements Callable<String>{
try
{
if
(
this
.
isExecuteAsUser
)
{
final
String
cmd
=
String
.
format
(
"%s %s %s -9 %d"
,
this
.
executeAsUserBinary
,
this
.
effectiveUser
,
KILL_COMMAND
,
this
.
processId
);
System
.
out
.
println
(
"执行命令"
+
cmd
);
log
.
debug
(
"执行命令:{}"
,
cmd
);
Runtime
.
getRuntime
().
exec
(
cmd
);
}
else
{
final
String
cmd
=
String
.
format
(
"%s -9 %d"
,
KILL_COMMAND
,
this
.
processId
);
System
.
out
.
println
(
"执行命令"
+
cmd
);
log
.
debug
(
"执行命令:{}"
,
cmd
);
Runtime
.
getRuntime
().
exec
(
cmd
);
}
return
this
.
completeLatch
.
await
(
time
,
unit
);
}
catch
(
final
IOException
e
)
{
log
.
error
(
"尝试杀死失败.
"
,
e
);
log
.
error
(
"尝试杀死失败.
{}"
,
e
.
getMessage
()
);
}
}
//如果是杀死将退出值改为-100
...
...
@@ -246,7 +246,7 @@ public class MythJobProcess implements Callable<String>{
* 强制杀死这个过程
*/
public
void
hardKill
()
{
log
.
info
(
"强制杀死这个过程"
);
log
.
debug
(
"强制杀死这个过程"
);
//判断是否已经开始运行
checkStarted
();
//判断是否正在执行
...
...
@@ -255,15 +255,15 @@ public class MythJobProcess implements Callable<String>{
try
{
if
(
this
.
isExecuteAsUser
)
{
final
String
cmd
=
String
.
format
(
"%s %s %s -9 %d"
,
this
.
executeAsUserBinary
,
this
.
effectiveUser
,
KILL_COMMAND
,
this
.
processId
);
System
.
out
.
println
(
"执行命令"
+
cmd
);
log
.
debug
(
"执行命令:{}"
,
cmd
);
Runtime
.
getRuntime
().
exec
(
cmd
);
}
else
{
final
String
cmd
=
String
.
format
(
"%s -9 %d"
,
KILL_COMMAND
,
this
.
processId
);
System
.
out
.
println
(
"执行命令"
+
cmd
);
log
.
debug
(
"执行命令:{}"
,
cmd
);
Runtime
.
getRuntime
().
exec
(
cmd
);
}
}
catch
(
final
IOException
e
)
{
log
.
error
(
"Kill attempt failed.
"
,
e
);
log
.
error
(
"Kill attempt failed.
:{}"
,
e
.
getMessage
()
);
}
}
this
.
process
.
destroy
();
...
...
byit-myth-register/myth-register-server/src/main/java/com/byit/registry/admin/controller/ApiController.java
View file @
bf814404
...
...
@@ -7,6 +7,7 @@ import com.byit.registry.admin.core.util.JacksonUtil;
import
com.byit.registry.admin.service.IRegistryService
;
import
com.byit.registry.client.model.RegistryDataParamVO
;
import
com.byit.registry.client.model.RegistryParamVO
;
import
lombok.extern.slf4j.Slf4j
;
import
org.springframework.stereotype.Controller
;
import
org.springframework.web.bind.annotation.RequestBody
;
import
org.springframework.web.bind.annotation.RequestMapping
;
...
...
@@ -20,6 +21,7 @@ import java.util.Map;
@Controller
@RequestMapping
(
"/api"
)
@Slf4j
public
class
ApiController
{
@Resource
...
...
@@ -65,7 +67,7 @@ public class ApiController {
try
{
registryParamVO
=
JacksonUtil
.
readValue
(
data
,
RegistryParamVO
.
class
);
}
catch
(
Exception
e
)
{
System
.
out
.
print
(
e
);
log
.
error
(
e
.
getMessage
()
);
}
// parse param
...
...
@@ -197,7 +199,7 @@ public class ApiController {
//注册数据就是从这里加载的
registryParamVO
=
JacksonUtil
.
readValue
(
data
,
RegistryParamVO
.
class
);
}
catch
(
Exception
e
)
{
System
.
out
.
println
(
e
);
log
.
error
(
"{}"
,
e
.
getMessage
()
);
}
// parse param
...
...
byit-myth-rpc/src/main/java/com/byit/rpc/remoting/invoker/route/LoadBalance.java
View file @
bf814404
...
...
@@ -40,12 +40,4 @@ public enum LoadBalance {
}
return
defaultRouter
;
}
}
class
test
{
public
static
void
main
(
String
[]
args
)
{
for
(
LoadBalance
value
:
LoadBalance
.
values
(
))
{
System
.
out
.
println
(
value
);
}
}
}
\ No newline at end of file
byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/param/defaultparam/DefaultResultCallback.java
deleted
100644 → 0
View file @
77a9cae7
package
com
.
byit
.
param
.
defaultparam
;
import
com.alibaba.fastjson.JSON
;
import
com.byit.packet.response.PluginRpcResponsePacket
;
import
com.byit.param.ResultCallback
;
import
io.netty.channel.ChannelHandlerContext
;
/**
* 默认的回调方式
* @author huangfu
*/
public
class
DefaultResultCallback
implements
ResultCallback
{
@Override
public
void
resultCallback
(
ChannelHandlerContext
ctx
,
PluginRpcResponsePacket
pluginRpcResponsePacket
)
{
System
.
out
.
println
(
pluginRpcResponsePacket
);
System
.
out
.
println
(
JSON
.
toJSONString
(
pluginRpcResponsePacket
));
}
}
byit-plugin-core/myth-plugin-rpc-client/src/main/java/com/byit/client/handler/NettyClientHandler.java
View file @
bf814404
...
...
@@ -7,11 +7,13 @@ import com.byit.param.PluginBeat;
import
io.netty.channel.ChannelHandlerContext
;
import
io.netty.channel.SimpleChannelInboundHandler
;
import
io.netty.handler.timeout.IdleStateEvent
;
import
lombok.extern.slf4j.Slf4j
;
/**
* Netty客户端业务处理类
* @author huangfu
*/
@Slf4j
public
class
NettyClientHandler
extends
SimpleChannelInboundHandler
<
PluginRpcResponsePacket
>
{
private
PluginClientInitialization
pluginClientInitialization
;
private
PluginConnectClient
pluginConnectClient
;
...
...
@@ -32,7 +34,7 @@ public class NettyClientHandler extends SimpleChannelInboundHandler<PluginRpcRes
//判断事件是否是心跳事件
if
(
evt
instanceof
IdleStateEvent
){
pluginConnectClient
.
send
(
PluginBeat
.
PLUGIN_RPC_REQUEST_PACKET
);
System
.
out
.
println
(
"------客户端发送心跳请求-----"
);
log
.
debug
(
"------客户端发送心跳请求-----"
);
}
else
{
super
.
userEventTriggered
(
ctx
,
evt
);
}
...
...
byit-plugin-core/myth-plugin-rpc-client/src/main/java/com/byit/client/handler/NettyPluginConnectionClient.java
View file @
bf814404
...
...
@@ -14,6 +14,7 @@ import io.netty.channel.nio.NioEventLoopGroup;
import
io.netty.channel.socket.SocketChannel
;
import
io.netty.channel.socket.nio.NioSocketChannel
;
import
io.netty.handler.timeout.IdleStateHandler
;
import
lombok.extern.slf4j.Slf4j
;
import
java.util.concurrent.TimeUnit
;
...
...
@@ -21,6 +22,7 @@ import java.util.concurrent.TimeUnit;
* netty客户端
* @author huangfu
*/
@Slf4j
public
class
NettyPluginConnectionClient
extends
PluginConnectClient
{
private
EventLoopGroup
group
;
private
Channel
channel
;
...
...
@@ -55,9 +57,8 @@ public class NettyPluginConnectionClient extends PluginConnectClient {
this
.
channel
=
bootstrap
.
connect
(
ip
,
port
).
sync
().
channel
();
// valid
if
(!
isValidate
())
{
System
.
out
.
println
(
"------关闭链接
"
);
log
.
debug
(
"------关闭链接--------
"
);
close
();
return
;
}
}
...
...
@@ -65,7 +66,7 @@ public class NettyPluginConnectionClient extends PluginConnectClient {
@Override
public
void
close
()
{
if
(
this
.
channel
!=
null
&&
isValidate
()){
System
.
out
.
println
(
"------关闭链接"
+
channel
.
id
().
asShortText
());
log
.
warn
(
"------关闭链接{}------"
,
channel
.
id
().
asShortText
());
this
.
channel
.
close
();
}
...
...
@@ -83,7 +84,7 @@ public class NettyPluginConnectionClient extends PluginConnectClient {
}
@Override
public
void
send
(
PluginRpcRequestPacket
pluginRpcRequestPacket
)
throws
Exception
{
public
void
send
(
PluginRpcRequestPacket
pluginRpcRequestPacket
)
{
this
.
channel
.
writeAndFlush
(
pluginRpcRequestPacket
);
}
}
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/factory/RpcSpringPluginServerFactory.java
View file @
bf814404
package
com
.
byit
.
factory
;
import
com.byit.callback.RemainingOperationsCallBack
;
import
com.byit.registry.PluginServiceRegistry
;
import
com.byit.server.PluginServer
;
import
com.byit.task.annotations.TaskHandler
;
import
com.byit.task.handler.interfaces.IJobHandler
;
import
lombok.extern.slf4j.Slf4j
;
import
org.apache.commons.lang3.StringUtils
;
import
org.springframework.aop.framework.AdvisedSupport
;
import
org.springframework.aop.framework.AopProxy
;
...
...
@@ -12,7 +10,6 @@ import org.springframework.aop.support.AopUtils;
import
org.springframework.beans.BeansException
;
import
org.springframework.beans.factory.DisposableBean
;
import
org.springframework.beans.factory.InitializingBean
;
import
org.springframework.beans.factory.annotation.Value
;
import
org.springframework.context.ApplicationContext
;
import
org.springframework.context.ApplicationContextAware
;
...
...
@@ -24,6 +21,7 @@ import java.util.concurrent.ConcurrentHashMap;
* spring实现
* @author huangfu
*/
@Slf4j
public
class
RpcSpringPluginServerFactory
extends
PluginServerFactory
implements
ApplicationContextAware
,
InitializingBean
,
DisposableBean
{
private
String
address
;
private
String
biz
;
...
...
@@ -64,7 +62,6 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
}
if
(
target
!=
null
){
TaskHandler
annotation
=
target
.
getClass
().
getAnnotation
(
TaskHandler
.
class
);
System
.
out
.
println
(
"注解:"
+
annotation
);
String
taskName
=
annotation
.
taskName
();
super
.
addService
(
taskName
,
value
);
String
expand
=
annotation
.
expand
();
...
...
@@ -73,7 +70,7 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
}
}
}
else
{
System
.
err
.
println
(
"警告!bean"
+
key
+
"不是【com.byit.task.handler.interfaces.IJobHandler】类型!忽略该bean!"
);
log
.
warn
(
"警告!bean {} 不是【com.byit.task.handler.interfaces.IJobHandler】类型!忽略该bean!"
,
key
);
}
});
}
...
...
@@ -93,7 +90,7 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
}
private
static
Object
getCglibProxyTargetObject
(
Object
proxy
)
throws
Exception
{
System
.
out
.
println
(
proxy
+
"-------该对象为cglib代理对象-------"
);
log
.
warn
(
"{}:-------该对象为cglib代理对象-------"
,
proxy
);
Field
h
=
proxy
.
getClass
().
getDeclaredField
(
"CGLIB$CALLBACK_0"
);
h
.
setAccessible
(
true
);
Object
dynamicAdvisedInterceptor
=
h
.
get
(
proxy
);
...
...
@@ -108,7 +105,7 @@ public class RpcSpringPluginServerFactory extends PluginServerFactory implements
private
static
Object
getJdkDynamicProxyTargetObject
(
Object
proxy
)
throws
Exception
{
System
.
out
.
println
(
proxy
+
"-------该对象为jdk代理对象-------"
);
log
.
debug
(
"-------{},该对象为jdk代理对象-------"
,
proxy
);
Field
h
=
proxy
.
getClass
().
getSuperclass
().
getDeclaredField
(
"h"
);
h
.
setAccessible
(
true
);
AopProxy
aopProxy
=
(
AopProxy
)
h
.
get
(
proxy
);
...
...
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/server/netty/MainNettyPluginServer.java
View file @
bf814404
...
...
@@ -14,6 +14,7 @@ import io.netty.channel.nio.NioEventLoopGroup;
import
io.netty.channel.socket.SocketChannel
;
import
io.netty.channel.socket.nio.NioServerSocketChannel
;
import
io.netty.handler.timeout.IdleStateHandler
;
import
lombok.extern.slf4j.Slf4j
;
import
java.util.concurrent.ThreadPoolExecutor
;
import
java.util.concurrent.TimeUnit
;
...
...
@@ -22,6 +23,7 @@ import java.util.concurrent.TimeUnit;
* main方法启动
* @author huangfu
*/
@Slf4j
public
class
MainNettyPluginServer
extends
PluginServer
{
@Override
public
void
start
(
PluginServerFactory
pluginServerFactory
)
{
...
...
@@ -53,7 +55,7 @@ public class MainNettyPluginServer extends PluginServer {
ChannelFuture
closeFuture
=
channelFuture
.
channel
().
closeFuture
().
sync
();
closeFuture
.
addListener
(
future
->{
if
(
future
.
isSuccess
())
{
System
.
out
.
println
(
"--------
---------"
);
log
.
warn
(
"--------服务关闭
---------"
);
}
});
}
catch
(
Exception
e
){
...
...
@@ -78,7 +80,7 @@ public class MainNettyPluginServer extends PluginServer {
@Override
public
void
stop
()
{
System
.
out
.
println
(
"----
---------"
);
log
.
warn
(
"--------服务关闭
---------"
);
super
.
onStop
();
}
}
byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/server/netty/handler/NettyPluginServerHandler.java
View file @
bf814404
...
...
@@ -13,6 +13,7 @@ import com.byit.task.handler.interfaces.IJobHandler;
import
io.netty.channel.ChannelHandlerContext
;
import
io.netty.channel.SimpleChannelInboundHandler
;
import
io.netty.handler.timeout.IdleStateEvent
;
import
lombok.extern.slf4j.Slf4j
;
import
java.util.Map
;
import
java.util.concurrent.ThreadPoolExecutor
;
...
...
@@ -21,6 +22,7 @@ import java.util.concurrent.ThreadPoolExecutor;
* netty rpc服务器的业务处理
* @author huangfu
*/
@Slf4j
public
class
NettyPluginServerHandler
extends
SimpleChannelInboundHandler
<
PluginRpcRequestPacket
>
{
public
static
final
String
TOKEN_NAME
=
"token"
;
...
...
@@ -42,7 +44,7 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
@Override
protected
void
channelRead0
(
ChannelHandlerContext
ctx
,
PluginRpcRequestPacket
msg
)
{
if
(
PluginBeat
.
BEAT_ID
.
equals
(
msg
.
getRequestId
())){
System
.
out
.
println
(
"------接收到客户端的心跳连接-------"
);
log
.
info
(
"------接收到客户端的心跳连接-------"
);
return
;
}
PluginRpcResponsePacket
transferPluginRpcResponse
=
new
PluginRpcResponsePacket
();
...
...
@@ -96,7 +98,7 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
post
.
header
(
HEADER_TOKEN_NAME
,
"Token"
);
post
.
header
(
HEADER_CONTENT_TYPE
,
"application/json"
);
post
.
body
(
responseStr
).
execute
();
System
.
out
.
println
(
"-------消息回复成功-----------"
);
log
.
debug
(
"-------消息回复成功-----------"
);
}
/**
...
...
@@ -114,7 +116,7 @@ public class NettyPluginServerHandler extends SimpleChannelInboundHandler<Plugin
public
void
userEventTriggered
(
ChannelHandlerContext
ctx
,
Object
evt
)
throws
Exception
{
if
(
evt
instanceof
IdleStateEvent
){
ctx
.
channel
().
close
();
System
.
out
.
println
(
"-------心跳超时,关闭链接-------"
);
log
.
debug
(
"-------心跳超时,关闭链接-------"
);
}
else
{
super
.
userEventTriggered
(
ctx
,
evt
);
}
...
...
byit-plugin-core/pom.xml
View file @
bf814404
...
...
@@ -29,5 +29,21 @@
</snapshotRepository>
</distributionManagement>
<dependencies>
<dependency>
<groupId>
org.projectlombok
</groupId>
<artifactId>
lombok
</artifactId>
</dependency>
<dependency>
<groupId>
org.slf4j
</groupId>
<artifactId>
slf4j-api
</artifactId>
</dependency>
<dependency>
<groupId>
org.slf4j
</groupId>
<artifactId>
slf4j-log4j12
</artifactId>
</dependency>
</dependencies>
</project>
\ No newline at end of file
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment