ロボット研究開発、ソフトウェア開発、コンテンツ制作配信
ホーム > 記事一覧 > Iot開発 > 【Iot開発ノウハウ】MQTT通信を実装してみよう

【Iot開発ノウハウ】MQTT通信を実装してみよう

Iot開発 コメント


どうもこんにちは、InomaCreateです。
今回は、Iot開発でよく使われる通信プロトコル　「MQTT」通信について解説していこうと思います。
まずはMQTTの簡単な概要を説明し、その後、MQTT通信を使った簡単なシステムを実装していきたいと思います。

※実装したAndroidアプリとPythonのソースコードは、GitHubで公開しています。
Androidアプリ(MQTT Broker&Publisher)
lightsensor_matt_app
Pythonプログラム(MQTT Subscriber)
mqtt_python
スポンサーリンク

1.MQTTの概要

MQTTは、Message Queueing Telemetry Transportの略で以下のような特徴があります。
・デバイス間での短いメッセージを頻繁に送受信する
・軽量、省電力

Iot(Inter net of Things)は、モノのインターネットを呼ばれ、様々なモノやセンサー等がお互いに通信します。
ですので、上記の様な特徴を備えたMQTT通信は、Iot時代に適した通信方式です。

MQTTの仕組み

続いて、簡単にMQTTの仕組みを説明します。
MQTTには、「ブローカー」、「パブリッシャー」、「サブスクライバー」という３つの役割が存在します。

「ブローカー」は、MQTT通信を中継するサーバー的な役割です。
「パブリッシャー」はメッセージを送信する側、
「サブスクライバー」はメッセージを受信する側となります。
MQTTでやり取りするメッセージには、トピックと呼ばれる属性情報を記述して通信します。このトピックを使って、受信側はどのメッセージを受信するかどうかを選択でき、必要なメッセージのみ受信することが可能になります。

MQTTの仕組み　例１

以下に簡単に仕組みを説明していきます。
まず、受信側デバイスが受け取りたいメッセージ（トピックA）をブローカーへサブスクライブします。

送信側デバイスは、メッセージ（トピックA）をブローカーへパブリッシュします。
ブローカーは、トピックAをサブスクライブしているデバイスへメッセージを送信する流れとなります。

MQTTの仕組み　例２

以下の例では、上のサブスクライバーは、トピックAをサブスクライブ、下のサブスクライバーはトピックBをサブスクライブしています。

このようなパターンでは、
トピックAがパブリッシュされたとき、トピックAをサブスクライブしているデバイスのみに送信されます。

次にトピックBがパブリッシュされたとき、トピックBをサブスクライブしているデバイスのみに送信される仕組みになります。

MQTTの仕組みを出版業界に例えると

この仕組を出版業界に例えると、
ブローカーを書店
パブリッシャーを作者(出版社）
サブスクライバーを読者に置き換えるとわかりやすいと思います。
読者が本を書店に購買予約（サブスクライブ）し、作者（パブリッシャー）が本を出版（パブリッシュ）すると、書店（ブローカー）は購買予約している読者（サブスクライバー）に本を届けるイメージです。

MQTT通信も同じように受信デバイスが受信したいトピックをサブスクライブし、該当のトピックが発行されると、そのメッセージを受信できるイメージです。

2.今回作るシステム

次に今回作成する簡易的なMQTT通信システムについて説明します。
今回はAndroid端末をパブリッシャー兼、ブローカー、
PC側をサブスクライバーとします。
作成するシステムとしては、Android端末で取得した照度センサー値を定期的にPC上へ通知するシステムを作りたいと思います。

3.Android端末でパブリッシャー（ブローカーも含む）を作る

照度センサー値を取得する

まずは、Android端末で照度センサーの値を取得する処理を追加してみましょう。

AndroidStudioで新規プロジェクトを追加します。
MainActivity.javaに以下の処理を実装しましょう。

public class MainActivity extends AppCompatActivity implements SensorEventListener {

    private SensorManager mSensorManager;
    private Sensor mLight;
    private String TAG = "MainActivity";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        // センサーオブジェクトを取得
        mSensorManager = (SensorManager) getSystemService(Context.SENSOR_SERVICE);

        // 照度センサーのオブジェクトを取得
        mLight = mSensorManager.getDefaultSensor(Sensor.TYPE_LIGHT);


    }

    @Override
    protected void onResume(){
        super.onResume();

        // センサーのリスナー登録
        mSensorManager.registerListener(this,mLight,SensorManager.SENSOR_DELAY_NORMAL);

    }
    @Override
    protected void onPause() {
        super.onPause();
        // センサーを無効
        mSensorManager.unregisterListener(this);
    }

    @Override
    public void onSensorChanged(SensorEvent event) {
        if(event.sensor.getType() == Sensor.TYPE_LIGHT){
            Log.i(TAG,"Light!! value:"+event.values[0]);
            TextView textView = (TextView)findViewById(R.id.lightValueText);
            textView.setText(String.valueOf(event.values[0]));
        }

    }
    @Override
    public void onAccuracyChanged(Sensor sensor, int accuracy) {
        // センサーの精度を変更するときに使う
    }

}

