Spring Integr手动将消息发布到频道

Question

我正在学习如何使用Java Spring框架，并开始尝试Spring集成。我试图使用Spring集成将我的应用程序连接到MQTT代理以发布和订阅消息，但我很难找到将消息手动发布到出站通道的方法。如果可能的话，我想专门使用java代码中的符号来构建它，而不是使用定义bean和其他相关配置的xml文件。

在每个示例中，我都看到手动发布消息的解决方案似乎是使用MessagingGateway接口，然后使用SpringApplicationBuilder获取ConfigurableApplicationContext，以在主方法中获取对网关接口的引用。然后使用该引用发布消息。是否可以将AutoWired用于接口？在我的尝试中，我只得到了一个空指针。

我的目标是构建一个游戏，在其中我订阅一个主题以获取游戏消息，然后每当用户准备好进行下一步操作时，就向该主题发布一条新消息。

更新：这是我一直在看的关于如何设置出站通道的示例之一：https://docs.spring.io/spring-integration/reference/html" target="_blank">html/mqtt.html

Gary Russel回答后更新2：

这是我在查看了一些示例后编写的示例代码，这些示例在运行Gateway时为网关使用@AutoWired时会得到一个null指针。控制器中的sendToMqtt。Java语言这里我想要实现的是在控制器处理GET请求时手动发送mqtt消息。

应用Java语言

@SpringBootApplication
public class Application {

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

控制器。Java语言

@RestController
@RequestMapping("/publishMessage")
public class Controller {

    @Autowired
    static Gateway gateway;

    @RequestMapping(method = RequestMethod.GET)
    public int request(){
        gateway.sendToMqtt("Test Message!");
        return 0;
    }
}

MqttPublisher。Java语言

@EnableIntegration
@Configuration
public class MqttPublisher {
    @Bean
    public MqttPahoClientFactory mqttClientFactory(){
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setServerURIs("tcp://localhost:1883");
        return factory;
    }

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

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

    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface Gateway {

        void sendToMqtt(String data);
    }
}

更新：

不确定这是否是正确的日志记录，但这是我从添加中得到的：

logging.level.org.springframework.web=Debug
logging.level.org.hibernate=Error

应用程序。属性。

https://hastebin.com/cuvonufeco.hs

Answer 1

使用消息网关或只需向通道发送消息。

编辑

@SpringBootApplication
public class So47846492Application {

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

    @Bean
    public ApplicationRunner runner(MyGate gate) {
        return args -> {
            gate.send("someTopic", "foo");
            Thread.sleep(5_000);
        };
    }

    @Bean
    @ServiceActivator(inputChannel = "toMqtt")
    public MqttPahoMessageHandler mqtt() {
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler("tcp://localhost:1883", "foo",
                clientFactory());
        handler.setDefaultTopic("myTopic");
        handler.setQosExpressionString("1");
        return handler;
    }

    @Bean
    public MqttPahoClientFactory clientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setUserName("guest");
        factory.setPassword("guest");
        return factory;
    }

    @Bean
    public MqttPahoMessageDrivenChannelAdapter mqttIn() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "bar", "someTopic");
        adapter.setOutputChannelName("fromMqtt");
        return adapter;
    }

    @ServiceActivator(inputChannel = "fromMqtt")
    public void in(String in) {
        System.out.println(in);
    }

    @MessagingGateway(defaultRequestChannel = "toMqtt")
    public interface MyGate {

        void send(@Header(MqttHeaders.TOPIC) String topic, String out);

    }

}