java - Spring Boot 未连接到 RabbitMQ

Question

更新:在 https://github.com/justsomecoder/boot-rabbit-issue 上发布了非工作版本

我正在尝试实现一些基本的 RabbitMQ 发送和接收功能，但似乎无法让它在我的两个 Spring Boot 项目中运行。虽然我尝试了多个示例，但截至目前，我在 https://spring.io/guides/gs/messaging-rabbitmq/ 实现了该示例，唯一的异常(exception)是我在 Application.java 中用 @Configuration 注释的单独 RabbitConfig 类中定义了 bean。

我也按照示例中的方式进行了尝试，但这也不起作用。

有趣的是，该实现在一个(较旧的)Spring Boot 项目中有效，而在较新的 Boot 项目中无效。在另一个项目中，运行时的输出显示与 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 管理界面中显示为队列的消费者。

在我的其他项目中，这些都没有显示。但是，在这两个项目中都找到了 Rabbit 配置类并加载了 bean(使用 Print all the Spring beans that are loaded - Spring Boot 检查)。

两个项目共享相同的依赖项“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!");
    }

}

Answer 1

看到 project you linked重现问题，当应用程序启动时，TcpServer component被初始化，这导致 startListen() 方法被执行。

该方法有一个无限循环，没有创建任何单独的线程，因此阻塞了剩余的启动过程(下面的代码摘录)。

public void startListen() throws IOException {
    while (true) {
        Socket clientSocket = serverSocket.accept();
        threadPool.submit(new XClientHandler(clientSocket,eventProducer));
    }
}

删除该组件表明 RabbitMQ 监听器已启动，因此我建议(至少)为您的 TcpServer 使用不同的 Thread。