センサーイベントを受信するため、MainActivityクラスにSensorEventListenerインターフェイスを実装します。(1行目)
onCreateで、センサーオブジェクト、照度センサーのオブジェクトを取得します。（12行目〜16行目）
Activityが開始された時に呼ばれるonResumeメソッドを実装します。
onResume内でセンサーリスナーを登録します。（２１行目〜２８行目）
また、Activityが一時停止した場合に呼ばれるonPauseメソッドを実装します。
onPause内でセンサーリスナーの登録を解除します。（29行目〜34行目）
最後に、onSensorChangedメソッドとonAccuracyChangedメソッドを実装します。
onSensorChangedメソッドはセンサー値の値が変化したときに呼ばれます。
event.values[0]内にセンサー値が入っていますので、その値をtextViewに表示させます。（36行目〜42行目）
onAccuracyChangedメソッドはセンサーの精度を変更するときに使います。今回は使いませんので中身は空にしておきます。（45行目〜48行目）

なお、照度センサー値は、textViewに表示させていますので、activity_main.xmlは以下のように変更しています。

<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".MainActivity">

    <TextView
        android:id="@+id/lightValueText"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:text=""
        app:layout_constraintBottom_toBottomOf="parent"
        app:layout_constraintLeft_toLeftOf="parent"
        app:layout_constraintRight_toRightOf="parent"
        app:layout_constraintTop_toTopOf="parent" />

</androidx.constraintlayout.widget.ConstraintLayout>

ここで一旦ビルドをかけると、以下の様なエラーが発生しました。（同じようなエラーが出た方は以下参考までに）
Installed Build Tools revision 31.0.0 is corrupted. Remove and install again using the SDK Manager.
BuildToolのバージョンが31.0.0では問題があるようですので、以下のようにbuildToolsVersionを”30.0.2″に変更することでエラー回避できました。

plugins {
    id 'com.android.application'
}

