问题描述
我尝试使用 Spring Integration 从 RabbitMQ 接收 protobuf 消息。
我的集成流程:
public class FacadeIntegrationFlowAdapter extends IntegrationFlowAdapter {
@SuppressWarnings("rawtypes")
private final Facade facade;
private final FacadeProperties facadeProperties;
@SuppressWarnings("unchecked")
@Override
protected IntegrationFlowDeFinition<?> buildFlow() {
return from(facadeProperties.getQueueName())
.handle(facade::getNewMessages);
}
}
getNewMessages
方法:
@Override
public ExchangeResponse getNewMessages(Message<ExchangeRequest> message) {
ExchangeRequest request = message.getPayload();
log.info("Receiving new message: " + request.toString());
这就是我将消息发送到队列的方式。使测试易于遵循是如此简单。
ExchangeRequest request = ExchangeRequest.newBuilder()
.addAllAuthors(List.of("author1","author2"))
.addAllBooks(List.of("book1","book2"))
.build();
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setUsername("user");
connectionFactory.setPassword("password");
connectionFactory.setHost("localhost");
connectionFactory.setPort(24130);
try {
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
var basicProperties = new AMQP.BasicProperties().builder()
.contentType("application/x-protobuf")
.type(request.getDescriptorForType().getFullName())
.build();
channel.basicpublish(
"facade-exchange","facade-routing-key",basicProperties,request.toByteArray());
} catch (IOException e) {
不幸的是,我不断收到异常:
com.google.protobuf.InvalidProtocolBufferException: Type of the Any message does not match the given class.
但是,当我将 getNewMessages
方法更改为以下内容时,一切似乎都很好。
@Override
public ExchangeResponse getNewMessages(Message message) {
try {
Any payload = (Any) message.getPayload();
ByteString value = payload.getValue();
ExchangeRequest request = ExchangeRequest.parseFrom(value);
log.info("Receiving new message: " + request.toString());
我哪里出错了?发送!
解决方法
暂无找到可以解决该程序问题的有效方法,小编努力寻找整理中!
如果你已经找到好的解决方法,欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@)