利用Java获取被nginx代理的emqx客户端真实ip

2025-08-22 08:30:49 作者：BothSavage

使用nginx作为负载均衡（Load Balancing）的时候,发现真实ip无法获取,所以本文小编就来和大家介绍一下如何使用Java获取被nginx代理的emqx客户端真实ip地址吧

契机

使用nginx作为负载均衡（Load Balancing）的时候，发现真实ip无法获取。几经折腾终于拿到真实ip，又发现被代理的端口又无法使用非代理模式连接，由于之前暴露的docker端口有限，从中部分取巧终于拨开云雾见光明，再结合Java客户端获取真实ip一气呵成。

EMQX配置

#docker部署
#18083为管理页面端口，1883为默认mqtt端口，8883为默认mqtts端口
#如果还没有新建emqx容器，建议多暴露端口，具体原因见下文
docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:5.7.2

进入emqx管理页面，新建proxy_tcp监听器，类型为tcp，端口为8883
然后关闭之前ssl-8883监听
借用端口是因为docker暴露到宿主机的端口不够了，一旦设置比如proxy_tcp监听器的8883代理监听后，客户端就只能通过nginx代理连接这个8883端口了，直接连接8883就不行了

#获取emqx的容器id，假如为1111
docker ps | grep emqx 

#将配置文件拷贝出来
docker cp 1111:/opt/emqx/etc/emqx.conf .

#修改，添加下面项目
vim emqx.conf 

#打开proxy_tcp的代理监听
listeners.tcp.proxy_tcp {
  proxy_protocol = true
}

#关闭default的代理，默认是关闭的
#但是一旦打开过，就要手动设置为false
listeners.tcp.default {
  proxy_protocol = false
}

#放置回去
docker cp ./emqx.conf 1111:/opt/emqx/etc/

#重启容器
docker restart 1111

此时

直接使用客户端设备连接8883端口失败，无论mqtt/mqtts协议
1883端口可以使用mqtt协议可以正常连接
emqx不存放证书，ssl认证是一个耗时的操作，丢到nginx去搞

nginx配置

#nginx安装 - 略

#nginx安装模块
sudo yum install nginx-mod-stream

#修改nginx配置文件
vim /etc/nginx/nginx.conf

#配置文件如下
#192.168.0.1为emqx所在服务器，与nginx不在一个服务器
#监听1883为mqtt
#监听8883为mqtts
stream {
    upstream stream_backend {
      zone tcp_servers 64k;
      hash $remote_addr;
      server 192.168.0.1:1883 max_fails=2 fail_timeout=30s;
    }

    server {
        listen 1883;
        proxy_pass stream_backend;
        proxy_buffer_size 4k;
    }

    upstream stream_backend_ssl {
      zone tcp_servers 64k;
      hash $remote_addr;
      server 192.168.0.1:8883 max_fails=2 fail_timeout=30s;
    }

    server {
        listen 8883 ssl;
        proxy_protocol on;        
        proxy_pass stream_backend_ssl;
        proxy_buffer_size 4k;
        ssl_handshake_timeout 15s;
        ssl_certificate     /etc/nginx/cert/_.xx.pem;
        ssl_certificate_key /etc/nginx/cert/_.xx.key;
    }

}

#重启nginx
sudo systemctl restart nginx

此时

8883为mqtts端口，链接mqtts，emqx控制台可以看到真实ip
1883为mqtt端口，链接mqtt，emqx控制台看不到真实ip
理论上业务只暴露mqtts端口到外网，mqtt为内部调试方便抓包用的

脚本获取

设置api密钥，保存下username和pasword

#!/bin/bash

