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 ]; thenecho "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;@Overridepublic 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端口暴露需要前期規劃好，要不然特別麻煩

寫到最后