如何在java中使用mqtt从云（iothub）检索数据

Question

我是新来的。我已经成功地使用python将消息发送到IOT中心（D2C）。我们使用的协议是mqtt。我们正在尝试使用java从云（IOT中心）检索数据，但无法找到从云接收消息的正确方法。我的疑问是我们是否可以直接从IOT中心读取消息，或者我们需要将传入消息重定向到事件中心来检索消息。

我还尝试在向云发送数据的同时从java中的iothub读取消息，但得到的错误如下。。（与服务器的连接中断。重新连接0次。）

我使用此代码从iothub读取数据，

import com.microsoft.azure.sdk.iot.device.DeviceClient;
import com.microsoft.azure.sdk.iot.device.IotHubMessageResult;
import com.microsoft.azure.sdk.iot.device.Message;
import com.microsoft.azure.sdk.iot.device.MessageCallback;
import com.microsoft.azure.sdk.iot.device.IotHubClientProtocol;
import com.microsoft.azure.sdk.iot.service.sdk.IotHubServiceClientProtocol;
import java.io.IOException;
import java.net.URISyntaxException;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
public class Kafkareception {

    public static void main(String[] args) throws IOException {
        try {
            String connString = "HostName=";
            IotHubClientProtocol protocol = IotHubClientProtocol.MQTT;
            DeviceClient client = new DeviceClient(connString, protocol);

            MessageCallback callback = new AppMessageCallback();
            client.setMessageCallback(callback, null);
            client.open();
        } catch (URISyntaxException ex) {
            Logger.getLogger(Kafkareception.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    private static class AppMessageCallback implements MessageCallback {

        public IotHubMessageResult execute(Message msg, Object context) {
            System.out.println(new String(msg.getBytes(), Message.DEFAULT_IOTHUB_MESSAGE_CHARSET) + "Received message from hub: ");

            return IotHubMessageResult.COMPLETE;
        }
    }
}

Answer 1

我读了iothub的数据。我们可以用密码

import java.io.IOException;
import com.microsoft.azure.eventhubs.*;
import com.microsoft.azure.servicebus.*;

import java.nio.charset.Charset;
import java.time.*;
import java.util.function.*;

public class Datafetch {

    public static void main(String[] args) throws IOException {
        EventHubClient client0 = receiveMessages("0");
        EventHubClient client1 = receiveMessages("1");
        System.out.println("Press ENTER to exit.");
        System.in.read();
        try {
            client0.closeSync();
            client1.closeSync();
            System.exit(0);
        } catch (ServiceBusException sbe) {
            System.exit(1);
        }
    }

private static EventHubClient receiveMessages(final String partitionId) {

        String connStr = "Endpoint={youreventhubcompatibleendpoint};EntityPath={youreventhubcompatiblename};SharedAccessKeyName=iothubowner;SharedAccessKey={youriothubkey}";
        EventHubClient client = null;
        try {
            client = EventHubClient.createFromConnectionStringSync(connStr);
        } catch (Exception e) {
            System.out.println("Failed to create client: " + e.getMessage());
            System.exit(1);
        }
        try {
            client.createReceiver(
                    EventHubClient.DEFAULT_CONSUMER_GROUP_NAME,
                    partitionId,
                    Instant.now()).thenAccept(new Consumer<PartitionReceiver>() {
                        public void accept(PartitionReceiver receiver) {
                            System.out.println("** Created receiver on partition " + partitionId);
                            try {
                                while (true) {
                                    Iterable<EventData> receivedEvents = receiver.receive(100).get();
                                    System.out.println(receivedEvents);
                                    int batchSize = 0;
                                    if (receivedEvents != null) {
                                        for (EventData receivedEvent : receivedEvents) {
                                            System.out.println(String.format("Offset: %s, SeqNo: %s, EnqueueTime: %s",
                                                    receivedEvent.getSystemProperties().getOffset(),
                                                    receivedEvent.getSystemProperties().getSequenceNumber(),
                                                    receivedEvent.getSystemProperties().getEnqueuedTime()));
                                            System.out.println(String.format("| Device ID: %s", receivedEvent.getSystemProperties().getClass()));
                                            System.out.println(String.format("| Message Payload: %s", new String(receivedEvent.getBody(),
                                                    Charset.defaultCharset())));
                                            batchSize++;
                                        }
                                    }
                                    System.out.println(String.format("Partition: %s, ReceivedBatch Size: %s", partitionId, batchSize));
                                }
                            } catch (Exception e) {
                                System.out.println("Failed to receive messages: " + e.getMessage());
                            }
                        }
                    });
        } catch (Exception e) {
            System.out.println("Failed to create receiver: " + e.getMessage());
        }
        return client;
    }
}

Answer 2

根据您提供的信息，您可以尝试使用DeviceClient，设置一台设备到Azure IoT Hub的两个活动连接：一个是发送D2C消息，另一个是“从iothub读取数据”。出现错误的原因可能是：

IoT集线器仅支持每个设备一个活动MQTT连接。代表同一设备ID的任何新MQTT连接都会导致IoT Hub断开现有连接。

Ref：使用MQTT协议与IoT集线器通信。

如果您想接收发送到Azure IoT Hub的D2C消息，可以使用事件中心兼容endpoint（Java）。无需自行将传入消息重定向到事件中心。

IoT中心公开后端服务的内置终结点的消息/事件，以读取中心接收的设备到云消息。此终结点与事件中心兼容，可使用事件中心服务支持的任何机制读取消息。

Ref：了解Azure物联网中心消息传递和物联网中心endpoint。