首页 / JAVA / java – 提供单个TaskExecutor / ThreadPool的多个DefaultMessageListenerContainer(公平)

java – 提供单个TaskExecutor / ThreadPool的多个DefaultMessageListenerContainer(公平)

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了java – 提供单个TaskExecutor / ThreadPool的多个DefaultMessageListenerContainer(公平)，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含11355字，纯文字阅读大概需要17分钟。

内容图文

java – 提供单个TaskExecutor / ThreadPool的多个DefaultMessageListenerContainer(公平)

我正在开发一个Spring Boot应用程序,我打算在其中使用来自多个队列的JMS消息 – 但是我想控制处理这些消息的执行线程的总数,而不是控制每个JMS队列执行的线程数.

简而言之：大量的JMS队列.一个线程池来进行处理.

有些队列比其他队列更繁忙(有些可能总是有工作,有些则长时间闲置) – 所以我想利用可用的处理能力做任何需要完成的工作,无论源队列如何.

我已经设置了一系列DefaultMessageListenerContainers,每个都使用具有固定池大小的共享TaskExecutor.我观察到的行为是,一个队列将占用所有可用的插槽 – 然后(即使第一个队列变空),插槽也不可用于其他队列的DefaultMessageListenerContainers使用.

这在javadocs for DefaultMessageListenerContainer.setTaskExecutor()中有详细说明：

A plain thread pool does not add much value, as this listener
container will occupy a number of threads for its entire lifetime.

>有办法解决这个问题吗？它如何在javadocs中提到的“J2EE环境”中以不同的方式工作？
>例如,我可以执行类似MessageListenerContainer的操作吗？
>如果我想要的是可能的 – 那么是否可以应用一些排序,从而可以给予先前空闲队列上收到的消息更高的优先级？

解决方法:

我试图重新创建你的问题,但我不能.可能是因为您在此问题中缺少源代码.但这就是我的尝试.

import java.io.File;

import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.adapter.MessageListenerAdapter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.FileSystemUtils;

@Configuration
@EnableAutoConfiguration
public class Application {

    public class Receiver {

        private String name;

        public Receiver(String name) {
            this.name = name;
        }

        /**
         * When you receive a message, print it out, then shut down the application.
         * Finally, clean up any ActiveMQ server stuff.
         */
        public void receiveMessage(String message) {
            System.out.println("Received <" + message + "> @ "+name);
            //context.close();
            FileSystemUtils.deleteRecursively(new File("activemq-data"));
        }
    }

    static String mailboxDestination = "mailbox-destination";

    static String mailboxDestination2= "mailbox-destination2";

    @Bean
    MessageListenerAdapter adapter1() {
        MessageListenerAdapter messageListener
                = new MessageListenerAdapter(new Receiver("MailBox1"));
        messageListener.setDefaultListenerMethod("receiveMessage");
        return messageListener;
    }

    @Bean
    MessageListenerAdapter adapter2() {
        MessageListenerAdapter messageListener
                = new MessageListenerAdapter(new Receiver("MailBox2"));
        messageListener.setDefaultListenerMethod("receiveMessage");
        return messageListener;
    }

    @Bean
    DefaultMessageListenerContainer container(ConnectionFactory connectionFactory) throws Exception {
        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
        container.setMessageListener(adapter1());
        container.setConnectionFactory(connectionFactory);
        container.setDestinationName(mailboxDestination);
        container.setTaskExecutor(taskExecutor());

        return container;
    }

    @Bean
    DefaultMessageListenerContainer container2(ConnectionFactory connectionFactory) throws Exception {
        DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
        container.setMessageListener(adapter2());
        container.setConnectionFactory(connectionFactory);
        container.setDestinationName(mailboxDestination2);
        container.setTaskExecutor(taskExecutor());

        return container;
    }

    @Bean
    TaskExecutor taskExecutor() throws Exception {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(10);
        taskExecutor.setMaxPoolSize(100);
        taskExecutor.setQueueCapacity(1000);
        taskExecutor.setThreadGroupName("MyThreads");

        return taskExecutor;
    }

