Spring引导应用程序未能从错误开始：“应用程序需要一个bean，但是找到了两个

Question

在使用spring-integration-mqtt依赖项编写Spring应用程序时，我使用CommandLineRunner bean在应用程序启动时启动MQTTSubscriber。

但是，在运行应用程序时，会出现以下错误：

Error starting ApplicationContext. To display the conditions report re-run your application with 'debug' enabled.
2019-01-26 01:48:40.386 ERROR 59171 --- [           main] o.s.b.d.LoggingFailureAnalysisReporter   : 
***************************
APPLICATION FAILED TO START
***************************
Description:
Field MessageListener in im.sma.mqtt.mqttclient.DemoApplication required a single bean, but 2 were found:
    - messageListener: defined in file [/Users/sma/sandbox/slidecab/mqtt/mqtt-client/target/classes/im/sma/mqtt/mqttclient/config/MessageListener.class]
    - integrationHeaderChannelRegistry: defined in null
Action:
Consider marking one of the beans as @Primary, updating the consumer to accept multiple beans, or using @Qualifier to identify the bean that should be consumed

我注意到，当我从代码中删除以下部分时，错误就消失了：

@Autowired
Runnable MessageListener;

@Bean
public CommandLineRunner schedulingRunner(TaskExecutor executor) {
    return new CommandLineRunner() {
        public void run(String... args) throws Exception {
            executor.execute(MessageListener);
        }
    };
}

这是发生错误的DemoApplication类：

@SpringBootApplication
public class DemoApplication extends SpringBootServletInitializer {
    @Autowired
    Runnable MessageListener;

    @Override
    protected SpringApplicationBuilder configure(SpringApplicationBuilder application) {
        return application.sources(DemoApplication.class);
    }

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }
    @Bean
    public CommandLineRunner schedulingRunner(TaskExecutor executor) {
        return new CommandLineRunner() {
            public void run(String... args) throws Exception {
                executor.execute(MessageListener);
            }
        };
    }
}

此外，我还有下面的AppConfig类来设置TaskExecutor

@Component
public class AppConfig {

    @Bean
    @Primary
    public TaskExecutor taskExecutor() {
        return new SimpleAsyncTaskExecutor();
    }
}

这就是MessageListener豆，它不能自动运行：

@Component
public class MessageListener implements Runnable {

    @Autowired
    MQTTSubscriberBase subscriber;

    @Override
    public void run() {
        while(true) {
            subscriber.subscribeMessage("demoTopic2019");
        }

    }
}

此外，我还有下面的配置来设置MQTTSubscriber

public abstract class MQTTConfig {
    protected final String broker = "localhost";
    protected final int qos = 2;
    protected Boolean hasSSL = false; /* By default SSL is disabled */
    protected Integer port = 1883; /* Default port */
    protected final String userName = "guest" ;//"testUserName";
    protected final String password = "guest";//"demoPassword";
    protected final String TCP = "tcp://";
    protected final String SSL = "ssl://";

    protected abstract void config(String broker, Integer port, Boolean ssl, Boolean withUserNamePass);

    protected abstract void config();
}

public interface MQTTSubscriberBase {
    public static final Logger logger = LoggerFactory.getLogger(MQTTSubscriberBase.class);

    public void subscribeMessage(String topic);
    public void disconnect();
}

@Component
public class MQTTSubscriber extends MQTTConfig implements MqttCallback, MQTTSubscriberBase {
    private String brokerUrl = null;
    private String colon = ":";
    private String clientId = "demoClient2";

    private MqttClient client = null;
    private MqttConnectOptions options = null;
    private MemoryPersistence persistence = null;

    private static final Logger logger = LoggerFactory.getLogger(MQTTSubscriber.class);

    public MQTTSubscriber() {
        this.config();
    }

    @Override
    public void connectionLost(Throwable cause) {
        logger.info("Connection lost");
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String time = new Timestamp(System.currentTimeMillis()).toString();
        System.out.println();
        System.out.println("********************************************************");
        System.out.println("Message arrived at " + time + "Topic: " + topic + " Message: " + new String(message.getPayload()));
        System.out.println("********************************************************");
        System.out.println();
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // Not required for subscriber
    }


    @Override
    public void subscribeMessage(String topic) {
        try {
            this.client.subscribe(topic, this.qos);
        } catch (MqttException exception) {
            logger.error("ERROR", exception);
        }
    }

    @Override
    public void disconnect() {
        try {
            this.client.disconnect();
        } catch (MqttException exception) {
            logger.error("ERROR: ", exception);
        }
    }