android {
    compileSdkVersion 31
    buildToolsVersion "30.0.2"

MQTT処理を実装する

今回、MQTTブローカーは、「Mosquitto」、MQTTクライアントは「Paho」というライブラリを使用します。
※Android端末側でブローカーも兼ねるので、Mosquittoも使います。

補足：gradleバージョンについて
ライブラリを使用するにあたり、gradleバージョン7.2では、ビルドエラーになりました。
もし、ビルドエラーになった方は、私が作成したgradleバージョンに合わせてみてください。（以下）
AndroidStudio File -> Project Structure -> Projectタブを開き、バージョンを確認。

・Android Gradle Plugin Version:4.1.1
・Gradle Version:6.5
なお、プロジェクトのbuild.gradleファイルは以下です。（参考までに）

// Top-level build file where you can add configuration options common to all sub-projects/modules.
buildscript {
    repositories {
        google()
        jcenter()
    }
    dependencies {
        classpath "com.android.tools.build:gradle:4.1.1"

        // NOTE: Do not place your application dependencies here; they belong
        // in the individual module build.gradle files
    }
}

allprojects {
    repositories {
        google()
        jcenter()
    }
}

task clean(type: Delete) {
    delete rootProject.buildDir
}

build.gradele内にライブラリを追加します。(44行目〜49行目）
複数ライブラリの衝突を避けるため、packagingOptionsを追加します。（29行目〜32行目）

plugins {
    id 'com.android.application'
}

android {
    compileSdkVersion 31
    buildToolsVersion "30.0.2"

    defaultConfig {
        applicationId "com.example.lightsensor_mqtt_app"
        minSdkVersion 26
        targetSdkVersion 31
        versionCode 1
        versionName "1.0"

        testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"
    }

    buildTypes {
        release {
            minifyEnabled false
            proguardFiles getDefaultProguardFile('proguard-android-optimize.txt'), 'proguard-rules.pro'
        }
    }
    compileOptions {
        sourceCompatibility JavaVersion.VERSION_1_8
        targetCompatibility JavaVersion.VERSION_1_8
    }
    packagingOptions {
        pickFirst "META-INF/INDEX.LIST"
        pickFirst "META-INF/io.netty.versions.properties"
    }
}

dependencies {

    implementation 'androidx.appcompat:appcompat:1.4.1'
    implementation 'com.google.android.material:material:1.5.0'
    implementation 'androidx.constraintlayout:constraintlayout:2.1.3'
    testImplementation 'junit:junit:4.+'
    androidTestImplementation 'androidx.test.ext:junit:1.1.3'
    androidTestImplementation 'androidx.test.espresso:espresso-core:3.4.0'
    
    implementation 'io.moquette:moquette-netty-parser:0.9'
    implementation 'io.moquette:moquette-broker:0.12.1'
    implementation 'io.moquette:moquette-parser-commons:0.8.1'

    implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.0'
    implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'
}

AndroidManifest.xml(マニフェストファイル)に権限関係の処理を追加します。

<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
    package="com.example.lightsensor_mqtt_app">

    <uses-permission android:name="android.permission.INTERNET" />
    <uses-permission android:name="android.permission.WAKE_LOCK"/>
    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE"/>


    <application
        android:allowBackup="true"
        android:icon="@mipmap/ic_launcher"
        android:label="@string/app_name"
        android:roundIcon="@mipmap/ic_launcher_round"
        android:supportsRtl="true"
        android:theme="@style/Theme.Lightsensor_mqtt_app">
        <activity android:name=".MainActivity">
            <intent-filter>
                <action android:name="android.intent.action.MAIN" />

                <category android:name="android.intent.category.LAUNCHER" />
            </intent-filter>
        </activity>
        <service android:name="org.eclipse.paho.android.service.MqttService" >
        </service>
    </application>

</manifest>

続いて、MainActivity.java　のonCreate内にBroker起動処理を追加します。

public class MainActivity extends AppCompatActivity implements SensorEventListener {

    private SensorManager mSensorManager;
    private Sensor mLight;
    private String TAG = "MainActivity";
    io.moquette.broker.Server broker;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        // Broker起動処理
        try {
            broker = new io.moquette.broker.Server();

            MemoryConfig memoryConfig = new MemoryConfig(new Properties());
            memoryConfig.setProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, this.getFilesDir() + BrokerConstants.DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME);

            broker.startServer(memoryConfig);
            Log.i(TAG,"startServer");

        } catch (IOException e){
            e.printStackTrace();
        }

起動したBrokerと接続する処理を追加します。
onCreateに以下を追加しましょう。
まずは、Android端末自体（自分自身）がブローカーなので、自分のIPアドレスを取得します。（29行目〜44行目）
その後、ブローカーとの接続処理を実行します。（46-71行目）

public class MainActivity extends AppCompatActivity implements SensorEventListener {

    private SensorManager mSensorManager;
    private Sensor mLight;
    private String TAG = "MainActivity";
    io.moquette.broker.Server broker;
    String IPAddress=null;
    private MqttAndroidClient mqttAndroidClient;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        // Broker起動処理
        try {
            broker = new io.moquette.broker.Server();

            MemoryConfig memoryConfig = new MemoryConfig(new Properties());
            memoryConfig.setProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, this.getFilesDir() + BrokerConstants.DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME);

            broker.startServer(memoryConfig);
            Log.i(TAG,"startServer");

        } catch (IOException e){
            e.printStackTrace();
        }

        // 自分のIPアドレスを取得する
        try {
            for (NetworkInterface n : Collections.list(NetworkInterface.getNetworkInterfaces())) {
                for (InetAddress addr : Collections.list(n.getInetAddresses())) {
                    if(addr instanceof Inet4Address && !addr.isLoopbackAddress()) {
                        Log.i(TAG,"IP ADDRESS:"+addr.getHostAddress());
                        IPAddress = addr.getHostAddress();
                        TextView textView = (TextView)findViewById(R.id.textIP);
                        textView.setText(IPAddress);
                    }
                }
            }
        } catch (SocketException e){
            Log.e(TAG,"IP address not get!!");
            e.printStackTrace();
        }

        // クライアントブローカー接続
        if(IPAddress!=null) {
            mqttAndroidClient = new MqttAndroidClient(this, "tcp://" + IPAddress + ":1883", "test-android") {
                @Override
                public void onReceive(Context context, Intent intent) {
                    super.onReceive(context, intent);
                }
            };
            try {
                MqttConnectOptions options = new MqttConnectOptions();
                mqttAndroidClient.connect(options, null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken iMqttToken) {
                        Log.d(TAG, "onSuccess");
                    }

                    @Override
                    public void onFailure(IMqttToken iMqttToken, Throwable throwable) {
                        Log.d(TAG, "onFailure");
                    }
                });

            } catch (MqttException e) {
                Log.d(TAG, e.toString());
            }
        }

