MqttClientFactory mqttClientFactory = new DefaultMqttClientFactory();
        //使用redis消息存储器
        //MqttMsgStore mqttMsgStore = new RedisMqttMsgStore(new JedisPool("127.0.0.1", 6379));
        //mqttClientFactory.setMqttMsgStore(mqttMsgStore);
	//使用内存消息存储器
        MqttMsgStore mqttMsgStore = new MemoryMqttMsgStore();
        mqttClientFactory.setMqttMsgStore(mqttMsgStore);
        //创建连接参数，设置客户端ID
        MqttConnectParameter mqttConnectParameter = new MqttConnectParameter("xzc_test");
        //是否自动重连
        mqttConnectParameter.setAutoReconnect(true);
        //Host
        mqttConnectParameter.setHost("broker.emqx.io");
        //端口
        mqttConnectParameter.setPort(1883);
        //是否使用SSL/TLS
        mqttConnectParameter.setSsl(false);
        //是否清除会话
        mqttConnectParameter.setCleanSession(true);
        //心跳间隔
        mqttConnectParameter.setKeepAliveTimeSeconds(60);
        //连接超时时间
        mqttConnectParameter.setConnectTimeoutSeconds(30);
        //创建一个客户端
        MqttClient mqttClient = mqttClientFactory.createMqttClient(mqttConnectParameter);
        //添加回调器
        mqttClient.addMqttCallback(new DefaultMqttCallback());

        //阻塞连接至完成或超时
        mqttClient.connect();

	//非阻塞连接
	MqttFutureWrapper connectFuture = mqttClient.connectFuture();
	//阻塞至连接完成
        connectFuture.awaitComplete();

        //阻塞断开连接至完成
        mqttClient.disconnect();

	//非阻塞断开连接
	MqttFutureWrapper disconnectFuture = mqttClient.disconnectFuture();
	//阻塞至断开连接完成
        disconnectFuture.awaitComplete();

	//关闭客户端，关闭之后无法再进行操作
	mqttClient.close();

        //阻塞订阅至完成
        mqttClient.subscribe("testMqttClient", MqttQoS.AT_LEAST_ONCE);

        //非阻塞订阅
        MqttFutureWrapper subscribeFuture = mqttClient.subscribeFuture("testMqttClient", MqttQoS.AT_LEAST_ONCE);
        //阻塞至订阅完成
        subscribeFuture.awaitComplete();

        //阻塞取消订阅至完成
        mqttClient.unsubscribe("testMqttClient");

        //非阻塞取消订阅
        MqttFutureWrapper unsubscribeFuture = mqttClient.unsubscribeFuture("testMqttClient");
        //阻塞至取消订阅完成
        unsubscribeFuture.awaitComplete();

        //阻塞发送消息至完成
        mqttClient.publish("hello world!".getBytes(StandardCharsets.UTF_8), "testMqttClient", MqttQoS.EXACTLY_ONCE);

        //非阻塞发送消息
        MqttFutureWrapper publishFuture = mqttClient.publishFuture("hello world!".getBytes(StandardCharsets.UTF_8), "testMqttClient", MqttQoS.EXACTLY_ONCE);
        //阻塞至发送消息完成
        publishFuture.awaitComplete();

@Intercepts(type = {MqttClient.class})
public class LogInterceptorExample implements Interceptor {
    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        Object target = invocation.getTarget();
        Object[] args = invocation.getArgs();
        Method method = invocation.getMethod();
        //执行目标方法
        Object returnObj = invocation.proceed();
        LogUtils.info(LogInterceptorExample.class, "拦截目标：" + target.getClass().getSimpleName() + "，拦截方法：" + method.getName() + "，拦截参数：" + Arrays.toString(args) + "，拦截返回值：" + returnObj.toString());
        return returnObj;
    }
}