Spring 引导未连接到 RabbitMQ
Spring Boot not connecting to RabbitMQ
更新:在 https://github.com/justsomecoder/boot-rabbit-issue
上发布了非工作版本
我正在尝试实现一些基本的 RabbitMQ 发送和接收功能,但似乎无法让它在我的两个 Spring 引导项目中工作。虽然我尝试了多个示例,但截至目前,我在 https://spring.io/guides/gs/messaging-rabbitmq/ 实现了该示例,唯一的例外是我在单独的 RabbitConfig class 中定义了 Application.java 中的 bean,并用 @Configuration 进行了注释.
我也按照示例中的方式进行了尝试,但这也不起作用。
有趣的是,该实现在一个(较旧的)Spring 引导项目中有效,而在较新的引导项目中不起作用。在另一个项目中,运行时的输出显示与 RabbitMQ 的连接已成功设置:
部分输出工作项目
2017-12-18 10:53:45,205 INFO [restartedMain] o.s.j.e.a.AnnotationMBeanExporter [MBeanExporter.java:431] Registering beans for JMX exposure on startup
2017-12-18 10:53:45,215 INFO [restartedMain] o.s.j.e.a.AnnotationMBeanExporter [MBeanExporter.java:916] Bean with name 'rabbitConnectionFactory' has been autodetected for JMX exposure
2017-12-18 10:53:45,219 INFO [restartedMain] o.s.j.e.a.AnnotationMBeanExporter [MBeanExporter.java:678] Located managed bean 'rabbitConnectionFactory': registering with JMX server as MBean [org.springframework.amqp.rabbit.connection:name=rabbitConnectionFactory,type=CachingConnectionFactory]
2017-12-18 10:53:45,238 INFO [restartedMain] o.s.c.s.DefaultLifecycleProcessor [DefaultLifecycleProcessor.java:343] Starting beans in phase -2147482648
2017-12-18 10:53:45,239 INFO [restartedMain] o.s.c.s.DefaultLifecycleProcessor [DefaultLifecycleProcessor.java:343] Starting beans in phase 2147483647
2017-12-18 10:53:45,268 INFO [container-1] o.s.a.r.c.CachingConnectionFactory [AbstractConnectionFactory.java:359] Created new connection: SimpleConnection@6bb6caa0 [delegate=amqp://guest@127.0.0.1:32770/, localPort= 51889]
2017-12-18 10:53:45,273 INFO [container-1] o.s.a.r.c.RabbitAdmin [RabbitAdmin.java:491] Auto-declaring a non-durable, auto-delete, or exclusive Queue (floors) 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.
输出非工作项目
2017-12-18 18:40:39.452 INFO 4722 --- [ restartedMain] name.nameFootstepProcessor : Starting nameFootstepProcessor on Macbook-Pro-van-Lars.local with PID 4722 (/Users/lars/IdeaProjects/name-footstep-processor/target/classes started by lars in /Users/lars/IdeaProjects/name-footstep-processor)
2017-12-18 18:40:39.453 INFO 4722 --- [ restartedMain] name.nameFootstepProcessor : No active profile set, falling back to default profiles: default
2017-12-18 18:40:39.453 DEBUG 4722 --- [ restartedMain] o.s.boot.SpringApplication : Loading source class name.nameFootstepProcessor
2017-12-18 18:40:39.945 INFO 4722 --- [ restartedMain] s.c.a.AnnotationConfigApplicationContext : Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@426b0d00: startup date [Mon Dec 18 18:40:39 CET 2017]; root of context hierarchy
2017-12-18 18:40:40.697 INFO 4722 --- [ restartedMain] .s.d.r.c.RepositoryConfigurationDelegate : Multiple Spring Data modules found, entering strict repository configuration mode!
2017-12-18 18:40:41.389 INFO 4722 --- [ restartedMain] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.amqp.rabbit.annotation.RabbitBootstrapConfiguration' of type [org.springframework.amqp.rabbit.annotation.RabbitBootstrapConfiguration$$EnhancerBySpringCGLIB$$b410db6c] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
它还在 RabbitMQ Admin 管理界面中显示为队列的消费者。
在我的其他项目中,显示了 none。但是,在这两个项目中,都找到了 Rabbit 配置 class 并加载了 bean(使用 检查)。
两个项目共享相同的依赖项'spring-boot-starter-amqp'。它们还共享相同的 application.properties 文件,其中包含用于连接到我的本地 RabbitMQ 服务器的正确信息。我能做些什么来找出为什么一个项目正在加载 Rabbit
配置 bean 正确而另一个不正确?
下面我附上了一些我认为有助于进一步理解问题的文件,但如果需要任何其他文件或输出,请告诉我。我更改了一些包名
出于隐私考虑。
谢谢!
此致,
larsl95.
工作项目
Spring 启动 1.5.3
非工作项目
Spring 引导 1.5.9
application.properties(两个项目相同,端口归因于Docker,映射到5276)
spring.rabbitmq.host=localhost
spring.rabbitmq.port=32770
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
RabbitConfig.java(两个项目相同)
package name.configuration;
import org.springframework.context.ConfigurableApplicationContext;
import name.queue.Receiver;
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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import java.util.Arrays;
@Component
@Configuration
public class RabbitConfig {
public final static String queueName = "floors";
public RabbitConfig(ConfigurableApplicationContext ctx) {
this.printBeans(ctx);
}
@Bean
Queue queue() {
return new Queue(queueName, false);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("test-exchange");
}
@Bean
Binding binding(Queue queue, TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(queueName);
}
@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");
}
// just added this to know which beans are loaded, called it from constructor to see
// if RabbitConfig class is found by Spring at all
private void printBeans(ConfigurableApplicationContext ctx) {
String[] beanNames = ctx.getBeanDefinitionNames();
Arrays.sort(beanNames);
for (String beanName : beanNames) {
System.out.println(beanName);
}
}
}
Receiver.java(两个项目相同)
package name.queue;
import org.springframework.stereotype.Component;
import java.util.concurrent.CountDownLatch;
@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;
}
}
Main.java(非工作项目)
@SpringBootApplication
public class Main {
public static void main(String[] args) {
SpringApplication app = new SpringApplication(Main.class);
app.setWebEnvironment(false);
ConfigurableApplicationContext appContext = app.run(args);
name.tcp.TcpServer tcpServer = new name.tcp.TcpServer();
}
}
Main.java(工作项目)
package name;
import name.tcp.TcpServer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.support.SpringBootServletInitializer;
import org.springframework.context.ConfigurableApplicationContext;
import java.io.IOException;
@SpringBootApplication
public class Main extends SpringBootServletInitializer {
public static void main(String[] args) {
ConfigurableApplicationContext app = SpringApplication.run(Main.class, args);
TcpServer tcpServer = new TcpServer();
System.out.println("starting TCP server from main");
}
}
Runner.java(两个项目相同)
package name.queue;
import java.util.concurrent.TimeUnit;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.stereotype.Component;
import name.configuration.RabbitConfig;
@Component
public class Runner implements CommandLineRunner {
private final RabbitTemplate rabbitTemplate;
@Autowired
public Runner(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
@Override
public void run(String... args) throws Exception {
System.out.println("Sending message...");
rabbitTemplate.convertAndSend(RabbitConfig.queueName, "Hello from RabbitMQ!");
}
}
看到project you linked that reproduced the issue, when the application is started, the TcpServer component被初始化,导致startListen()
方法被执行
该方法有一个无限循环,没有创建任何单独的线程,因此阻塞了启动过程的剩余部分(下面的代码摘录)。
public void startListen() throws IOException {
while (true) {
Socket clientSocket = serverSocket.accept();
threadPool.submit(new XClientHandler(clientSocket,eventProducer));
}
}
删除该组件表明 RabbitMQ 侦听器已启动,因此我建议(至少)为您的 TcpServer
.
使用不同的 Thread
更新:在 https://github.com/justsomecoder/boot-rabbit-issue
上发布了非工作版本我正在尝试实现一些基本的 RabbitMQ 发送和接收功能,但似乎无法让它在我的两个 Spring 引导项目中工作。虽然我尝试了多个示例,但截至目前,我在 https://spring.io/guides/gs/messaging-rabbitmq/ 实现了该示例,唯一的例外是我在单独的 RabbitConfig class 中定义了 Application.java 中的 bean,并用 @Configuration 进行了注释.
我也按照示例中的方式进行了尝试,但这也不起作用。
有趣的是,该实现在一个(较旧的)Spring 引导项目中有效,而在较新的引导项目中不起作用。在另一个项目中,运行时的输出显示与 RabbitMQ 的连接已成功设置:
部分输出工作项目
2017-12-18 10:53:45,205 INFO [restartedMain] o.s.j.e.a.AnnotationMBeanExporter [MBeanExporter.java:431] Registering beans for JMX exposure on startup
2017-12-18 10:53:45,215 INFO [restartedMain] o.s.j.e.a.AnnotationMBeanExporter [MBeanExporter.java:916] Bean with name 'rabbitConnectionFactory' has been autodetected for JMX exposure
2017-12-18 10:53:45,219 INFO [restartedMain] o.s.j.e.a.AnnotationMBeanExporter [MBeanExporter.java:678] Located managed bean 'rabbitConnectionFactory': registering with JMX server as MBean [org.springframework.amqp.rabbit.connection:name=rabbitConnectionFactory,type=CachingConnectionFactory]
2017-12-18 10:53:45,238 INFO [restartedMain] o.s.c.s.DefaultLifecycleProcessor [DefaultLifecycleProcessor.java:343] Starting beans in phase -2147482648
2017-12-18 10:53:45,239 INFO [restartedMain] o.s.c.s.DefaultLifecycleProcessor [DefaultLifecycleProcessor.java:343] Starting beans in phase 2147483647
2017-12-18 10:53:45,268 INFO [container-1] o.s.a.r.c.CachingConnectionFactory [AbstractConnectionFactory.java:359] Created new connection: SimpleConnection@6bb6caa0 [delegate=amqp://guest@127.0.0.1:32770/, localPort= 51889]
2017-12-18 10:53:45,273 INFO [container-1] o.s.a.r.c.RabbitAdmin [RabbitAdmin.java:491] Auto-declaring a non-durable, auto-delete, or exclusive Queue (floors) 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.
输出非工作项目
2017-12-18 18:40:39.452 INFO 4722 --- [ restartedMain] name.nameFootstepProcessor : Starting nameFootstepProcessor on Macbook-Pro-van-Lars.local with PID 4722 (/Users/lars/IdeaProjects/name-footstep-processor/target/classes started by lars in /Users/lars/IdeaProjects/name-footstep-processor)
2017-12-18 18:40:39.453 INFO 4722 --- [ restartedMain] name.nameFootstepProcessor : No active profile set, falling back to default profiles: default
2017-12-18 18:40:39.453 DEBUG 4722 --- [ restartedMain] o.s.boot.SpringApplication : Loading source class name.nameFootstepProcessor
2017-12-18 18:40:39.945 INFO 4722 --- [ restartedMain] s.c.a.AnnotationConfigApplicationContext : Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@426b0d00: startup date [Mon Dec 18 18:40:39 CET 2017]; root of context hierarchy
2017-12-18 18:40:40.697 INFO 4722 --- [ restartedMain] .s.d.r.c.RepositoryConfigurationDelegate : Multiple Spring Data modules found, entering strict repository configuration mode!
2017-12-18 18:40:41.389 INFO 4722 --- [ restartedMain] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.amqp.rabbit.annotation.RabbitBootstrapConfiguration' of type [org.springframework.amqp.rabbit.annotation.RabbitBootstrapConfiguration$$EnhancerBySpringCGLIB$$b410db6c] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
它还在 RabbitMQ Admin 管理界面中显示为队列的消费者。
在我的其他项目中,显示了 none。但是,在这两个项目中,都找到了 Rabbit 配置 class 并加载了 bean(使用
两个项目共享相同的依赖项'spring-boot-starter-amqp'。它们还共享相同的 application.properties 文件,其中包含用于连接到我的本地 RabbitMQ 服务器的正确信息。我能做些什么来找出为什么一个项目正在加载 Rabbit 配置 bean 正确而另一个不正确?
下面我附上了一些我认为有助于进一步理解问题的文件,但如果需要任何其他文件或输出,请告诉我。我更改了一些包名 出于隐私考虑。
谢谢!
此致, larsl95.
工作项目 Spring 启动 1.5.3
非工作项目 Spring 引导 1.5.9
application.properties(两个项目相同,端口归因于Docker,映射到5276)
spring.rabbitmq.host=localhost
spring.rabbitmq.port=32770
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
RabbitConfig.java(两个项目相同)
package name.configuration;
import org.springframework.context.ConfigurableApplicationContext;
import name.queue.Receiver;
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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import java.util.Arrays;
@Component
@Configuration
public class RabbitConfig {
public final static String queueName = "floors";
public RabbitConfig(ConfigurableApplicationContext ctx) {
this.printBeans(ctx);
}
@Bean
Queue queue() {
return new Queue(queueName, false);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("test-exchange");
}
@Bean
Binding binding(Queue queue, TopicExchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(queueName);
}
@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");
}
// just added this to know which beans are loaded, called it from constructor to see
// if RabbitConfig class is found by Spring at all
private void printBeans(ConfigurableApplicationContext ctx) {
String[] beanNames = ctx.getBeanDefinitionNames();
Arrays.sort(beanNames);
for (String beanName : beanNames) {
System.out.println(beanName);
}
}
}
Receiver.java(两个项目相同)
package name.queue;
import org.springframework.stereotype.Component;
import java.util.concurrent.CountDownLatch;
@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;
}
}
Main.java(非工作项目)
@SpringBootApplication
public class Main {
public static void main(String[] args) {
SpringApplication app = new SpringApplication(Main.class);
app.setWebEnvironment(false);
ConfigurableApplicationContext appContext = app.run(args);
name.tcp.TcpServer tcpServer = new name.tcp.TcpServer();
}
}
Main.java(工作项目)
package name;
import name.tcp.TcpServer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.support.SpringBootServletInitializer;
import org.springframework.context.ConfigurableApplicationContext;
import java.io.IOException;
@SpringBootApplication
public class Main extends SpringBootServletInitializer {
public static void main(String[] args) {
ConfigurableApplicationContext app = SpringApplication.run(Main.class, args);
TcpServer tcpServer = new TcpServer();
System.out.println("starting TCP server from main");
}
}
Runner.java(两个项目相同)
package name.queue;
import java.util.concurrent.TimeUnit;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.stereotype.Component;
import name.configuration.RabbitConfig;
@Component
public class Runner implements CommandLineRunner {
private final RabbitTemplate rabbitTemplate;
@Autowired
public Runner(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
@Override
public void run(String... args) throws Exception {
System.out.println("Sending message...");
rabbitTemplate.convertAndSend(RabbitConfig.queueName, "Hello from RabbitMQ!");
}
}
看到project you linked that reproduced the issue, when the application is started, the TcpServer component被初始化,导致startListen()
方法被执行
该方法有一个无限循环,没有创建任何单独的线程,因此阻塞了启动过程的剩余部分(下面的代码摘录)。
public void startListen() throws IOException {
while (true) {
Socket clientSocket = serverSocket.accept();
threadPool.submit(new XClientHandler(clientSocket,eventProducer));
}
}
删除该组件表明 RabbitMQ 侦听器已启动,因此我建议(至少)为您的 TcpServer
.
Thread