Flutter MQTT连接报错SocketException: Connection reset by peer求助

原Java参考代码 #

private fun checkMqttConnection(): Pair<Int, String> {

    try {
        val sampleClient = MqttClient(broker, clientId, MemoryPersistence())
        val factory = getTrustAllHostsSSLSocketFactory()

        if (factory == null) {
            return Pair(1, getString(R.string.wifi_ko_test_no_test))
        }

        val connOpts = MqttConnectOptions().apply {
            isCleanSession = true
            userName = username
            password = pwd.toCharArray()
            socketFactory = factory
            connectionTimeout = 5
        }

        sampleClient.connect(connOpts)
        Log.d(TAG, "Connected")
        sampleClient.disconnect()
        Log.d(TAG, "Disconnected")
    } catch (me: MqttException) {
        if (me.toString().contains("SocketTimeoutException"))
            return Pair(1, getString(R.string.wifi_ko_test_no_port))
        if (me.toString().contains("UnknownHostException"))
            return Pair(1, getString(R.string.wifi_ko_test_no_dns))
        return Pair(1, getString(R.string.wifi_ko_test_no_connection))
    }
    return Pair(0, "Ok")
}

现有Flutter代码 #

Future<ApiResponse> checkMqttConnection() async {
    try {
      String clientID = "APP-${DateTime.now().millisecondsSinceEpoch}";
      String username = "USERNAME";
      String password = "PASSWORD";
      MqttServerClient client = MqttServerClient('XXXXXXXX', clientID);

      client.port = 8883;
      client.logging(on: true);

      await client.connect(username, password);
      print("CONNECTED");
      client.disconnect();
      print("DISCONEVTED");
      return ApiResponse.completed(null);
    } catch (e) {
      print(e);
      return ApiResponse.error(e.toString());
    }
  }

报错日志 #

I/flutter (19493): 1-2023-05-15 16:32:54.339294 -- Authenticating with username '{XXXXXXX}' and password '{XXXXXXX}'
I/flutter (19493): 1-2023-05-15 16:32:54.345251 -- MqttClient::connect - Connection timeout period is 5000 milliseconds
I/flutter (19493): 1-2023-05-15 16:32:54.350014 -- MqttClient::connect - keep alive is disabled
I/flutter (19493): 1-2023-05-15 16:32:54.355110 -- MqttConnectionHandlerBase::connect - server XXXXXXXX, port 8883
I/flutter (19493): 1-2023-05-15 16:32:54.357529 -- SynchronousMqttServerConnectionHandler::internalConnect entered
I/flutter (19493): 1-2023-05-15 16:32:54.357768 -- SynchronousMqttServerConnectionHandler::internalConnect - initiating connection try 0, auto reconnect in progress false
I/flutter (19493): 1-2023-05-15 16:32:54.358157 -- SynchronousMqttServerConnectionHandler::internalConnect - insecure TCP selected
I/flutter (19493): 1-2023-05-15 16:32:54.358849 -- SynchronousMqttServerConnectionHandler::internalConnect - calling connect
I/flutter (19493): 1-2023-05-15 16:32:54.359608 -- MqttNormalConnection::connect - entered
I/flutter (19493): 1-2023-05-15 16:32:54.515812 -- MqttServerConnection::_startListening
I/flutter (19493): 1-2023-05-15 16:32:54.518787 -- SynchronousMqttServerConnectionHandler::internalConnect - connection complete
I/flutter (19493): 1-2023-05-15 16:32:54.519274 -- SynchronousMqttServerConnectionHandler::internalConnect sending connect message
I/flutter (19493): 1-2023-05-15 16:32:54.520494 -- MqttConnectionHandlerBase::sendMessage - MQTTMessage of type MqttMessageType.connect
I/flutter (19493): Header: MessageType = MqttMessageType.connect, Duplicate = false, Retain = false, Qos = MqttQos.atMostOnce, Size = 0
I/flutter (19493): Connect Variable Header: ProtocolName=MQIsdp, ProtocolVersion=3, ConnectFlags=Connect Flags: Reserved1=false, CleanStart=true, WillFlag=false, WillQos=MqttQos.atMostOnce, WillRetain=false, PasswordFlag=true, UserNameFlag=true, KeepAlive=0
I/flutter (19493): MqttConnectPayload - client identifier is : APP-1684161174323
I/flutter (19493): 1-2023-05-15 16:32:54.551319 -- SynchronousMqttServerConnectionHandler::internalConnect - pre sleep, state = Connection status is connecting with return code of noneSpecified and a disconnection origin of none
I/flutter (19493): 1-2023-05-15 16:32:54.586137 -- MqttConnectionBase::_onError - calling disconnected callback
E/flutter (19493): [ERROR:flutter/runtime/dart_vm_initializer.cc(41)] Unhandled Exception: SocketException: Connection reset by peer (OS Error: Connection reset by peer, errno = 104), address = evlink-brk10.evco.services, port = 59406

