我目前正在阅读Spring AMQP的official sample project沿着Spring AMQP文档中的相应解释。该项目涉及sync和async两个版本,两者仅略有不同。以下是async版本:
生成器配置:
@Configuration
public class ProducerConfiguration {
protected final String helloWorldQueueName = "hello.world.queue";
@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
template.setRoutingKey(this.helloWorldQueueName);
return template;
}
@Bean
public ConnectionFactory connectionFactory() {
return new CachingConnectionFactory();
}
@Bean
public ScheduledProducer scheduledProducer() {
return new ScheduledProducer();
}
@Bean
public BeanPostProcessor postProcessor() {
return new ScheduledAnnotationBeanPostProcessor();
}
static class ScheduledProducer {
@Autowired
private volatile RabbitTemplate rabbitTemplate;
private final AtomicInteger counter = new AtomicInteger();
@Scheduled(fixedRate = 3000)
public void sendMessage() {
rabbitTemplate.convertAndSend("Hello World " + counter.incrementAndGet());
}
}
}
使用者配置:
@Configuration
public class ConsumerConfiguration {
protected final String helloWorldQueueName = "hello.world.queue";
@Bean
public ConnectionFactory connectionFactory() {
return new CachingConnectionFactory();
}
@Bean
public SimpleMessageListenerContainer listenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueueNames(this.helloWorldQueueName);
container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler()));
return container;
}
@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
template.setRoutingKey(this.helloWorldQueueName);
template.setDefaultReceiveQueue(this.helloWorldQueueName);
return template;
}
@Bean
public Queue helloWorldQueue() {
return new Queue(this.helloWorldQueueName);
}
}
你好世界处理程序:
public class HelloWorldHandler {
public void handleMessage(String text) {
System.out.println("Received: " + text);
}
}
正如医生解释的那样:
由于此示例演示了异步消息接收,因此生成端被设计为连续发送消息(如果它是类似于同步版本的每次执行消息模型,则它实际上不是一个消息驱动的使用者)。负责连续发送消息的组件被定义为ProducerConfiguration中的一个内部类。它被配置为每三秒运行一次。
我无法理解这段代码的“异步”是什么,因为,根据我的理解,在一个基本的“同步方式”中,像amqpTemplate.converAndSend()
和amqpTemplate.receiveAndConvert()
这样的操作已经执行了Rabbitmq的异步操作,当发送/接收消息时,生产者和消费者都没有阻塞。
那么,这个例子中的异步是如何表现出来的呢?如何在SpringAMQP上下文中理解异步与同步?
1条答案
按热度按时间2eafrhcq1#
对于async,
MessageListener
由框架调用;只要消息可用,它们就会到达。使用sync时,应用程序会呼叫receive方法,如果没有可用的消息,则会立即传回,或者封锁,直到消息到达或逾时过期为止。
在同步的情况下,应用程序控制何时接收消息,而在异步的情况下,框架处于控制之中。