問題描述

所以我已經(jīng)讓 MQTT -> MQTT 和 AMQP -> AMQP 工作了；MQTT -> AMQP 的翻譯似乎并沒有在某處工作.這是我的測試，如果我的監(jiān)聽器"也在使用 paho 的 MQTT 中，則它通過了，但是這個 rabbitmq 實(shí)現(xiàn)沒有.

So I've gotten MQTT -> MQTT and AMQP -> AMQP to work; the translation of MQTT -> AMQP doesn't seem to be working somewhere though. Here's my test, it passes if my "listener" is also in MQTT using paho, but this rabbitmq implementation doesn't.

@SpringBootTest
@SpringJUnitConfig
internal open class ProvisioningTest @Autowired constructor(
    private val mqtt: IMqttAsyncClient,
    private val mapper: ObjectMapper
) {

    @Test
    fun provision() {
        val entity = Foley(
            rfid = UUID.randomUUID().toString(),
        )

        val called = AtomicBoolean(false)
        mqtt.subscribe("foley/created", 1) { _, _ -> called.set(true) }

        mqtt.publish("foley/new", MqttMessage(mapper.writeValueAsBytes(entity)))

        Awaitility.await().atMost(10, TimeUnit.SECONDS).untilTrue(called)
    }
}

這是將保存的實(shí)體發(fā)布到另一個隊(duì)列的偵聽器；當(dāng)我發(fā)布到 MQTT 時，它永遠(yuǎn)不會被調(diào)用.

this is the listener that publishes the saved entity to the other queue; it never gets called when I publish to MQTT.

@Service
open class Provisioning(private val repo: FoleyRepo) {
    private val log: Logger = LogManager.getLogger(this::class.java)

    @SendTo("foley.created")
    @RabbitListener(queuesToDeclare = [Queue("foley.new")] )
    open fun listen(entity: Foley): Foley {
        log.trace("saving: {}", entity)
        val save = repo.save(entity)
        log.debug("saved: {}", save)
        return save
    }

}

我的全部消息配置

@Configuration
open class MessagingConfig {

    @Bean
    open fun client(
        @Value("tcp://${mqtt.client.host:localhost}:${mqtt.client.port:1883}") uri: String,
        @Value("${mqtt.client.user:#{null}}") user: String?,
        @Value("${mqtt.client.pass:#{null}}") pass: CharArray?
    ): IMqttAsyncClient {

        val connOpt = MqttConnectOptions()
        user?.let { connOpt.userName = it }
        pass?.let { connOpt.password = it }
        connOpt.isCleanSession = false
        connOpt.isAutomaticReconnect = true
        val client = MqttAsyncClient(uri, MqttAsyncClient.generateClientId(), MemoryPersistence())
        client.connect(connOpt)
        return client
    }

    @Bean
    open fun messageConverter( om: ObjectMapper): MessageConverter {
        return Jackson2JsonMessageConverter(om)
    }

    @Bean
    open fun builder(): Jackson2ObjectMapperBuilderCustomizer {
        return Jackson2ObjectMapperBuilderCustomizer {
            it.modules(JavaTimeModule(), KotlinModule())
        }
    }
}

使用啟用了 mqtt 的 官方 docker rabbitmq 映像.

using the official docker rabbitmq image with mqtt enabled.

我需要糾正什么才能完成這項(xiàng)工作?

What do I need to correct to make this work?

推薦答案

MQTT插件發(fā)布到amq.topic，以mqtt主題名作為路由key.

The MQTT plugin publishes to the amq.topic with the mqtt topic name as the routing key.

在消費(fèi)者端，它使用路由鍵將自動刪除隊(duì)列綁定到該交換；在以下示例中，隊(duì)列名為 mqtt-subscription-mqttConsumerqos1.

On the consumer side, it binds an auto-delete queue to that exchange, with the routing key; in the following example, the queue is named mqtt-subscription-mqttConsumerqos1.

為了通過 AMQP 接收 MQTT 消息，您需要將自己的隊(duì)列綁定到交換器.這是一個例子:

In order to receive MQTT messages over AMQP, you need to bind your own queue to the exchange. Here is an example:

@SpringBootApplication
public class So54995261Application {

    public static void main(String[] args) {
        SpringApplication.run(So54995261Application.class, args);
    }

    @Bean
    @ServiceActivator(inputChannel = "toMQTT")
    public MqttPahoMessageHandler sendIt(MqttPahoClientFactory clientFactory) {
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler("clientId", clientFactory);
        handler.setAsync(true);
        handler.setDefaultTopic("so54995261");
        return handler;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[] { "tcp://localhost:1883" });
        options.setUserName("guest");
        options.setPassword("guest".toCharArray());
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    public MessageProducerSupport mqttInbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("mqttConsumer",
                mqttClientFactory(), "so54995261");
        adapter.setCompletionTimeout(5000);
        return adapter;
    }

    @Bean
    public IntegrationFlow flow() {
        return IntegrationFlows.from(mqttInbound())
                .handle(System.out::println)
                .get();
    }

    @RabbitListener(queues = "so54995261")
    public void listen(byte[] in) {
        System.out.println(new String(in));
    }

    @Bean
    public Queue queue() {
        return new Queue("so54995261");
    }

    @Bean
    public Binding binding() {
        return new Binding("so54995261", DestinationType.QUEUE, "amq.topic", "so54995261", null);
    }

    @Bean
    public ApplicationRunner runner(MessageChannel toMQTT) {
        return args -> toMQTT.send(new GenericMessage<>("foo"));
    }

}

這篇關(guān)于如何通過 MQTT 傳輸并使用 RabbitMQ 和 Spring-AMQP 在 AMQP 上接收的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，也希望大家多多支持html5模板網(wǎng)！