延迟subscribe可能导致消息消费不到 about emqx HOT 6 CLOSED

public static void main(String[] args) throws Exception {
    MqttSubscribe subscribe = new MqttSubscribe();
    subscribe.connect("emqx_test_subscribe_10");

    new Thread(() -> {
        try {
           //延迟 20s
            Thread.sleep(10 * 1000);
            // 订阅
            MqttSubscription mqttSubscription = new MqttSubscription("testtopic10/10/#", 1);
            System.out.println(simpleDateFormat.format(System.currentTimeMillis()) + "subscribe");
            subscribe.client.subscribe(new MqttSubscription[]{mqttSubscription}, new IMqttMessageListener[]{new DefaultIMqttMessageListener()});
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        }
    }).start();
}

void connect(String clientId) throws Exception {
    MemoryPersistence persistence = new MemoryPersistence();

    String broker = "xxxxx";
    client = new MqttClient(broker, clientId, persistence);
    // MQTT 连接选项
    MqttConnectionOptions connOpts = new MqttConnectionOptions();
    connOpts.setUserName("test");
    connOpts.setPassword("test".getBytes());
    // 保留会话
    connOpts.setCleanStart(false);
    connOpts.setSessionExpiryInterval(7200L);
    //自动重连
    connOpts.setAutomaticReconnect(true);

    // 建立连接
    System.out.println("Connecting to broker: " + broker);
    client.connect(connOpts);
}