Иногда мы используем его в наших проектахобмен мгновенными сообщениямиФункция,Например, функция чата службы поддержки клиентов в системе электронной коммерции.,Также в процессе оплаты,Когда пользователь успешно платит,Сторонний платежный сервис перезвонит нам через интерфейс обратного вызова.,На этом этапе нам нужно уведомить фронтенд об успешном совершении платежа.。недавно обнаруженRabbitMQМожет быть легко реализованобмен мгновенными сообщениямиФункция,Если у вас нет особых деловых потребностей,Вам даже не нужно писать внутренний код,Сегодня я расскажу вам, как им пользоваться.RabbitMQосознатьобмен мгновенными сообщениями！

Протокол MQTT

MQTT（Message Queuing Telemetry Transport,Транспортный протокол телеметрии очереди сообщений),является издательской базой/подписка（publish/subscribe）шаблонлегкийПротокол связи,Этот протокол построен наTCP/IPпо соглашению。MQTTСамым большим преимуществом является,С минимальным кодом и ограниченной пропускной способностью,Предоставляйте надежные услуги обмена сообщениями в режиме реального времени для подключения удаленных устройств.

Связанные понятия

• Издатель: отправитель сообщения, ответственный за отправку сообщения.
• Подписчик: подписчик сообщения, ответственный за получение и обработку сообщения.
• Брокер: брокер сообщений,Между издателями сообщений и подписчиками,Все виды поддержки Протокол Можно использовать промежуточное программное обеспечение сообщений MQTT.
• Тема: можно понимать как маршрутизацию в очереди сообщений. После того, как подписчик подпишется на тему, он или она может получать сообщения, отправленные в эту тему.
• Полезную нагрузку можно понимать как содержимое отправленного сообщения.
• QoS (качество сообщения): качество полного имени of Service,То есть качество отправки сообщения,В основном включаютQoS 0、QoS 1、QoS 2три уровня,Давайте представим их отдельно ниже：
  • QoS 0 (почти один раз): не более одного раза, отправляется только один раз, может произойти потеря или дублирование сообщения;
  • QoS 1 (хотя бы один раз): сообщение гарантированно прибудет хотя бы один раз, но может произойти дублирование сообщения;
  • QoS 2（Exactly Один раз): Только один раз, гарантируя, что сообщение придет только один раз.

RabbitMQ включает функциональность MQTT

RabbitMQ включает функцию MQTT, вам необходимо установить, а затем включить плагин.

• Сначала нам нужно установить и запустить RabbitMQ.,верноRabbitMQДрузья, кто не знает, могут обратиться«Прошло 3 дня. Подведем итог практические советы по RabbitMQ, есть что! 》；
• Следующим шагом является включение подключаемого модуля MQTT для RabbitMQ. По умолчанию он не включен. Просто используйте следующую команду, чтобы включить его.

rabbitmq-plugins enable rabbitmq_mqtt

• После успешного открытия,Просмотр консоли управления,мы можем найтиMQTTСлужба работает1883Порт работает。

Это может быть полезный для вас проект с открытым исходным кодом.,Проект торгового центра представляет собой комплекс SpringBoot3 + JDK 17 + Vue выполнитьиз Система электронной коммерции（Звезда Github 60 тыс.）,Развертывание с использованием Docker-контейнеров,Серверная часть поддерживает многомодульную и микросервисную архитектуру. Включая внешний проект торгового центра и внутреннюю систему управления.,Может поддерживать полную Заказпроцесс！Охватываемая продукция、Заказ、корзина、Разрешения、Купон、член、Оплата и другие функции!

• Bootпроект：https://github.com/macrozheng/mall
• Cloudпроект：https://github.com/macrozheng/mall-swarm
• Видеоурок：https://www.macrozheng.com/video/

Демонстрация проекта:

MQTT-клиент

Мы можем использовать MQTT-клиент для тестирования обмена MQTT. мгновенными сообщениями Функция,Здесь используетсяMQTTBoxЭтот клиентский инструмент。

• Сначала скачайте и установитеMQTTBox,Адрес загрузки: http://workswithweb.com/mqttbox.html.

• НажмитеCreate MQTT Clientкнопка для созданияMQTT-клиент；

• Далее Конфигурация для MQTT-клиента,В основном Конфигурация Хороший порт протокола、соединять Имя пользователя и парольиQoSВот и все；

• Конфигурация еще одного абонента,подписка ВОЗподпискаtestTopicAэта тема,Мы отправим сообщение в эту тему;

• Издатель публикует сообщения в теме, а подписчики могут получать их в режиме реального времени.

Фронтенд напрямую реализует обмен мгновенными сообщениями

теперь этоMQTTBoxКлиент может напрямую пройтиRabbitMQвыполнитьобмен мгновенными сообщениями, можем ли мы напрямую использовать интерфейсные технологии для достижения обмена мгновенными сообщениями？Ответ: да！Ниже мы пройдемсяhtml+javascriptвыполнитьпростойизчат Функция,Реально обеспечьте обмен мгновенными сообщениями, не написав ни единой строки внутреннего кода!