    @Override
    public void config(String broker, Integer port, Boolean ssl, Boolean withUserNamePass) {
        String protocol = this.TCP;
        if (true == ssl) {
            protocol = this.SSL;
        }

        this.brokerUrl = protocol + broker + this.colon + port;
        this.persistence = new MemoryPersistence();
        this.options = new MqttConnectOptions();

        try {
            this.client = new MqttClient(this.brokerUrl, clientId, persistence);
            this.options.setCleanSession(true);
            if (true == withUserNamePass) {
                if (this.password != null) {
                    this.options.setPassword(this.password.toCharArray());
                }
                if (this.userName != null) {
                    this.options.setUserName(this.userName);
                }
            }
            this.client.connect(this.options);
            this.client.setCallback(this);
        }
        catch(MqttException exception) {
            this.logger.error("ERROR ", exception);
        }
    }

    @Override
    public void config() {
        this.brokerUrl = this.TCP + this.broker + this.colon + this.port;
        this.persistence = new MemoryPersistence();
        this.options = new MqttConnectOptions();

        try {
            this.client = new MqttClient(brokerUrl, clientId, persistence);
            this.options.setCleanSession(true);
            this.client.connect(options);
            this.client.setCallback(this);
        }
        catch(MqttException exception) {
            logger.error("ERROR", exception);
        }
    }

}

这些是我使用的依赖项：

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.2.RELEASE</version>
    <relativePath/>
  </parent>
  <groupId>im.sma.mqtt</groupId>
  <artifactId>mqtt-client</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <name>demo</name>
  <description>Demo project for Spring Boot</description>
  <properties>
    <java.version>1.8</java.version>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.springframework.integration</groupId>
      <artifactId>spring-integration-mqtt</artifactId>
      <version>5.1.2.RELEASE</version>
    </dependency>

  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
</project>

为什么我不能注射豆子，我应该采取什么行动？

Answer 1

问题是，您正在使用一个非常通用的接口作为自动连接，即Runnable。

因此，似乎存在两个与Runnable接口相匹配的bean：

1. 其中一个是你自己创建的MessageListener类
2. 另一个是DefaultHeaderChannelRegistry，作为一个名为integrationHeaderChannelRegistry的bean公开。它可能是公开的，因为类路径上有Spring集成。

问题是，由于这个原因，Spring IoC容器无法确定它应该注入什么bean，因此它提供了一些解决方案。

将其中一个bean标记为@Primary

这可以在99%的场景中使用其中一个bean的场景中使用。通过将MessageListener类标记为@Primary，它将在尝试注入它时获得优先权，例如：

@Primary // Add this
@Component
public class MessageListener implements Runnable {
    // ...
}

更新使用者以接受多个bean

当您希望引用所有Runnable bean时，此场景非常有用。在您的情况下，这可能不是解决方案，但在某些情况下，您可能希望获得特定类型的所有bean。要做到这一点，你可以：

@Autowired
private List<Runnable> runnables; // Change the type to List<Runnable>

使用@Qualifier来识别bean

另一种可能是使用@Qualifier来指定要注入的bean的确切名称。在您的情况下，您可以在messageListener或integrationHeaderChannelRegistry之间进行选择。例如：

@Autowired
@Qualifier("messageListener")
private Runnable mesageListener;

在您的情况下，这可能是提出的最佳解决方案。

还有一些额外的解决方案，我也想提供。

使用特定类型

如果将自动处理字段的类型更改为MessageListener，就不会混淆应该向哪个bean注入，因为只有一个MessageListener类型的bean

@Autowired
private MessageListener mesageListener;

不使用Spring集成库

到目前为止，您展示的代码与Spring集成无关。如果您的唯一目标是设置一个MQTT客户端，您可能需要考虑删除spring-integration-mqtt包，转而使用像日食泛美卫生组织这样的简单MQTT客户端实现。

由于添加了spring-integration-mqtt库，所以另一个bean是自动创建的，因此删除它将停止导致创建bean，这也将解决问题。

Answer 2

我的代码出现了一个星期的问题，今天我发现了这个问题，因为我给两个metods取了相同的名字：

   @Bean(name = "sqlServer")
public DataSource sqlServerDataSource() {
    DriverManagerDataSource dataSource = new DriverManagerDataSource();
    dataSource.setDriverClassName(env.getProperty("spring.ds-sql.driverClassName"));
    dataSource.setUrl(env.getProperty("spring.ds-sql.url"));
    dataSource.setUsername(env.getProperty("spring.ds-sql.username"));
    dataSource.setPassword(env.getProperty("spring.ds-sql.password"));
    return dataSource;
}

@Bean(name = "sqlJdbc")
public JdbcTemplate sqlServerDataSource(@Qualifier("sqlServer") DataSource dsSqlServer) {
    return new JdbcTemplate(dsSqlServer);
}

希望这对有同样情况的人有帮助。

祝好运