问题分析与解决方案 #

报错Connection reset by peer的核心原因是：原Java代码使用了信任所有证书的SSLSocketFactory来建立MQTTs（8883端口）连接，但现有Flutter代码默认用不安全的TCP连接，未配置SSL/TLS，导致服务端直接拒绝连接。

完整修正后的Flutter代码 #

import 'dart:io';
import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';

Future<ApiResponse> checkMqttConnection() async {
  try {
    final String clientID = "APP-${DateTime.now().millisecondsSinceEpoch}";
    const String username = "USERNAME";
    const String password = "PASSWORD";
    const String broker = 'XXXXXXXX'; // 替换为你的broker地址

    // 初始化客户端
    final client = MqttServerClient(broker, clientID);
    client.port = 8883;
    client.logging(on: true);
    client.keepAlivePeriod = 0; // 和Java代码保持一致，禁用keep alive
    client.connectionTimeoutPeriod = 5000; // 5秒超时，对应Java的connectionTimeout=5

    // 配置SSL：信任所有证书（模拟Java的getTrustAllHostsSSLSocketFactory）
    final securityContext = SecurityContext.defaultContext;
    // 禁用证书验证（测试场景用，生产环境建议配置真实证书）
    securityContext.setTrustedCertificatesBytes([]);
    securityContext.alpnProtocols = ['mqtt'];
    client.securityContext = securityContext;
    // 开启SSL/TLS连接
    client.useTls = true;

    // 配置连接选项，对齐Java逻辑
    final connMessage = MqttConnectMessage()
        .withClientIdentifier(clientID)
        .startClean() // 对应Java的isCleanSession=true
        .withUsername(username)
        .withPassword(password)
        .keepAliveFor(0); // 禁用keep alive

    // 发起连接
    await client.connect(username, password);
    print("CONNECTED");
    
    // 断开连接
    client.disconnect();
    print("DISCONNECTED");
    
    return ApiResponse.completed(null);
  } catch (e) {
    print(e);
    // 错误分类处理，和Java原逻辑对齐
    String errorMsg;
    if (e.toString().contains("SocketTimeoutException")) {
      errorMsg = "端口不可达";
    } else if (e.toString().contains("UnknownHostException")) {
      errorMsg = "DNS解析失败";
    } else {
      errorMsg = "连接失败";
    }
    return ApiResponse.error(errorMsg);
  }
}

关键配置说明 #

• SSL/TLS启用：8883是MQTTs标准端口，必须设置client.useTls = true并配置securityContext
• 信任所有证书：通过securityContext.setTrustedCertificatesBytes([])模拟Java中跳过证书验证的逻辑，仅适合测试环境；生产环境需配置合法的服务端证书
• 连接参数对齐：
  • startClean()对应Java的isCleanSession = true
  • connectionTimeoutPeriod = 5000对应Java的connectionTimeout = 5
  • keepAliveFor(0)对应Java日志中显示的禁用keep alive设置

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