# Check if both host and password arguments are provided
if [ $# -lt 2 ]; then
  echo "Usage: $0 <host> <password>"
  exit 1
fi

# Define host, username, and password
HOST="$1"
USERNAME="username"
PASSWORD="$2"

# Encode username and password in Base64
AUTH=$(echo -n "$USERNAME:$PASSWORD" | base64)

# Create the curl command
curl -H "Authorization: Basic $AUTH" "http://$HOST:18083/api/v5/clients?like_username=server"

#运行
chmod +x ./emqx_curl.sh
./emqx_curl.sh localhost your_password_here

Java获取

service

package com.bothsavage.common.mqtt.service;

import com.alibaba.fastjson.JSONObject;
import com.bothsavage.common.mqtt.config.interceptor.EmqxAuthInterceptor;
import com.bothsavage.common.mqtt.entity.dto.ClientInfoDTO;
import com.bothsavage.common.mqtt.entity.dto.EmqxRespDTO;

import feign.Headers;

import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestParam;

/**
 * Emqx请求
 *
 * @author bothSavage
 */
@FeignClient(name = "emqxRmiService", url = "${emqx.rmi.url:http://127.0.0.1:18083/}",configuration = EmqxAuthInterceptor.class)
public interface EmqxRmiService {

    /**
     * 获取客户端信息
     *
     * @param clientId 客户端id
     * @return -
     */
    @GetMapping(value = "/api/v5/clients/{id}")
    @Headers(value = {"Content-Type=application/json;charset=UTF-8", "Accept=application/json"})
    ClientInfoDTO getClientById(@PathVariable("id") String clientId);

    /**
     * 查询客户端信息
     *
     * @param likeUserName 客户端名称
     * @return -
     */
    @GetMapping(value = "/api/v5/clients")
    @Headers(value = {"Content-Type=application/json;charset=UTF-8", "Accept=application/json"})
    EmqxRespDTO<ClientInfoDTO> getClientsByUserName(@RequestParam("like_username") String likeUserName);

}

auth

package com.bothsavage.common.mqtt.config.interceptor;

import feign.RequestInterceptor;
import feign.RequestTemplate;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import java.util.Base64;

/**
 * Emqx请求权限拦截
 *
 * @author bothSavage
 */
@Configuration
public class EmqxAuthInterceptor implements RequestInterceptor {

    @Value("${emqx.rmi.username:x}")
    private String username;

    @Value("${emqx.rmi.password:x}")
    private String password;

    @Override
    public void apply(RequestTemplate template) {
        String auth = username + ":" + password;
        String encodedAuth = Base64.getEncoder().encodeToString(auth.getBytes());
        template.header("Authorization", "Basic " + encodedAuth);
    }

}

dto

package com.bothsavage.common.mqtt.entity.dto;

import com.alibaba.fastjson.annotation.JSONField;

import lombok.Data;

/**
 * dto
 *
 * @author bothSavage
 */
@Data
public class ClientInfoDTO {

    @JSONField(name = "clientid")
    private String clientid;

    @JSONField(name = "mqueue_len")
    private Integer mqueueLen;

    @JSONField(name = "reductions")
    private Integer reductions;

    @JSONField(name = "keepalive")
    private Integer keepalive;

    @JSONField(name = "listener")
    private String listener;

    @JSONField(name = "proto_ver")
    private Integer protoVer;

    @JSONField(name = "recv_msg.dropped.await_pubrel_timeout")
    private Integer recvMsgDroppedAwaitPubrelTimeout;

    @JSONField(name = "send_msg.dropped.expired")
    private Integer sendMsgDroppedExpired;

    @JSONField(name = "mountpoint")
    private Object mountpoint;

    @JSONField(name = "mailbox_len")
    private Integer mailboxLen;

    @JSONField(name = "send_msg")
    private Integer sendMsg;

    @JSONField(name = "zone")
    private String zone;

    @JSONField(name = "subscriptions_cnt")
    private Integer subscriptionsCnt;

    @JSONField(name = "heap_size")
    private Integer heapSize;

    @JSONField(name = "recv_msg")
    private Integer recvMsg;

    @JSONField(name = "recv_cnt")
    private Integer recvCnt;

    @JSONField(name = "send_msg.dropped.too_large")
    private Integer sendMsgDroppedTooLarge;

    @JSONField(name = "awaiting_rel_cnt")
    private Integer awaitingRelCnt;

    @JSONField(name = "subscriptions_max")
    private String subscriptionsMax;

    @JSONField(name = "recv_msg.qos0")
    private Integer recvMsgQos0;

    @JSONField(name = "recv_msg.qos1")
    private Integer recvMsgQos1;

    @JSONField(name = "recv_msg.qos2")
    private Integer recvMsgQos2;

    @JSONField(name = "node")
    private String node;

    @JSONField(name = "inflight_max")
    private Integer inflightMax;

    @JSONField(name = "port")
    private Integer port;

    @JSONField(name = "recv_pkt")
    private Integer recvPkt;

    @JSONField(name = "send_oct")
    private Integer sendOct;

    @JSONField(name = "inflight_cnt")
    private Integer inflightCnt;

    @JSONField(name = "awaiting_rel_max")
    private String awaitingRelMax;

    @JSONField(name = "is_persistent")
    private Boolean isPersistent;

    @JSONField(name = "send_msg.dropped")
    private Integer sendMsgDropped;

    @JSONField(name = "recv_msg.dropped")
    private Integer recvMsgDropped;

    @JSONField(name = "clean_start")
    private Boolean cleanStart;

    @JSONField(name = "send_msg.qos0")
    private Integer sendMsgQos0;

    @JSONField(name = "created_at")
    private String createdAt;

    @JSONField(name = "connected_at")
    private String connectedAt;

    @JSONField(name = "enable_authn")
    private Boolean enableAuthn;

    @JSONField(name = "mqueue_dropped")
    private Integer mqueueDropped;

    @JSONField(name = "is_bridge")
    private Boolean isBridge;

    @JSONField(name = "send_msg.dropped.queue_full")
    private Integer sendMsgDroppedQueueFull;

    @JSONField(name = "proto_name")
    private String protoName;

    @JSONField(name = "ip_address")
    private String ipAddress;

    @JSONField(name = "send_cnt")
    private Integer sendCnt;

    @JSONField(name = "connected")
    private Boolean connected;

    @JSONField(name = "recv_oct")
    private Integer recvOct;

    @JSONField(name = "mqueue_max")
    private Integer mqueueMax;

    @JSONField(name = "send_msg.qos2")
    private Integer sendMsgQos2;

    @JSONField(name = "send_msg.qos1")
    private Integer sendMsgQos1;

    @JSONField(name = "expiry_interval")
    private Integer expiryInterval;

    @JSONField(name = "send_pkt")
    private Integer sendPkt;

    @JSONField(name = "username")
    private String username;

}

test

clientIp = emqxRmiService.getClientById("clientId").getIpAddress();

总结

路线清晰，只是操作起来稍微麻烦点
注意：emqx打开了代理的访问的监听器，只能通过nginx来中转，无法直接链接了
当前只用了一个emqx，没有做集群
docker端口暴露需要前期规划好，要不然特别麻烦

到此这篇关于利用Java获取被nginx代理的emqx客户端真实ip的文章就介绍到这了,更多相关Java获取客户端ip内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！