|
Nginx搭建Netty负载均衡
参考链接:利用Nginx的stream实现Netty的TCP负载均衡
笔记记录一下用Nginx实现netty的负载均衡学习过程。
一、实现nettyserver，springboot+netty
ServerNetty示例代码
package com.example.demo1.netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
@Component
public class ServerNetty implements ApplicationRunner {
final static Logger log = LogManager.getLogger(ServerNetty.class);
@Value("${netty.port}")
private int port;
private String ip = "127.0.0.1";
public void start() throws InterruptedException {
NioEventLoopGroup boss = null;
NioEventLoopGroup worker = null;
try {
ServerBootstrap b = new ServerBootstrap();
boss = new NioEventLoopGroup();
worker = new NioEventLoopGroup();
b.group(boss, worker);
b.channel(NioServerSocketChannel.class);
b.localAddress(port);
b.option(ChannelOption.SO_KEEPALIVE, true);
b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
b.childHandler(new ChannelInitializer() {
@Override
protected void initChannel(Channel channel) throws Exception {
channel.pipeline().addLast(new ServerHandler());
}
});
log.info("启动 netty 服务端");
ChannelFuture future = b.bind(ip, port).sync();
log.info("服务器启动成功,监听端口{}", future.channel().localAddress());
future.channel().closeFuture().sync();
} catch (Exception ex) {
ex.printStackTrace();
} finally {
worker.shutdownGracefully();
boss.shutdownGracefully();
}
}
@Override
public void run(ApplicationArguments args) throws Exception {
System.out.println("执行.............");
start();
}
}
ServerHandler示例
package com.example.demo1.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.nio.charset.Charset;
public class ServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
ctx.fireChannelRegistered();
}
@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
ctx.fireChannelUnregistered();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("收到链接:"+ctx.channel().remoteAddress());
ctx.fireChannelActive();
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
ctx.fireChannelInactive();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf in = (ByteBuf) msg;
System.out.println("收到客户端"+ctx.channel().remoteAddress().toString()+"内容:"+in.toString(Charset.forName("UTF-8")));
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.fireChannelReadComplete();
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
ctx.fireUserEventTriggered(evt);
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
ctx.fireChannelWritabilityChanged();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.fireExceptionCaught(cause);
}
}
二、实现nettyclient
示例代码
package netty.demo1;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
public class NettyDiscardClient {
static final Logger log = LogManager.getLogger(NettyDiscardClient.class);
public void run(){
try{
NioEventLoopGroup boss = new NioEventLoopGroup();
Bootstrap b = new Bootstrap();
b.group(boss).channel(NioSocketChannel.class)
.handler(new ChannelInitializer<>() {
@Override
protected void initChannel(Channel channel) throws Exception {
channel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
static final Logger log = LogManager.getLogger(NettyDiscardServer.class);
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
byte[] b = (byte[]) msg;
log.info("收到服务器消息:{}",new String(b));
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.info(ctx.channel().remoteAddress().toString() + " 连入 ");
Scanner scanner = new Scanner(System.in);
while(true){
log.info("请输入测试数据:");
String content = scanner.nextLine();
ctx.writeAndFlush(Unpooled.copiedBuffer(content.getBytes(StandardCharsets.UTF_8)));
}
}
});
}
});
log.info("启动 netty 客户端");
ChannelFuture cf = b.connect("127.0.0.1",8300).sync();
cf.channel().closeFuture().sync();
}catch (Exception ex){
}
}
public static void main(String[] args) {
new NettyDiscardClient().run();
}
}
三、测试是否成功
把netty server的端口号 和client的端口号保持一致，先运行server，再运行client，在server中能收到消息就测试成功
四、编辑nginx配置文件
在conf/nginx.conf中增加下面的内容
stream {
upstream netty_server {
server 127.0.0.1:8301 weight=1;
server 127.0.0.1:8302 weight=1;
}
server {
listen 8300;
proxy_pass netty_server;
}
}
其中server 127.0.0.1:8301 server 127.0.0.1:8302 分别对应两个netty server的IP 和端口号
listen 8300 为对外开放的端口号，对应client示例代码的端口号
五、测试nginx反向代理
启动server
服务器启动成功,监听端口/127.0.0.1:8301
服务器启动成功,监听端口/[0:0:0:0:0:0:0:0]:8302
重新启动nginx
./nginx -s reload
启动多个client进程（同一份代码复制多份，分别启动），模拟多个不同的客户端
一个server打印下面的消息
收到链接:/127.0.0.1:51424
收到链接:/127.0.0.1:51516
另一个server打印下面的消息
收到链接:/127.0.0.1:49992
说明反向代理成功
|