java 中 publishsubscriber 的示例

在 sts / eclipse 中创建动态 web 项目下载 eclipse paho jar，点击这里下载并在 webcontent-> webinf-> folder-> lib 中粘贴 jar 文件

发布示例

 String broker = "tcp://localhost:1883"; 
String topicName = "test/topic";
int qos = 1;

 MqttClient mqttClient = new MqttClient(broker,String.valueOf(System.nanoTime()));
//Mqtt ConnectOptions is used to set the additional features to mqtt message

  MqttConnectOptions connOpts = new MqttConnectOptions();

    connOpts.setCleanSession(true); //no persistent session 
    connOpts.setKeepAliveInterval(1000);

MqttMessage message = new MqttMessage("Ed Sheeran".getBytes());

//这里 ed sheeran 是一条消息

    message.setQos(qos);     //sets qos level 1
    message.setRetained(true); //sets retained message 

MqttTopic topic2 = mqttClient.getTopic(topicName);

    mqttClient.connect(connOpts); //connects the broker with connect options
    topic2.publish(message);    // publishes the message to the topic(test/topic)

订阅示例

//We're using eclipse paho library  so we've to go with MqttCallback 
 MqttClient client = new MqttClient("tcp://localhost:1883","clientid");
     client.setCallback(this);
MqttConnectOptions mqOptions=new MqttConnectOptions();
     mqOptions.setCleanSession(true);
     client.connect(mqOptions);      //connecting to broker 
        client.subscribe("test/topic"); //subscribing to the topic name  test/topic

//Override methods from MqttCallback interface
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println("message is : "+message);
    }
.
.//other override methods 
.