java - 如何从监听器委托(delegate)可运行的代码？

Question

我是java执行程序包的新手。我想将创建线程的责任委托(delegate)给 Spring Rabbitmq(amqp) 中的另一个任务(类)。目前，我正在异步 MesageListener 的 onMessage() 方法中创建内部级别的可运行类，以在rabbitmq主题交换队列上有多个消息时实现并行性。该功能对我来说效果很好。我想将可运行的代码片段与 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

请理解，当这样进行切换时，消息将立即被确认，并且会在系统崩溃时丢失。

通常最好简单地增加监听器容器中的并发性并让容器管理线程。