EMQX 简单使用

导入MAVEN依赖

        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.2</version>
        </dependency>

编写消息发布者

package com.scsoul.emqx;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.Scanner;

public class App {
    
    public static void main(String[] args) {
    
        String broker = "tcp://localhost:1883";
        String clientId = "user01";
        //Use the memory persistence
        MemoryPersistence persistence = new MemoryPersistence();
        try {
    
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            System.out.println("Connecting to broker:" + broker);
            sampleClient.connect(connOpts);
            System.out.println("Connected");

            String topic = "fire/topics";
            String content = "";
            int qos = 2;
            while (true) {
    
                Scanner scanner = new Scanner(System.in);
                content = scanner.nextLine();
                MqttMessage message = new MqttMessage(content.getBytes());
                message.setQos(qos);
                sampleClient.publish(topic, message);
                System.out.println("Message published");
            }

        } catch (MqttException me) {
    
            System.out.println("reason" + me.getReasonCode());
            System.out.println("msg" + me.getMessage());
            System.out.println("loc" + me.getLocalizedMessage());
            System.out.println("cause" + me.getCause());
            System.out.println("excep" + me);
            me.printStackTrace();
        }
    }
}

编辑消息订阅者

package com.scsoul.emqx;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.text.MessageFormat;

public class App2 {
    
    public static void main(String[] args) {
    
        String broker = "tcp://localhost:1883";
        String clientId = "user02";
        //Use the memory persistence
        MemoryPersistence persistence = new MemoryPersistence();
        try {
    
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            System.out.println("Connecting to broker:" + broker);
            sampleClient.connect(connOpts);
            System.out.println("Connected");
            sampleClient.subscribe("fire/topics");
            sampleClient.setCallback(new MqttCallback() {
    
                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
    
                    System.out.println(new String(message.getPayload()));
                }
                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
    
                }
                @Override
                public void connectionLost(Throwable throwable) {
    
                }
            });
        } catch (MqttException me) {
    
            System.out.println("reason" + me.getReasonCode());
            System.out.println("msg" + me.getMessage());
            System.out.println("loc" + me.getLocalizedMessage());
            System.out.println("cause" + me.getCause());
            System.out.println("excep" + me);
            me.printStackTrace();
        }
    }
}