Android端Paho MQTT v3客户端HTTP代理连接失败求助

一、代理连接超时问题解决 #

Paho MQTT Java v3客户端本身没有直接暴露代理配置接口，必须通过自定义SocketFactory实现代理连接。之前的尝试失败大概率是因为代理类型、地址配置错误，或者SSL层处理不当。以下是针对SOCKS代理和HTTP代理的完整实现：

1. SOCKS代理适配（适用于ssl://协议） #

public class SocksSSLSocketFactory extends SSLSocketFactory {
    private final SSLSocketFactory delegate;
    private final Proxy proxy;

    public SocksSSLSocketFactory(SSLSocketFactory delegate, String proxyHost, int proxyPort) {
        this.delegate = delegate;
        this.proxy = new Proxy(Proxy.Type.SOCKS, new InetSocketAddress(proxyHost, proxyPort));
    }

    @Override
    public Socket createSocket(Socket s, String host, int port, boolean autoClose) throws IOException {
        Socket proxySocket = new Socket(proxy);
        proxySocket.connect(new InetSocketAddress(host, port));
        return delegate.createSocket(proxySocket, host, port, autoClose);
    }

    @Override
    public Socket createSocket(String host, int port) throws IOException {
        Socket proxySocket = new Socket(proxy);
        proxySocket.connect(new InetSocketAddress(host, port));
        return delegate.createSocket(proxySocket, host, port, true);
    }

    @Override
    public String[] getDefaultCipherSuites() {
        return delegate.getDefaultCipherSuites();
    }

    @Override
    public String[] getSupportedCipherSuites() {
        return delegate.getSupportedCipherSuites();
    }

    @Override
    public Socket createSocket(InetAddress host, int port) throws IOException {
        return delegate.createSocket(host, port);
    }

    @Override
    public Socket createSocket(String host, int port, InetAddress localHost, int localPort) throws IOException {
        return delegate.createSocket(host, port, localHost, localPort);
    }

    @Override
    public Socket createSocket(InetAddress address, int port, InetAddress localAddress, int localPort) throws IOException {
        return delegate.createSocket(address, port, localAddress, localPort);
    }
}

2. HTTP代理适配（适用于wss://协议） #

如果是HTTP代理，需要通过HTTP CONNECT方法建立隧道：

public class HttpSSLSocketFactory extends SSLSocketFactory {
    private final SSLSocketFactory delegate;
    private final String proxyHost;
    private final int proxyPort;

    public HttpSSLSocketFactory(SSLSocketFactory delegate, String proxyHost, int proxyPort) {
        this.delegate = delegate;
        this.proxyHost = proxyHost;
        this.proxyPort = proxyPort;
    }

    @Override
    public Socket createSocket(Socket s, String host, int port, boolean autoClose) throws IOException {
        Socket tunnelSocket = new Socket(proxyHost, proxyPort);
        OutputStream out = tunnelSocket.getOutputStream();
        out.write(("CONNECT " + host + ":" + port + " HTTP/1.1\r\nHost: " + host + "\r\n\r\n").getBytes());
        out.flush();

        BufferedReader in = new BufferedReader(new InputStreamReader(tunnelSocket.getInputStream()));
        String responseLine = in.readLine();
        if (responseLine == null || !responseLine.startsWith("HTTP/1.1 200")) {
            throw new IOException("Proxy tunnel failed: " + responseLine);
        }
        while (in.readLine() != null && !in.readLine().isEmpty());

        return delegate.createSocket(tunnelSocket, host, port, autoClose);
    }

    @Override
    public String[] getDefaultCipherSuites() {
        return delegate.getDefaultCipherSuites();
    }

    @Override
    public String[] getSupportedCipherSuites() {
        return delegate.getSupportedCipherSuites();
    }

    @Override
    public Socket createSocket(String host, int port) throws IOException {
        return createSocket(new Socket(), host, port, true);
    }

    @Override
    public Socket createSocket(InetAddress host, int port) throws IOException {
        return delegate.createSocket(host, port);
    }

    @Override
    public Socket createSocket(String host, int port, InetAddress localHost, int localPort) throws IOException {
        return delegate.createSocket(host, port, localHost, localPort);
    }

    @Override
    public Socket createSocket(InetAddress address, int port, InetAddress localAddress, int localPort) throws IOException {
        return delegate.createSocket(address, port, localAddress, localPort);
    }
}

3. 在你的代码中使用自定义SocketFactory #

替换原socketFactory = // code to get socket factory部分：

// 获取默认SSLSocketFactory
val defaultSslFactory = SSLSocketFactory.getDefault() as SSLSocketFactory
// SOCKS代理示例，替换为你的代理地址和端口
socketFactory = SocksSSLSocketFactory(defaultSslFactory, "proxy.example.com", 1080)

// HTTP代理示例
// socketFactory = HttpSSLSocketFactory(defaultSslFactory, "proxy.example.com", 8080)

若代理需要用户名密码认证，添加以下逻辑：

Authenticator.setDefault(new Authenticator() {
    @Override
    protected PasswordAuthentication getPasswordAuthentication() {
        return new PasswordAuthentication("proxyUser", "proxyPass".toCharArray());
    }
});

二、MQTT Android Service的PendingIntent Flag问题 #

Android 13（API 33+）要求创建PendingIntent时必须指定FLAG_IMMUTABLE或FLAG_MUTABLE，旧版本MQTT Android Service未处理此规则，解决方式如下：

1. 升级MQTT Android Service版本 #

升级至v1.3.0及以上版本，新版本已自动添加FLAG_IMMUTABLE，修复了该问题。

2. 自定义Service重写方法 #

无法升级时，继承MqttService并重写创建PendingIntent的方法：

public class CustomMqttService extends MqttService {
    @Override
    protected PendingIntent createPendingIntent() {
        Intent intent = new Intent(this, CustomMqttService.class);
        intent.setAction(MqttServiceConstants.ACTION_CONNECT);
        int flag = Build.VERSION.SDK_INT >= Build.VERSION_CODES.S ? PendingIntent.FLAG_IMMUTABLE : 0;
        return PendingIntent.getService(this, 0, intent, flag);
    }

    // 重写其他创建PendingIntent的方法（如createNotificationPendingIntent），同样添加Flag
}

在Manifest中注册自定义Service替换原MqttService：

<service android:name=".CustomMqttService" />

三、额外注意事项 #

• 确保代码中代理参数（类型、地址、端口）与设备系统代理配置完全一致
• 测试时可缩短connectionTimeout（如设为10秒），快速验证结果
• wss://协议要求代理支持HTTP CONNECT隧道，否则无法建立SSL连接

内容的提问来源于stack exchange，提问作者User 1765