    public static void main(String[] args) {
        // Clean out any ActiveMQ data from a previous run
        FileSystemUtils.deleteRecursively(new File("activemq-data"));

        // Launch the application
        ConfigurableApplicationContext context = SpringApplication.run(Application.class, args);

        // Send a message

        final JmsTemplate jmsTemplate = context.getBean(JmsTemplate.class);
        System.out.println("Sending a new message.");
        new Thread() {
            public void run() {
                for(int i=0;i<100;i++) {
                    final String message = (i+1)+" ping!";
                    MessageCreator messageCreator = new MessageCreator() {
                        @Override
                        public Message createMessage(Session session) throws JMSException {
                            return session.createTextMessage(message);
                        }
                    };
                    jmsTemplate.send(mailboxDestination, messageCreator);
                }
            }
        }.start();

        new Thread() {
            public void run() {
                for(int i=0;i<100;i++) {
                    final String message = (i+1)+" ping!";
                    MessageCreator messageCreator = new MessageCreator() {
                        @Override
                        public Message createMessage(Session session) throws JMSException {
                            return session.createTextMessage(message);
                        }
                    };
                    jmsTemplate.send(mailboxDestination2, messageCreator);
                }
            }
        }.start();

    }

}

我得到的输出是;

Received <1 ping!> @ MailBox2
Received <1 ping!> @ MailBox1
Received <2 ping!> @ MailBox2
Received <2 ping!> @ MailBox1
Received <3 ping!> @ MailBox2
Received <3 ping!> @ MailBox1
Received <4 ping!> @ MailBox2
Received <4 ping!> @ MailBox1
Received <5 ping!> @ MailBox2
Received <5 ping!> @ MailBox1
Received <6 ping!> @ MailBox2
Received <6 ping!> @ MailBox1
Received <7 ping!> @ MailBox2
Received <7 ping!> @ MailBox1
Received <8 ping!> @ MailBox2
Received <8 ping!> @ MailBox1
Received <9 ping!> @ MailBox2
Received <9 ping!> @ MailBox1
Received <10 ping!> @ MailBox2
Received <10 ping!> @ MailBox1
Received <11 ping!> @ MailBox2
Received <11 ping!> @ MailBox1
Received <12 ping!> @ MailBox1
Received <12 ping!> @ MailBox2
Received <13 ping!> @ MailBox1
Received <13 ping!> @ MailBox2
Received <14 ping!> @ MailBox1
Received <14 ping!> @ MailBox2
Received <15 ping!> @ MailBox2
Received <15 ping!> @ MailBox1
Received <16 ping!> @ MailBox2
Received <16 ping!> @ MailBox1
Received <17 ping!> @ MailBox2
Received <17 ping!> @ MailBox1
Received <18 ping!> @ MailBox2
Received <18 ping!> @ MailBox1
Received <19 ping!> @ MailBox1
Received <19 ping!> @ MailBox2
Received <20 ping!> @ MailBox1
Received <20 ping!> @ MailBox2
Received <21 ping!> @ MailBox1
Received <21 ping!> @ MailBox2
Received <22 ping!> @ MailBox1
Received <22 ping!> @ MailBox2
Received <23 ping!> @ MailBox1
Received <23 ping!> @ MailBox2
Received <24 ping!> @ MailBox1
Received <24 ping!> @ MailBox2
Received <25 ping!> @ MailBox1
Received <25 ping!> @ MailBox2
Received <26 ping!> @ MailBox1
Received <26 ping!> @ MailBox2
Received <27 ping!> @ MailBox1
Received <27 ping!> @ MailBox2
Received <28 ping!> @ MailBox2
Received <28 ping!> @ MailBox1
Received <29 ping!> @ MailBox2
Received <29 ping!> @ MailBox1
Received <30 ping!> @ MailBox2
Received <30 ping!> @ MailBox1
Received <31 ping!> @ MailBox2
Received <31 ping!> @ MailBox1
Received <32 ping!> @ MailBox2
Received <33 ping!> @ MailBox2
Received <32 ping!> @ MailBox1
Received <34 ping!> @ MailBox2
Received <33 ping!> @ MailBox1
Received <34 ping!> @ MailBox1
Received <35 ping!> @ MailBox2
Received <35 ping!> @ MailBox1
Received <36 ping!> @ MailBox2
Received <36 ping!> @ MailBox1
Received <37 ping!> @ MailBox2
Received <37 ping!> @ MailBox1
Received <38 ping!> @ MailBox2
Received <38 ping!> @ MailBox1
Received <39 ping!> @ MailBox2
Received <39 ping!> @ MailBox1
Received <40 ping!> @ MailBox2
Received <40 ping!> @ MailBox1
Received <41 ping!> @ MailBox2
Received <42 ping!> @ MailBox2
Received <43 ping!> @ MailBox2
Received <44 ping!> @ MailBox2
Received <45 ping!> @ MailBox2
Received <46 ping!> @ MailBox2
Received <47 ping!> @ MailBox2
Received <48 ping!> @ MailBox2
Received <49 ping!> @ MailBox2
Received <50 ping!> @ MailBox2
Received <51 ping!> @ MailBox2
Received <52 ping!> @ MailBox2
Received <53 ping!> @ MailBox2
Received <54 ping!> @ MailBox2
Received <55 ping!> @ MailBox2
Received <56 ping!> @ MailBox2
Received <57 ping!> @ MailBox2
Received <58 ping!> @ MailBox2
Received <59 ping!> @ MailBox2
Received <60 ping!> @ MailBox2
Received <61 ping!> @ MailBox2
Received <62 ping!> @ MailBox2
Received <63 ping!> @ MailBox2
Received <64 ping!> @ MailBox2
Received <65 ping!> @ MailBox2
Received <66 ping!> @ MailBox2
Received <67 ping!> @ MailBox2
Received <68 ping!> @ MailBox2
Received <69 ping!> @ MailBox2
Received <70 ping!> @ MailBox2
Received <71 ping!> @ MailBox2
Received <72 ping!> @ MailBox2
Received <73 ping!> @ MailBox2
Received <74 ping!> @ MailBox2
Received <75 ping!> @ MailBox2
Received <76 ping!> @ MailBox2
Received <77 ping!> @ MailBox2
Received <78 ping!> @ MailBox2
Received <79 ping!> @ MailBox2
Received <80 ping!> @ MailBox2
Received <81 ping!> @ MailBox2
Received <82 ping!> @ MailBox2
Received <83 ping!> @ MailBox2
Received <84 ping!> @ MailBox2
Received <85 ping!> @ MailBox2
Received <86 ping!> @ MailBox2
Received <87 ping!> @ MailBox2
Received <88 ping!> @ MailBox2
Received <89 ping!> @ MailBox2
Received <90 ping!> @ MailBox2
Received <91 ping!> @ MailBox2
Received <92 ping!> @ MailBox2
Received <93 ping!> @ MailBox2
Received <94 ping!> @ MailBox2
Received <95 ping!> @ MailBox2
Received <96 ping!> @ MailBox2
Received <97 ping!> @ MailBox2
Received <98 ping!> @ MailBox2
Received <99 ping!> @ MailBox2
Received <100 ping!> @ MailBox2
Received <41 ping!> @ MailBox1
Received <42 ping!> @ MailBox1
Received <43 ping!> @ MailBox1
Received <44 ping!> @ MailBox1
Received <45 ping!> @ MailBox1
Received <46 ping!> @ MailBox1
Received <47 ping!> @ MailBox1
Received <48 ping!> @ MailBox1
Received <49 ping!> @ MailBox1
Received <50 ping!> @ MailBox1
Received <51 ping!> @ MailBox1
Received <52 ping!> @ MailBox1
Received <53 ping!> @ MailBox1
Received <54 ping!> @ MailBox1
Received <55 ping!> @ MailBox1
Received <56 ping!> @ MailBox1
Received <57 ping!> @ MailBox1
Received <58 ping!> @ MailBox1
Received <59 ping!> @ MailBox1
Received <60 ping!> @ MailBox1
Received <61 ping!> @ MailBox1
Received <62 ping!> @ MailBox1
Received <63 ping!> @ MailBox1
Received <64 ping!> @ MailBox1
Received <65 ping!> @ MailBox1
Received <66 ping!> @ MailBox1
Received <67 ping!> @ MailBox1
Received <68 ping!> @ MailBox1
Received <69 ping!> @ MailBox1
Received <70 ping!> @ MailBox1
Received <71 ping!> @ MailBox1
Received <72 ping!> @ MailBox1
Received <73 ping!> @ MailBox1
Received <74 ping!> @ MailBox1
Received <75 ping!> @ MailBox1
Received <76 ping!> @ MailBox1
Received <77 ping!> @ MailBox1
Received <78 ping!> @ MailBox1
Received <79 ping!> @ MailBox1
Received <80 ping!> @ MailBox1
Received <81 ping!> @ MailBox1
Received <82 ping!> @ MailBox1
Received <83 ping!> @ MailBox1
Received <84 ping!> @ MailBox1
Received <85 ping!> @ MailBox1
Received <86 ping!> @ MailBox1
Received <87 ping!> @ MailBox1
Received <88 ping!> @ MailBox1
Received <89 ping!> @ MailBox1
Received <90 ping!> @ MailBox1
Received <91 ping!> @ MailBox1
Received <92 ping!> @ MailBox1
Received <93 ping!> @ MailBox1
Received <94 ping!> @ MailBox1
Received <95 ping!> @ MailBox1
Received <96 ping!> @ MailBox1
Received <97 ping!> @ MailBox1
Received <98 ping!> @ MailBox1
Received <99 ping!> @ MailBox1
Received <100 ping!> @ MailBox1

