多线程使用Executor和WebSocket

Question

在我的例子中，我想创建多个线程执行器来处理收到的陷阱。在同一个应用程序中，我想实现websocket使我的应用程序实时运行。

我有一个配置类来创建ThreadPoolExecutor，像这样

@Configuration
@EnableAsync
@EnableScheduling
@Profile("!" + Constants.SPRING_PROFILE_FAST)
public class AsyncConfiguration implements AsyncConfigurer, EnvironmentAware {
 .......
 @Override
    @Bean
    public Executor getAsyncExecutor() {
        log.debug("Creating Async Task Executor");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(propertyResolver.getProperty("corePoolSize", Integer.class, 30));
        executor.setMaxPoolSize(propertyResolver.getProperty("maxPoolSize", Integer.class, 150));
        executor.setQueueCapacity(propertyResolver.getProperty("queueCapacity", Integer.class, 10000));
        executor.setThreadNamePrefix("ems-Executor-");
        return new ExceptionHandlingAsyncTaskExecutor(executor);
    }

然后我在TrapReceiver类中使用执行器，

@Component
public class TrapReceiver extends Thread implements CommandResponder {
    @Inject
    private ApplicationContext applicationContext;
    @Inject
    private Executor executor;
    public TrapReceiver(){
    }
    List<PDUv1> listPdu = new ArrayList<PDUv1>();
    String message = "";
    long totReceivedTrap = 0;
    @PostConstruct
    public void init() {
        //create thread pool untuk memanage thread puller (thread yang pull dan save device oid value)
        System.out.println("Running trap listener");
        this.start();
    }
    public synchronized void processPdu(CommandResponderEvent cmdRespEvent) {
        PDUv1 pdu = (PDUv1) cmdRespEvent.getPDU();
        listPdu.add(pdu);
        if (pdu != null) {
            if(listPdu.size() == 3){ //3trap per thread
                List<PDUv1> temp = new ArrayList<PDUv1>();
                temp.addAll(listPdu);
                TrapInsertor trapInsertor = (TrapInsertor) applicationContext.getBean("trapInsertor");
                trapInsertor.setProperty(temp);
                executor.execute(trapInsertor);
                listPdu.clear();
            }
        }
        totReceivedTrap++;
        if(totReceivedTrap % 10000 == 0)
            System.out.println("total received trap "+totReceivedTrap);
    }
    public void run() {
        while (true) {
            try {
                this.listen(new UdpAddress(getIp()+"/162")); //where to listen
            } catch (Exception e) {
                e.printStackTrace();
            } 
        }
    }

这段代码运行良好，但是当我试图在我的应用程序中添加websocket功能时，应用程序得到了错误。当我使用@EnableWebSocketMessageBroker注释时出现错误。这是我的websocket配置

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketAppConfig extends AbstractWebSocketMessageBrokerConfigurer {
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/topic");
        config.setApplicationDestinationPrefixes("/app");
    }

这是错误

Caused by: org.springframework.beans.factory.BeanCreationException: Could not autowire field: private java.util.concurrent.Executor com.satunol.ems.snmp.TrapReceiver.executor; nested exception is org.springframework.beans.factory.NoUniqueBeanDefinitionException: No qualifying bean of type [java.util.concurrent.Executor] is defined: expected single matching bean but found 5: getAsyncExecutor,messageBrokerSockJsTaskScheduler,clientInboundChannelExecutor,clientOutboundChannelExecutor,brokerChannelExecutor
    at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor$AutowiredFieldElement.inject(AutowiredAnnotationBeanPostProcessor.java:555)
    at org.springframework.beans.factory.annotation.InjectionMetadata.inject(InjectionMetadata.java:87)
    at org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor.postProcessPropertyValues(AutowiredAnnotationBeanPostProcessor.java:331)
    ... 16 more
Caused by: org.springframework.beans.factory.NoUniqueBeanDefinitionException: No qualifying bean of type [java.util.concurrent.Executor] is defined: expected single matching bean but found 5: getAsyncExecutor,messageBrokerSockJsTaskScheduler,clientInboundChannelExecutor

我的代码有什么问题，如果我在websocket配置或线程池中出错，它应该如何。

Answer 1

这里的问题是由于[java.util.concurrent]有多个可能的bean。执行器]，Spring无法选择正确的。弹簧自动连接机制默认是基于类型的。因为WebSocket有自己的ThreadPoolTaskExecutor实现，所以你最终得到了5个可能的bean。

自动装配机制是基于这样一个假设:您将提供一个匹配bean，而不需要任何其他注释。由于有多个匹配的bean，因此需要告诉Spring要自动装配的bean。您可以使用注释@Qualifier("beanName")

来实现它

@Autowired
@Qualifier("getAsyncExecutor")
private Executor executor;

希望有帮助!

一些例子:@Qualifier注释

Answer 2

这既不是多线程的问题，也不是web-socket的问题，Spring和它的自动连接功能的问题。查看错误-

由:org.springframework.beans.factory.BeanCreationException引起:无法自动连接字段: private java.util.concurrent.Executorcom.satunol.ems.snmp.TrapReceiver.executor;嵌套异常是org.springframework.beans.factory.NoUniqueBeanDefinitionException: 类型 [java.util.concurrent.]Executor]定义为:期望有一个匹配的bean，但发现了5个:getAsyncExecutor、messageBrokerSockJsTaskScheduler clientInboundChannelExecutor、clientOutboundChannelExecutor brokerChannelExecutor

我的建议:删除threadpool和threadpoolexecutor标签，并添加spring和spring-annotations标签，一些spring人员应该能够快速帮助您。