あとからPC側でブローカーと接続するために、IPアドレスをアプリ画面に表示しておきます。（上記36-37行目）
activity_main.xmlに以下のようにTextViewを追加します。

<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".MainActivity">

    <TextView
        android:id="@+id/lightValueText"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:text=""
        app:layout_constraintBottom_toBottomOf="parent"
        app:layout_constraintLeft_toLeftOf="parent"
        app:layout_constraintRight_toRightOf="parent"
        app:layout_constraintTop_toTopOf="parent" />

    <TextView
        android:id="@+id/textView"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_marginStart="40dp"
        android:text="IP Address"
        app:layout_constraintBottom_toBottomOf="parent"
        app:layout_constraintStart_toStartOf="parent"
        app:layout_constraintTop_toTopOf="parent"
        app:layout_constraintVertical_bias="0.091" />

    <TextView
        android:id="@+id/textIP"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:text=""
        app:layout_constraintBottom_toBottomOf="parent"
        app:layout_constraintEnd_toEndOf="parent"
        app:layout_constraintHorizontal_bias="0.065"
        app:layout_constraintStart_toEndOf="@+id/textView"
        app:layout_constraintTop_toTopOf="parent"
        app:layout_constraintVertical_bias="0.091" />

</androidx.constraintlayout.widget.ConstraintLayout>

最後に、照度センサー値をパブリッシュする処理を追加します。
まずは、センサー値を保存しておくための変数：LightValueを追加します。

public class MainActivity extends AppCompatActivity implements SensorEventListener {

    private SensorManager mSensorManager;
    private Sensor mLight;
    private String TAG = "MainActivity";
    io.moquette.broker.Server broker;
    String IPAddress=null;
    private MqttAndroidClient mqttAndroidClient;
    private int LightValue;

LightValueに取得したセンサー値を保存しましょう。

    @Override
    public void onSensorChanged(SensorEvent event) {
        if(event.sensor.getType() == Sensor.TYPE_LIGHT){
            Log.i(TAG,"Light!! value:"+event.values[0]);
            TextView textView = (TextView)findViewById(R.id.lightValueText);
            textView.setText(String.valueOf(event.values[0]));
            LightValue = (int)event.values[0];
        }

    }

今回センサー値は、１秒毎にパブリッシュするようにしたいと思いますので、timerを追加します。
TimerTask()クラスのrunメソッド内で、トピック「sensor/light」で照度センサー値をbyte型配列に格納してパブリッシュしています。（25行目)

