如何从侦听器委托可运行代码？

Question

我是 java 的执行程序包的新手。我想将创建线程的责任委托给来自 spring rabbitmq(amqp) 的另一个任务 (class)。目前，我正在异步消息监听器的 onMessage() 方法中创建内部级别运行nable class 以在 rabbitmq 主题交换队列中有多个消息时实现并行性。该功能对我来说很好用。我想将运行nable 代码片段从 onMessage(Message message) 方法中分离出来。以下是代码。

package com.xyz.forum.event.listener;

import com.xyz.forum.constant.ClassType;
import com.xyz.forum.domain.dto.impl.AnswerDto;
import com.xyz.forum.domain.dto.impl.PollDto;
import com.xyz.forum.domain.dto.impl.QuestionDto;
import com.xyz.forum.manager.PumpManager;
import com.xyz.forum.utils.XyzForumConsumerUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;

import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * Created by bhupati on 11/3/16.
 */
public class XyzForumListener implements MessageListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(XyzForumListener.class);
    private static final String CLASS_HEADER = "__TypeId__";

    private Executor threadPool = Executors.newFixedThreadPool(10);

    @Autowired @Qualifier("pollPump") PumpManager pollPumpManager;
    @Autowired @Qualifier("questionPump") PumpManager questionPumpManager;
    @Autowired @Qualifier("answerPump") PumpManager answerPumpManager;

    @Override
    public void onMessage(final Message message) {

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                handleMessage(message);
            }
        };
        threadPool.execute(runnable);
    }

    private void handleMessage(Message message) {
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        String classType = (String) headers.get(CLASS_HEADER);
        LOGGER.info("Got the message: " + classType);

        switch (classType) {
            case ClassType.QUESTION:
                QuestionDto questionDto = XyzForumConsumerUtils.parseDto(message.getBody(), QuestionDto.class);
                LOGGER.info("questionDto : " + questionDto.getBody());
                questionPumpManager.executePumpService(questionDto, headers);
                break;

            case ClassType.POLL:
                PollDto pollDto = XyzForumConsumerUtils.parseDto(message.getBody(), PollDto.class);
                LOGGER.info("pollDto : " + pollDto.getBody());
                pollPumpManager.executePumpService(pollDto, headers);
                break;

            case ClassType.ANSWER:
                AnswerDto answerDto = XyzForumConsumerUtils.parseDto(message.getBody(), AnswerDto.class);
                LOGGER.info("answerDto : " + answerDto.getBody());
                answerPumpManager.executePumpService(answerDto, headers);

            default: LOGGER.warn("Unknown Type");
                break;

        }

    }

}

一般来说，运行一个监听器的方法同时使用 java.util.concurrent 包有一个标准模式。

Answer 1

请注意，当这样传递消息时，消息将立即被确认并在系统崩溃时丢失。

一般情况下，简单地增加侦听器容器中的并发性，让容器管理线程会更好。

如何从侦听器委托可运行代码？

How to delegate runnable code from listener?

java

multithreading

executorservice

java.util.concurrent

spring-amqp