我是RxJava的新手，它试图从支持无损背压的RabbitMQ队列中实现消息的Observable。 我设法从Spring AMQP MessageListener创建了一个Observable。 这样可以在同步环境中很好地处理背压(例如，调用堆栈阻塞)，但是一旦引入了多个线程，背压就会如您所愿地消失在窗口之外。 该类如下：

import org.springframework.amqp.core.MessageListener;

import org.springframework.amqp.rabbit.connection.ConnectionFactory;

import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;

import org.springframework.amqp.support.converter.MessageConverter;

import org.springframework.stereotype.Component;

import rx.Observable;

import rx.subscriptions.Subscriptions;

import javax.inject.Inject;

@Component

public class CommandExchange {

private final MessageConverter messageConverter;

private final ConnectionFactory connectionFactory;

@Inject

public CommandExchange(MessageConverter messageConverter, ConnectionFactory connectionFactory) {

this.messageConverter = messageConverter;

this.connectionFactory = connectionFactory;

}

public Observable< T > observeQueue(String... queueNames) {

return Observable.create(subscriber -> {

SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();

container.setConnectionFactory(connectionFactory);

container.setQueueNames(queueNames);

container.setMessageListener((MessageListener) message -> {

T command = (T) messageConverter.fromMessage(message);

if (!subscriber.isUnsubscribed()) {

System.out.println("Being asked for a message.");

subscriber.onNext(command);

}

});

container.start();

Subscriptions.create(container::shutdown);

});

}

}

我无法在没有阻塞或缓冲的情况下了解如何实现无损反压。 使用缓冲没有任何意义，因为Rabbit MQ队列已经是一个缓冲区-因此，仅当订阅者准备好从队列中使用消息时，才应使用该消息。 解决方案是否是使用基于拉式的可观察的(即停止使用侦听器，而是在订户有需求时获取消息)？ 如果是这样，处理队列中当前没有消息的最佳实践是什么？

是的，我将停止使用侦听器，并按需从队列中获取消息。 请求记帐和背压都将为您处理，如果您使用

Observable.create(new SyncOnSubscribe< T >() {...});

在SyncOnSubscribe中，您或多或少只指定获取一条消息所采取的操作(如果没有等待，则不执行任何操作)。

在这方面需要更多帮助吗？ 如果您需要，我可以添加更多细节。

public Observable< T > observeQueue(String queueName) { Observable.OnSubscribe< T > stateless = SyncOnSubscribe.createStateless(s -> { s.onNext((T) rabbitTemplate.receiveAndConvert(queueName, -1)); }); return Observable.create(stateless).subscribeOn(Schedulers.newThread()); }这很好用，尽管我不确定在新线程上运行的阻塞观察器是否是最佳实践。