确保您的 MQTT 客户端、MQTT 服务(Netty MQTT 服务)和 Nginx 都支持代理协议。不是所有的软件都支持这种协议。
以下是Ngin与MQTT服务端修改，客户端针对每个平台自行搜索修改
Nginx配置修改
stream {upstream mqtt_backend {server mqtt_server_ip:mqtt_server_port;}server {listen mqtt_proxy_port proxy_protocol;proxy_pass mqtt_backend;proxy_protocol on;}
}
示例
stream {upstream mqtt_backend {server 192.168.1.10:1883; # MQTT 后端服务器的 IP 地址和端口}server {listen 8888 proxy_protocol; # 使用代理协议监听的端口，客户端将连接到此端口proxy_pass mqtt_backend;proxy_protocol on;}
}
在这个示例中：
192.168.1.10:1883是您的 MQTT 服务(Netty MQTT 服务)的 IP 地址和端口。8888是 Nginx 代理将监听的端口，客户端将连接到此端口。这个端口可以是任何可用的端口号，但请确保它与您的实际需求相符。proxy_protocol是一个指令，用于告诉 Nginx 使用代理协议。在listen 8888 proxy_protocol;中，proxy_protocol指示 Nginx 使用代理协议处理传入的连接。
客户端将连接到 Nginx 代理的 8888 端口，Nginx 会将连接转发到 MQTT 后端服务器(IP 地址为 192.168.1.10，端口为 1883)。通过使用代理协议，Nginx 将把客户端的连接信息(如原始 IP 地址和端口)传递给 MQTT 服务。
Netty服务端
- 首先，将以下 HAProxy 代理协议库添加到您的项目中：
Maven 依赖：
<dependency><groupId>io.netty</groupId><artifactId>netty-codec-haproxy</artifactId><version>4.1.72.Final</version>
</dependency>
- 在
ChannelInitializer类中将HAProxyMessageDecoder添加到管道中：
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.haproxy.HAProxyMessageDecoder;public class MyMqttServerInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();// 将 HAProxyMessageDecoder 添加到管道中pipeline.addLast(new HAProxyMessageDecoder());// ... 添加其他处理器，例如 MQTT 编解码器和业务处理器 ...}
}
- 在您的自定义
ChannelInboundHandler中，获取代理协议头中的客户端真实 IP 和端口：
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.haproxy.HAProxyMessage;public class ProxyProtocolHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if (msg instanceof HAProxyMessage) {HAProxyMessage haProxyMessage = (HAProxyMessage) msg;String clientIp = haProxyMessage.sourceAddress();int clientPort = haProxyMessage.sourcePort();// 在这里，您可以将客户端的真实 IP 和端口传递给后续处理器// 例如，将它们设置为 ChannelHandlerContext 的属性ctx.channel().attr(AttributeKey.valueOf("clientIp")).set(clientIp);ctx.channel().attr(AttributeKey.valueOf("clientPort")).set(clientPort);// 继续处理其他消息ctx.fireChannelRead(msg);} else {super.channelRead(ctx, msg);}}
}