如果我错了,请纠正我但是使用这种配置我没有看到任何问题.两个队列和一个任务执行器.

内容总结

以上是互联网集市为您收集整理的java – 提供单个TaskExecutor / ThreadPool的多个DefaultMessageListenerContainer(公平)全部内容，希望文章能够帮你解决java – 提供单个TaskExecutor / ThreadPool的多个DefaultMessageListenerContainer(公平)所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/817391.html

来源：【匿名】

【上一篇】Java clone()和equals()检查【下一篇】JAVA/JSP学习系列之五

更多 ►

【java – 提供单个TaskExecutor / ThreadPool的多个DefaultMessageListenerContainer(公平)】教程文章相关的互联网学习教程文章

Java并发编程-ThreadPool线程池【代码】【图】

ThreadPool线程池 1.线程池的优势1.1.引言1.2.为什么要使用线程池 2.线程池的使用2.1.架构说明2.2.线程池的三大方法2.2.1.newFixedThreadPool(int)方法2.2.2.newSingleThreadExector2.2.3.newCachedThreadPool3.ThreadPoolExecutor底层原理4.线程池7大重要参数 1.线程池的优势 1.1.引言与数据库线程池类似，如果没有数据库连接池，那么每次对数据库的连接池都要new来获取连接池。重复的连接和释放操作会消费大量的系统资源，我们可...