public class MainActivity extends AppCompatActivity implements SensorEventListener {

    private SensorManager mSensorManager;
    private Sensor mLight;
    private String TAG = "MainActivity";
    io.moquette.broker.Server broker;
    String IPAddress=null;
    private MqttAndroidClient mqttAndroidClient;
    private int LightValue;
    private Timer timer = new Timer();
    
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                if(mqttAndroidClient == null)
                    return;

                try {
                    Log.i(TAG,"publish!!");
                    mqttAndroidClient.publish("sensor/light", String.valueOf(LightValue).getBytes(), 0, true);
                } catch (MqttPersistenceException e) {
                    Log.d(TAG,e.toString());
                } catch (MqttException e) {
                    Log.d(TAG,e.toString());
                }

            }
        },1000,1000);

これでAndroid端末側の処理ができました。

4.PC側でサブスクライバーを作る

次に、Android端末でパブリッシュしている照度センサー値をPC側（サブスクライバー）で受信するための処理を追加します。
言語はPythonで記載します。（※pythonのバージョンは、Python 3.8.5です）

まずは、MQTTクライアントのライブラリPahoをインストールしましょう。
Python3をインストールしていれば、MACであれば、以下を実行するとインストールできます。

$ pip3 install paho-mqtt

続いて、サブスクライブ処理をPythonで実装します。
以下のように実装しましょう。

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flag, rc):
    print("Connected with result code " + str(rc))
    client.subscribe("sensor/light")
 
def on_disconnect(client, userdata, flag, rc):
    if rc != 0:
        print("Unexpected disconnection.") 
 
 
def on_message(client, userdata, msg):
    msg.payload = msg.payload.decode("utf-8")
    print("Receive message ="+str(msg.payload)+" on topic = "+ msg.topic+ " with Qos "+str(msg.qos))
 
client = mqtt.Client()
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_message = on_message
 
try:
    client.connect("192.168.11.8",1883,300)
except:
    print("broker host is not started!! waiting!!")
 
client.loop_forever()

まず、client = mqtt.Client()でMQTTクライアントを作成します。（16行目）
それぞれブローカーと接続された時、切断時、メッセージを受けた時のコールバックメソッドを追加します。（17-19行目）

client.connect(IPアドレス、ポート番号、待受時間）でクライントをブローカーと接続します。
※IPアドレスは、Android端末側のIPアドレスを設定します。
client.loop_forever()は接続されるまでループして待ち続ける処理です。

def on_connectが、ブローカーと接続した際に呼ばれる処理です。
「sensor/light」トピックをサブスクライブしています。※Android端末側で決めたトピック名です。

def on_disconnectは、ブローカーと切断された時に呼ばれる処理です。

def on_messageがメッセージを受信した際の処理です。
msg.payloadに受信したデータが入っています。
実際に受信したデータは、byte型配列なので、utf-8でデコードします。
受信したメッセージをプリントで表示させています。

5.動かしてみる

では、Android端末でアプリを起動し、PC側でPythonスクリプトを実行してみましょう。
以下のように、Android端末で取得した照度センサー値が、PC側で受信できていればOKです！

動画解説

YouTubeに解説動画もアップしていますので、よろしければどうぞ。

【Iot開発ノウハウ】MQTT通信を実装してみよう(Part1:MQTT通信の概要)

【Iot開発ノウハウ】MQTT通信を実装してみよう(Part2:Android端末でセンサー値を取得する)

【Iot開発ノウハウ】MQTT通信を実装してみよう(Part3: Android端末でMQTT通信処理を追加する)

【Iot開発ノウハウ】MQTT通信を実装してみよう(Part4: MQTT受信側処理の実装)

6.最後に

以上、今回はMQTT通信の簡単なシステムを実装してみました。
今回の実装を応用していけば、色々なIotシステムが作れそうですね。
少しでも参考になれば幸いです。

それでは！

関連記事(一部広告含む)

コメントを残す

メールアドレスが公開されることはありません。 * が付いている欄は必須項目です

CAPTCHA


ページトップボタン