• Поскольку базовое взаимодействие между RabbitMQ и Интернетом использует WebSocket, нам необходимо включить поддержку MQTT WEB RabbitMQ. Просто используйте следующую команду, чтобы включить ее:

rabbitmq-plugins enable rabbitmq_web_mqtt

• После успешного открытия,Просмотр консоли управления,мы можем найтиMQTTизWEBСлужба работает15675Порт работает；

• WEBДуань иMQTTСервисам необходимо взаимодействовать с помощью функции под названиемMQTT.jsиз Библиотека,Адрес проекта: https://github.com/mqttjs/MQTT.js

• Реализованная функция очень проста,Единая функция чата,Следует отметить, что адрес доступа службы «Конфигурация хорошей MQTT»: ws://localhost:15675/ws.

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<div>
    <label>ЦельTopic：<input id="targetTopicInput" type="text"></label><br>
    <label>Отправить сообщение：<input id="messageInput" type="text"></label><br>
    <button onclick="sendMessage()">отправлять</button>
    <button onclick="clearMessage()">Прозрачный</button>
    <div id="messageDiv"></div>
</div>
</body>
<script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
<script>
    //Адрес подключения RabbitMQ к web-mqtt
    const url = 'ws://localhost:15675/ws';
    //Получаем подписанную тему
    const topic = getQueryString("topic");
    //Подключаемся к очереди сообщений
    let client = mqtt.connect(url);
    client.on('connect', function () {
        //Подписка на тему после успешного подключения
        client.subscribe(topic, function (err) {
            if (!err) {
                showMessage("Подписаться на тему:" + topic + "успех!");
            }
        });
    });
    //Получаем сообщения в теме подписки
    client.on('message', function (topic, message) {
        showMessage("Получено сообщение:" + message.toString());
    });

    //Отправить сообщение
    function sendMessage() {
        let targetTopic = document.getElementById("targetTopicInput").value;
        let message = document.getElementById("messageInput").value;
        //Отправляем сообщение в целевую тему
        client.publish(targetTopic, message);
        showMessage("Отправить сообщение" + targetTopic + "Сообщение:" + message);
    }

    //Получаем параметры из URL
    function getQueryString(name) {
        let reg = new RegExp("(^|&)" + name + "=([^&]*)(&|$)", "i");
        let r = window.location.search.substr(1).match(reg);
        if (r != null) {
            return decodeURIComponent(r[2]);
        }
        return null;
    }

    //Отображать сообщения в списке сообщений
    function showMessage(message) {
        let messageDiv = document.getElementById("messageDiv");
        let messageEle = document.createElement("div");
        messageEle.innerText = message;
        messageDiv.appendChild(messageEle);
    }

    //Очищаем список сообщений
    function clearMessage() {
        let messageDiv = document.getElementById("messageDiv");
        messageDiv.innerHTML = "";
    }
</script>
</html>

• Далее подписываемся на разные темы и открываем две страницы для проверки функции (страницы размещаются в директории ресурсов приложения SpringBoot),Вам необходимо запустить приложение, прежде чем получить к нему доступ）：
  • первыйподпискатемаtestTopicA,Адрес доступа: http://localhost:8088/page/index?topic=testTopicA
  • второйподпискатемаtestTopicB,Адрес доступа: http://localhost:8088/page/index?topic=testTopicB
• Затем отправьте друг другу сообщения и посмотрим эффект!

Используется в SpringBoot

Когда нет особых бизнес-потребностей, интерфейс можно напрямую подключить к RabbitMQ для обмена мгновенными сообщениями. Но иногда нам необходимо уведомить интерфейсную часть через сервер. На данный момент нам необходимо интегрировать MQTT в приложение. Далее давайте поговорим о том, как использовать MQTT в приложении SpringBoot.

• Сначала нам нужноpom.xmlДобавитьMQTTСвязанные зависимости；

<!--SpringинтегрированныйMQTT-->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
</dependency>

• существоватьapplication.ymlДобавитьMQTTСвязанный Конфигурация,В основном Адрес доступа、Имя пользователя и пароль、Информация о теме по умолчанию;

rabbitmq:
  mqtt:
    url: tcp://localhost:1883
    username: guest
    password: guest
    defaultTopic: testTopic

• Напишите класс конфигурации Java для чтения конфигурации из файла конфигурации для удобства использования;

/**
 * MQTTСвязанный Конфигурация * Created by macro on 2020/9/15.
 */
@Data
@EqualsAndHashCode(callSuper = false)
@Component
@ConfigurationProperties(prefix = "rabbitmq.mqtt")
public class MqttConfig {
    /**
     * Имя пользователя соединения RabbitMQ
     */
    private String username;
    /**
     * Пароль подключения RabbitMQ
     */
    private String password;
    /**
     * Тема MQTT по умолчанию RabbitMQ
     */
    private String defaultTopic;
    /**
     * Адрес подключения MQTT RabbitMQ
     */
    private String url;
}

