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

#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的容器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

#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

#!/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

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);

}

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);
    }

}

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;

}

clientip = emqxrmiservice.getclientbyid("clientid").getipaddress();