Java 线程知识笔记 (四) Executor与ThreadPool 其二【代码】【图】

前言上一篇【Java 线程知识笔记 (三)】我们从源码里面探究了一个线程在线程池中是如何执行的，由于线程池有一个很大的框架，仅仅通过这点东西就说明白线程池，尤其是对于初学者来说更是难以理解。因此这篇将会对源码讲解中遗漏的一些内容，比如线程池的状态，以及线程池中线程是在哪里创建的等等做一个补漏，同时对线程池的基本内容进行一个概念性的阐述。线程池的优点上一篇中对线程池和单线程的效率做了一个对比的例子，同时开...

[Java] Java核心（2）深入理解线程池ThreadPool【代码】【图】

本文你将获得以下信息：● 线程池源码解读　　　　● 线程池执行流程分析　　　　● 带返回值的线程池实现　　　　● 延迟线程池实现为了方便读者理解，本文会由浅入深，先从线程池的使用开始再延伸到源码解读和源码分析等高级内容，读者可根据自己的情况自主选择阅读顺序和需要了解的章节。 1 线程池优点线程池能够更加充分的利用CPU、内存、网络、IO等系统资源，线程池的主要作用如下：● 利用线程池可以复用线程，控制最大并发数...

java-ThreadPool中任务的自定义顺序【代码】

我目前使用FixedThreadPool从Web下载图像,如下所示：ExecutorService mThreadPool = Executors.newFixedThreadPool(MAX_THREADS);然后,我只提交带有图像URL的新Runnables,它要么从URL下载图像,要么如果它存在于缓存中,则从那里加载它. 我希望能够确保一次只有一个线程可以处理特定的URL(以防止图像被下载MAX_THREADS次的情况),并且如果该线程线程完成并下载了该图像,我想允许下一个(或全部)相同的网址)运行,以从我的缓存加载以前下...