• добавить в Конфигурация, связанная с подписчиком сообщения MQTT,использовать@ServiceActivatorАннотация объявляет активатор службы,проходитьMessageHandlerсправитьсяподпискаинформация；

/**
 * Конфигурация, связанная с подписчиком сообщения MQTT
 * Created by macro on 2020/9/15.
 */
@Slf4j
@Configuration
public class MqttInboundConfig {
    @Autowired
    private MqttConfig mqttConfig;

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getUrl(), "subscriberClient",
                        mqttConfig.getDefaultTopic());
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        //Установить качество сообщения：0->максимум один раз；1->хотя бы один раз；2->только один раз
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                //Обработка сообщений о подписке
                log.info("handleMessage : {}",message.getPayload());
            }

        };
    }
}

• добавить в Конфигурация, связанная с издателем сообщений MQTT；

/**
 * Конфигурация, связанная с издателем сообщений MQTT
 * Created by macro on 2020/9/15.
 */
@Configuration
public class MqttOutboundConfig {

    @Autowired
    private MqttConfig mqttConfig;

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[] { mqttConfig.getUrl()});
        options.setUserName(mqttConfig.getUsername());
        options.setPassword(mqttConfig.getPassword().toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler =
                new MqttPahoMessageHandler("publisherClient", mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());
        return messageHandler;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }
}

• Добавить MQTT-шлюз для отправки сообщений в тему;

/**
 * Шлюз MQTT, передающий данные через интерфейс в поток интеграции.
 * Created by macro on 2020/9/15.
 */
@Component
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
    /**
     * Отправить сообщение в тему по умолчанию
     */
    void sendToMqtt(String payload);

    /**
     * Отправить сообщение в указанную тему
     */
    void sendToMqtt(String payload, @Header(MqttHeaders.TOPIC) String topic);

    /**
     * Отправить сообщение в указанную тему и установите QOS
     */
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
}

• Добавьте тестовый интерфейс MQTT и используйте шлюз MQTT для отправки сообщений по определенным темам;

/**
 * Тестовый интерфейс MQTT
 * Created by macro on 2020/9/15.
 */
@Api(tags = "MqttController", description = "Тестовый интерфейс MQTT")
@RestController
@RequestMapping("/mqtt")
public class MqttController {

    @Autowired
    private MqttGateway mqttGateway;

    @PostMapping("/sendToDefaultTopic")
    @ApiOperation("Отправить сообщение в тему по умолчанию")
    public CommonResult sendToDefaultTopic(String payload) {
        mqttGateway.sendToMqtt(payload);
        return CommonResult.success(null);
    }

    @PostMapping("/sendToTopic")
    @ApiOperation("Отправить сообщение в указанную тему")
    public CommonResult sendToTopic(String payload, String topic) {
        mqttGateway.sendToMqtt(payload, topic);
        return CommonResult.success(null);
    }
}

• Вызов интерфейса для отправки сообщений в тему для тестирования;

• Сообщение успешно получено и распечатано в фоновом режиме.

2020-09-17 14:29:01.689  INFO 11192 --- [ubscriberClient] c.m.mall.tiny.config.MqttInboundConfig   : handleMessage : Сообщения из Интернета
2020-09-17 14:29:06.101  INFO 11192 --- [ubscriberClient] c.m.mall.tiny.config.MqttInboundConfig   : handleMessage : Сообщения из Интернета
2020-09-17 14:29:07.384  INFO 11192 --- [ubscriberClient] c.m.mall.tiny.config.MqttInboundConfig   : handleMessage : Сообщения из Интернета

Подвести итог

Промежуточное программное обеспечение для сообщений используется все более широко. Оно может не только реализовывать надежную асинхронную связь, но и осуществлять мгновенную связь. По-прежнему необходимо освоить промежуточное программное обеспечение для обмена сообщениями. Если нет особых бизнес-потребностей, клиент или интерфейсная часть могут напрямую использовать MQTT для подключения к промежуточному программному обеспечению для обмена сообщениями. Если есть особые потребности, вы также можете использовать SpringBoot для интеграции MQTT. хороший способ добиться обмена мгновенными сообщениями.

Адрес исходного кода проекта

https://github.com/macrozheng/mall-learning/tree/master/mall-tiny-mqtt

GithubначальствоСтандартная звезда 60Киз Практика электронной коммерциипроектmall,полный комплект Видеоурок (последняя версия 2023 г.) Обновлено！полный комплект Учебникоколо 40 часов,Всего 113 выпусков,проходить这套Учебник你可以拥有一个Полный опыт реализации проектов, охватывающий основные стеки технологий Java.,Совершенствуйтесь одновременноВозможность самостоятельно разработать проект,Ниже представлена ​​общая архитектурная схема проекта.,заинтересованныйиз Друзья могут Нажмите Связь видеоурок по торговому центру Присоединяйтесь к обучению.