Spring Cloud Stream 和 @Publisher 注解兼容性答案

【问题标题】：Spring Cloud Stream and @Publisher annotation compatiblitySpring Cloud Stream 和 @Publisher 注解兼容性
【发布时间】：2019-06-06 15:16:37
【问题描述】：

由于 Spring Cloud Stream 没有用于将新消息发送到流的注释（@SendTo 仅在声明 @StreamListener 时有效），因此我尝试为此使用 Spring Integration 注释，即 @Publisher。

由于Spring Cloud Stream的@Publisher带一个通道，而Spring Cloud Stream的@EnableBinding注解可以使用@Output注解绑定一个输出通道，所以我尝试通过以下方式混合它们：

@EnableBinding(MessageSource.class)
@Service
public class ExampleService {

    @Publisher(channel = MessageSource.OUTPUT)
    public String sendMessage(String message){
        return message;
    }
}

另外，我在配置文件中声明了@EnablePublisher 注解：

@SpringBootApplication
@EnablePublisher("")
public class ExampleApplication {

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

我的测试：

@RunWith(SpringRunner.class)
@SpringBootTest
public class ExampleServiceTest {

    @Autowired
    private ExampleService exampleService;

    @Test
    public void testQueue(){
        exampleService.queue("Hi!");
        System.out.println("Ready!");
    }
}

但我收到以下错误：

org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'com.example.ExampleServiceTest': Unsatisfied dependency expressed through field 'exampleService'; nested exception is 
org.springframework.beans.factory.BeanNotOfRequiredTypeException: Bean named 'exampleService' is expected to be of type 'com.example.ExampleService' but was actually of type 'com.sun.proxy.$Proxy86'

这里的问题是无法注入ExampleService bean。

有谁知道我怎样才能完成这项工作？

谢谢！

【问题讨论】：

标签： spring spring-boot spring-integration spring-cloud spring-cloud-stream

【解决方案1】：

由于您在 ExampleService 中使用了 @Publisher 注释，因此它被代理用于发布内容。

解决此问题的唯一方法是为您的ExampleService 公开一个接口，并将该接口注入您的测试类：

public interface ExampleServiceInterface {

     String sendMessage(String message);

}

...

public class ExampleService implements ExampleServiceInterface {

...


@Autowired
private ExampleServiceInterface exampleService;

另一方面，您的ExampleService.sendMessage() 似乎对消息没有任何作用，因此您可以考虑在某些界面上使用@MessagingGateway：https://docs.spring.io/spring-integration/reference/html/messaging-endpoints-chapter.html#gateway

【讨论】：

几件事：1）但是，例如，当在服务方法中使用 Cacheable 注释时，这也会从服务中创建一个代理，但我可以毫无问题地注入该代理服务声明 EnableCaching(proxyTargetClass =真）。也许有类似 EnableIntegration(proxyTargetClass = true) 的东西。使用 Publisher 复制所有服务类真的很烦人 2) 我读过 MessageGateway 并且有一个可注入网关很有趣，但我喜欢 Publisher 的想法，因为它就像一个方面一样工作。另一方面，我必须显式调用 gateway.sendMessage()
嗯，有办法，比如用setProxyTargetClass(true)而不是@EnablePublisher声明你自己的PublisherAnnotationBeanPostProcessor静态bean。但我确实同意你的观点，我们需要在 @EnablePublsiher 注释中添加这样的 proxyTargetClass 属性。请提出问题并欢迎贡献：github.com/spring-projects/spring-integration/issues！
关于此事的问题：github.com/spring-projects/spring-integration/issues/2695
嗨，我在这里提出了一个问题：github.com/spring-projects/spring-integration/issues/2695，但我被你的临时解决方案卡住了。我可以重用 PublisherAnnotationBeanPostProcessor 的实现并且只更改该标志吗？如果你愿意，我可以在堆栈溢出中添加另一个问题。
在注释修复之前，我添加了一个围绕here的工作。

【解决方案2】：

为什么不手动将消息发送到流，如下所示。

@Component
@Configuration
@EnableBinding(Processor.class)
public class Sender {

    @Autowired
    private Processor processor;

    public void send(String message) {

        processor.output().send(MessageBuilder.withPayload(message).build());

    }

}

您可以通过测试仪进行测试。

@SpringBootTest
public class SenderTest {

    @Autowired
    private MessageCollector messageCollector;

    @Autowired
    private Processor processor;

    @Autowired
    private Sender sender;

    @SuppressWarnings("unchecked")
    @Test
    public void testSend() throws Exception{

        sender.send("Hi!");
        Message<String> message = (Message<String>) this.messageCollector.forChannel(this.processor.output()).poll(1, TimeUnit.SECONDS);
        String messageData = message.getPayload().toString();
        System.out.println(messageData);

    }

}

您应该会看到“嗨！”在控制台中。

【讨论】：