当我运行RabbitMQ的Messaging指南[我尝试了这个演示应用程序] [1]时，侦听器仅收到一条消息（赛跑者发送一次然后退出）。我希望监听器可以连续监听消息队列，我应该如何实现呢？
接收方代码：
package hello;
import java.util.concurrent.CountDownLatch;
import org.springframework.stereotype.Component;
@Component
public class Receiver {
private CountDownLatch latch = new CountDownLatch(1);
public void receiveMessage(String message) {
System.out.println("Received <" + message + ">");
latch.countDown();
}
public CountDownLatch getLatch() {
return latch;
}
}
package hello;
import java.util.concurrent.TimeUnit;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
@Component
public class Runner implements CommandLineRunner {
private final RabbitTemplate rabbitTemplate;
private final Receiver receiver;
public Runner(Receiver receiver, RabbitTemplate rabbitTemplate) {
this.receiver = receiver;
this.rabbitTemplate = rabbitTemplate;
}
@Override
public void run(String... args) throws Exception {
System.out.println("Sending message...");
rabbitTemplate.convertAndSend(Application.topicExchangeName, "foo.bar.baz", "====1===========Hello from RabbitMQ!");
rabbitTemplate.convertAndSend(Application.topicExchangeName, "foo.bar.baz", "====2===========Hello from RabbitMQ!");
rabbitTemplate.convertAndSend(Application.topicExchangeName, "foo.bar.baz", "=====3==========Hello from RabbitMQ!");
receiver.getLatch().await(10000, TimeUnit.MILLISECONDS);
}
}
ackage hello;
import hello.test.TestListener;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
@SpringBootApplication
public class Application {
static final String topicExchangeName = "spring-boot-exchange";
public static final String queueName = "spring-boot";
@Bean
Queue queue() {
return new Queue(queueName, false);
}
@Bean
TopicExchange exchange() {
return new TopicExchange(topicExchangeName);
}
@Bean
Binding binding(Queue queue, TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with("foo.bar.#");
}
@Bean
SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueueNames(queueName);
container.setMessageListener(listenerAdapter);
return container;
}
@Bean
MessageListenerAdapter listenerAdapter(Receiver receiver) {
return new MessageListenerAdapter(receiver, "receiveMessage");
}
public static void main(String[] args) throws InterruptedException {
SpringApplication.run(Application.class, args).close();
}
}
[opuser@iZ25fprd8a9Z java]$ java -jar gs-messaging-rabbitmq-0.1.0.jar
. ____ _ __ _ _
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
' |____| .__|_| |_|_| |_\__, | / / / /
=========|_|==============|___/=/_/_/_/
:: Spring Boot :: (v2.1.1.RELEASE)
2019-01-10 09:40:53.846 INFO 412 --- [ main] hello.Application : Starting Application on iZ25fprd8a9Z with PID 412 (/data/workspace/test/java/gs-messaging-rabbitmq-0.1.0.jar started by opuser in /data/workspace/test/java)
2019-01-10 09:40:53.854 INFO 412 --- [ main] hello.Application : No active profile set, falling back to default profiles: default
2019-01-10 09:40:55.019 INFO 412 --- [ main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.amqp.rabbit.annotation.RabbitBootstrapConfiguration' of type [org.springframework.amqp.rabbit.annotation.RabbitBootstrapConfiguration$$EnhancerBySpringCGLIB$$870d6481] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2019-01-10 09:40:56.047 INFO 412 --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [10.171.135.109:5672]
2019-01-10 09:40:56.145 INFO 412 --- [ main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#76f2b07d:0/SimpleConnection@49c43f4e [delegate=amqp://[email protected]:5672/, localPort= 41068]
2019-01-10 09:40:56.152 INFO 412 --- [ main] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable, auto-delete, or exclusive Queue (spring-boot) durable:false, auto-delete:false, exclusive:false. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
2019-01-10 09:40:56.244 INFO 412 --- [ main] hello.Application : Started Application in 3.116 seconds (JVM running for 4.001)
Sending message...
Received <====1===========Hello from RabbitMQ!>
Received <====2===========Hello from RabbitMQ!>
Received <=====3==========Hello from RabbitMQ!>
2019-01-10 09:40:56.269 INFO 412 --- [ main] o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
2019-01-10 09:40:57.267 INFO 412 --- [ main] o.s.a.r.l.SimpleMessageListenerContainer : Successfully waited for workers to finish.
2019-01-10 09:40:57.272 INFO 412 --- [ main] o.s.a.r.l.SimpleMessageListenerContainer : Shutdown ignored - container is not active already
如果添加依赖项，则>
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency>您的申请将不会关闭。