java-ThreadPool执行程序和大量客户端同时连接【代码】

我正在使用套接字在Java上构建客户端服务器应用程序.据我了解,为每个要连接的客户端创建线程太昂贵了.相反,我们可以使用ThreadPool Executor.如并发的documentation中所述,我们可以创建一个固定大小的线程池.class NetworkService implements Runnable {private final ServerSocket serverSocket;private final ExecutorService pool;public NetworkService(int port, int poolSize)throws IOException {serverSocket = new Server...

java-Threadpool Executor完成后如何给出消息？【代码】

当我的ThreadpoolExecutor执行完毕时,我试图给出一个弹出警报消息.它正在从网站搜索电子邮件地址,完成后,我希望收到一条警报消息“已完成”.这是我的主题：public class Constant{public static final int NUM_OF_THREAD = 60;public static final int TIME_OUT = 10000;}ThreadPoolExecutor poolMainExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(Constant.NUM_OF_THREAD);这是我的搜索操作类：-class Searchi...

可用的Java信号量允许使用ThreadPool【代码】

我正在使用教程https://www.baeldung.com/java-semaphore在Java中学习信号量. 教程中的第一个测试(下一个)运行良好.@Test public void givenLoginQueue_whenReachLimit_thenBlocked() { ????int slots = 10; ????ExecutorService executorService = Executors.newFixedThreadPool(slots); ????LoginQueueUsingSemaphore loginQueue = new LoginQueueUsingSemaphore(slots); ????IntStream.range(0, slots) ??????.forEach(user -> ...

java – threadpool类未指定错误【代码】

在获取调度程序对象后,在获得初始化的StdSchedulerFactory后出现错误. init方法失败;嵌套异常是org.quartz.SchedulerException：未指定ThreadPool类. 代码是：Properties properties = new Properties();properties.put("org.quartz.scheduler.instanceName", name);properties.put("org.quartz.threadPool.threadCount", String.valueOf(numberOfThreads));properties.put("org.quartz.scheduler.skipUpdateCheck", "true");StdSc...

java – 在Executor服务中的ThreadPool中添加线程【代码】

我正在进行多线程程序,我正在尝试确保每个线程运行30分钟.假设我们有10个线程,那么10个线程中的每个线程应运行30分钟. 以下是我的代码 – class ThreadTask implements Runnable {private final long endTime;public ThreadTask(long endTime) {this.endTime = endTime;}@Overridepublic void run() {while (System.currentTimeMillis() <= endTime) {// do something meaningful} } }public class TestPool {public static void...

java – 使用JMS或ThreadPool发送电子邮件

我想知道：我有一个场景.如果用户将产品添加到系统(我正在开发),则会有一个监听器向用户的客户端发送通知,通知用户添加的新产品. 我已经读过这个thread了(看到我之前从未使用过JMS和ThreadPool)我想知道是否应该使用JMS或ThreadPooling. 我使用Tomcat 5.5及更高版本和JBoss 5及更高版本(取决于公司的最后手段)来部署我的Web应用程序. 如果我使用JMS,我是否使用Apache ActiveMQ或JBoss Messaging？它们是否兼容在两个平台上运行(To...

Java等待线程(Threadpool)完成工作并开始另一项工作【代码】

所以我创建了5个线程,在完成工作后我想做另一个工作.那么如何找出执行者的线程何时完成他们的工作,然后才开始超级工作方法呢？主要：ExecutorService executor = Executors.newFixedThreadPool(5);CountDownLatch doneSignal = new CountDownLatch(5);//thread numberfor(int i=0;i<5;i++){getLogFile n = new getLogFile(doneSignal, i);// work methodexecutor.execute(n);doneSignal.await(); }//可能这里有类似executor.awai...

java – 未指定任何内容时使用的默认“org.quartz.threadPool.threadCount”是什么？【代码】

我在Spring应用程序中有一些计划任务,它们的配置如下：<bean id="scheduler" class="org.springframework.scheduling.quartz.SchedulerFactoryBean"><property name="triggers"><list> </list></property> </bean>我遇到了一些问题(某些任务在应用但不总是在很长一段时间后或在某些时间运行时都没有运行)我认为这可能是因为有很多任务(目前为止11个)且系统可以’同时运行它们.我想...