mirror of https://github.com/longtai-cn/hippo4j
小马哥
2 years ago
committed by
GitHub
commit
ce53d6b36b
@ -0,0 +1,32 @@
|
||||
package cn.hippo4j.config.config;
|
||||
|
||||
import cn.hippo4j.config.netty.MonitorNettyServer;
|
||||
import cn.hippo4j.config.service.biz.HisRunDataService;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
@Configuration
|
||||
public class NettyServerConfig {
|
||||
|
||||
@Bean
|
||||
public EventLoopGroup bossGroup(){
|
||||
return new NioEventLoopGroup();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public EventLoopGroup workGroup(){
|
||||
return new NioEventLoopGroup();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public MonitorNettyServer monitorNettyServer(ServerBootstrapProperties serverBootstrapProperties,
|
||||
HisRunDataService hisRunDataService,
|
||||
EventLoopGroup bossGroup,
|
||||
EventLoopGroup workGroup){
|
||||
return new MonitorNettyServer(serverBootstrapProperties,hisRunDataService,bossGroup,workGroup);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -0,0 +1,30 @@
|
||||
package cn.hippo4j.config.netty;
|
||||
|
||||
import cn.hippo4j.common.monitor.Message;
|
||||
import cn.hippo4j.common.monitor.MessageWrapper;
|
||||
import cn.hippo4j.common.toolkit.MessageConvert;
|
||||
import cn.hippo4j.config.monitor.QueryMonitorExecuteChoose;
|
||||
import cn.hippo4j.config.service.biz.HisRunDataService;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||
|
||||
/**
|
||||
* ServerHandler
|
||||
*
|
||||
* @author lk
|
||||
* @date 2022/06/18
|
||||
*/
|
||||
@Slf4j
|
||||
@AllArgsConstructor
|
||||
public class ServerHandler extends SimpleChannelInboundHandler<MessageWrapper> {
|
||||
|
||||
private HisRunDataService hisRunDataService;
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, MessageWrapper msg) throws Exception {
|
||||
hisRunDataService.dataCollect(msg);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,28 @@
|
||||
package cn.hippo4j.springboot.starter.config;
|
||||
|
||||
import cn.hippo4j.springboot.starter.monitor.send.netty.NettyConnectSender;
|
||||
import cn.hippo4j.springboot.starter.monitor.send.MessageSender;
|
||||
import cn.hippo4j.springboot.starter.remote.ServerNettyAgent;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
|
||||
/**
|
||||
* Netty ClientCon figuration
|
||||
*
|
||||
* @author lk
|
||||
* @date 2022/6/18
|
||||
*/
|
||||
@ConditionalOnProperty(prefix = BootstrapProperties.PREFIX, name = "report-type", matchIfMissing = false, havingValue = "netty")
|
||||
public class NettyClientConfiguration {
|
||||
|
||||
@Bean
|
||||
@SuppressWarnings("all")
|
||||
public ServerNettyAgent serverNettyAgent(BootstrapProperties properties) {
|
||||
return new ServerNettyAgent(properties);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public MessageSender messageSender(ServerNettyAgent serverNettyAgent){
|
||||
return new NettyConnectSender(serverNettyAgent);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,60 @@
|
||||
package cn.hippo4j.springboot.starter.monitor.send.netty;
|
||||
|
||||
import cn.hippo4j.common.monitor.Message;
|
||||
import cn.hippo4j.common.monitor.MessageWrapper;
|
||||
import cn.hippo4j.common.toolkit.MessageConvert;
|
||||
import cn.hippo4j.springboot.starter.monitor.send.MessageSender;
|
||||
import cn.hippo4j.springboot.starter.remote.ServerNettyAgent;
|
||||
import io.netty.bootstrap.Bootstrap;
|
||||
import io.netty.channel.ChannelInitializer;
|
||||
import io.netty.channel.ChannelPipeline;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.socket.SocketChannel;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.handler.codec.serialization.ClassResolvers;
|
||||
import io.netty.handler.codec.serialization.ObjectDecoder;
|
||||
import io.netty.handler.codec.serialization.ObjectEncoder;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* Netty ConnectSender
|
||||
*
|
||||
* @author lk
|
||||
* @date 2022/06/18
|
||||
*/
|
||||
@Slf4j
|
||||
@AllArgsConstructor
|
||||
@Component
|
||||
public class NettyConnectSender implements MessageSender {
|
||||
|
||||
private ServerNettyAgent serverNettyAgent;
|
||||
|
||||
@Override
|
||||
public void send(Message message) {
|
||||
MessageWrapper messageWrapper = MessageConvert.convert(message);
|
||||
EventLoopGroup eventLoopGroup = serverNettyAgent.getEventLoopGroup();
|
||||
try {
|
||||
Bootstrap bootstrap = new Bootstrap();
|
||||
bootstrap.group(eventLoopGroup)
|
||||
.channel(NioSocketChannel.class)
|
||||
.handler(new ChannelInitializer<SocketChannel>(){
|
||||
@Override
|
||||
protected void initChannel(SocketChannel ch) throws Exception {
|
||||
ChannelPipeline pipeline = ch.pipeline();
|
||||
pipeline.addLast(new ObjectEncoder());
|
||||
pipeline.addLast(new ObjectDecoder(Integer.MAX_VALUE,
|
||||
ClassResolvers.cacheDisabled(null)));
|
||||
pipeline.addLast(new SenderHandler(messageWrapper));
|
||||
}
|
||||
});
|
||||
|
||||
bootstrap.connect(serverNettyAgent.getNettyServerAddress(), serverNettyAgent.getNettyServerPort()).sync();
|
||||
} catch (Exception e) {
|
||||
log.error("netty send error ",e);
|
||||
} /*finally {
|
||||
eventLoopGroup.shutdownGracefully();
|
||||
}*/
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,30 @@
|
||||
package cn.hippo4j.springboot.starter.monitor.send.netty;
|
||||
|
||||
import cn.hippo4j.common.monitor.MessageWrapper;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.SimpleChannelInboundHandler;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
/**
|
||||
* SenderHandler
|
||||
*
|
||||
* @author lk
|
||||
* @date 2022/06/18
|
||||
*/
|
||||
@Slf4j
|
||||
@AllArgsConstructor
|
||||
public class SenderHandler extends SimpleChannelInboundHandler<MessageWrapper> {
|
||||
|
||||
private MessageWrapper messageWrapper;
|
||||
|
||||
@Override
|
||||
protected void channelRead0(ChannelHandlerContext ctx, MessageWrapper msg) throws Exception {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelActive(ChannelHandlerContext ctx) throws Exception {
|
||||
ctx.channel().writeAndFlush(messageWrapper);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,38 @@
|
||||
package cn.hippo4j.springboot.starter.remote;
|
||||
|
||||
import cn.hippo4j.springboot.starter.config.BootstrapProperties;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
|
||||
/**
|
||||
* Server Netty Agent
|
||||
*
|
||||
* @author lk
|
||||
* @date 2022/6/18
|
||||
*/
|
||||
public class ServerNettyAgent {
|
||||
|
||||
private final BootstrapProperties dynamicThreadPoolProperties;
|
||||
|
||||
private final ServerListManager serverListManager;
|
||||
|
||||
private EventLoopGroup eventLoopGroup;
|
||||
|
||||
public ServerNettyAgent(BootstrapProperties properties){
|
||||
this.dynamicThreadPoolProperties = properties;
|
||||
this.serverListManager = new ServerListManager(dynamicThreadPoolProperties);
|
||||
this.eventLoopGroup = new NioEventLoopGroup();
|
||||
}
|
||||
|
||||
public EventLoopGroup getEventLoopGroup(){
|
||||
return eventLoopGroup;
|
||||
}
|
||||
|
||||
public String getNettyServerAddress() {
|
||||
return serverListManager.getCurrentServerAddr().split(":")[1].replace("//","");
|
||||
}
|
||||
|
||||
public Integer getNettyServerPort(){
|
||||
return Integer.parseInt(serverListManager.getNettyServerPort());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue