多消费者(多线程)对MNS的使用

简介: 多消费者(多线程)对MNS的使用 背景 在阿里云MNS消费者的使用中,阿里云提供了使用 [消息服务 > 最佳实践 > 长轮询](https://help.aliyun.com/document_detail/34478.html?spm=a2c4g.11174283.3.3.AqOdUy#h2-u89E3u51B3u65B9u6848) 的代码和说明,在解决方案中阿里云这么说道 在开了上百个线程同时访问的情况下,如果队列里已经没有消息了,那么其实不需要上百个线程都同时挂LongPolling。

多消费者(多线程)对MNS的使用

背景

在阿里云MNS消费者的使用中,阿里云提供了使用 消息服务-最佳实践-长轮询
的代码和说明,在解决方案中阿里云这么说道

在开了上百个线程同时访问的情况下,如果队列里已经没有消息了,那么其实不需要上百个线程都同时挂LongPolling。只需要有1-N个线程挂LongPolling就足够了。挂LongPolling的线程在发现队列里有消息时,可以唤醒其他线程一起来取消息以达到快速响应的目的
Receiver内部做了LongPolling的排他机制,只要有一个线程在做LongPolling,那么其他线程只需要Wait就可以了。 —— [解决方案]

但是如何启动1-N个线程,同时产生多个消费者,并没有给出说明,阿里云官方提供的demo中是使用在main方法中启用:

CloudAccount account = new CloudAccount("ACCESS_ID", "ACCESS_KEY", "ENDPOINT");
        sMNSClient = account.getMNSClient();
        sMNSClient.getQueueRef("TestQueue").delete();
        
        sMNSClient.getQueueRef("TestQueue").create();

        Thread thread1 = new Thread(new Runnable() {
            public void run() {
                WorkerFunc(1);
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            public void run() {
                WorkerFunc(2);
            }
        });
        Thread thread3 = new Thread(new Runnable() {
            public void run() {
                WorkerFunc(3);
            }
        }); 

这里我提供一种比较好的方法,可以利用spring IOC容器的依赖注入,来管理和启动多个消费者(多线程)。

方法展示

Spring会通过依赖注入的方式,来管理关联对象的生命周期,所以我们可以将消费者的产生管理,都由Spring IOC容器代劳,也就是说,我把消费者创建的控制权都交给Spring容器。方法如下

@Component
public class NormalProcessComponent {
    
     private static Logger log = LoggerFactory.getLogger(NormalProcessComponent.class);
    private static ThreadPoolExecutor threadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(30);
    
    public NormalProcessComponent(){
         for(int i = 0; i < 50; i++){
             threadPool.execute(new Runnable() {
                
                @Override
                public void run() {
                    try {
                        process();
                    } catch (Exception e) {
                        
                        e.printStackTrace();
                    }
                    
                }
            });
         }

    }
    
     public void process() throws Exception {
              //使用阿里云官方提供的方法
            MessageReceiver receiver = new MessageReceiver(workerId, sMNSClient, "TestQueue");
            while (true) {
                Message message = receiver.receiveMessage();
                  try {
                    //取出Que中的信息
                     result = message .getMessageBodyAsString(); 
                     JSONObject params = JSONObject.parseObject(result);
                    if(params!=null){
                       //处理数据的方法                        
                    } else {
                        log.info("取出的数据为空!");
                         Thread.sleep(Constant.SLEEP_SECONDS);
                    }
                    
                 
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("fail to sleep"+message);
                    break;
                }
            }
        }
}

我们将消费者的产生方法,在类中的构造函数中定义,使用一个固定大小的线程池,来管理消费者(线程),同时加上Component注解,在项目启动时,Spring 的就会实例化这个类,注入到容器中,这个时候构造方法中的,多个消费者就会启动开始工作。

拓展

  • 阿里云官方MessageReceiver的解析 :长连接轮询,以及死锁和线程安全性问题的避免
  • 消费者的监控 :观察消费者的数量,避免消费者全部死亡,造成队列积压;

阿里云官方MessageReceiver的解析

在阿里云MNS消费者的使用中,阿里云提供了使用 消息服务-最佳实践-长轮询 ,官方已经提供了源代码和详细说明,我在这里就不贴代码了,主要说明其中的原理。
avatar

在MessageReceiver中,官方定义了一个

static final Map<String, Object> sLockObjMap

从而保证了,无论new出多少个MessageReceiver,都是从同一个Map,取出的lockObj。在使用lockObj中,均使用同步锁synchronized,从而实现了LongPolling的排他机制,只有一个线程在做LongPolling,其他线程都会Wait。避免了上百个线程同时访问MNS Server,一个Group只会产生,一条长连接进行长轮询。
可以将图中的Group,比作一台台服务器,而里面的多个Consumer,实际就是启动的多个消费线程。

消费者的监控

在上面代码中,使用了一个固定大小的线程池来管理多个线程(消费者),但是一旦子线程死亡,这个线程(消费者),并不会重启,这种情况就会产生队列积压。产生线程死亡一定是不正常,程序中的Bug存在。比如,有异常没有捕获到,或者在子线程中将异常throw出,就会使当前子线程死亡掉。这种情况一定是会有的,因为没有人写出的代码是完美无缺的,程序员只能尽可能避免bug的产生,所以我们需要用完善的日志和监控来完善,我们的项目。

这里我们可以利用监控线程池中的存活线程数量从而来,进行报警。

      //当消费者低于一定阈值触发报警
       if(threadPool.getActiveCount()<threshold){
           //报警       
        }

可以将这个封装成一个API接口,通过监控这个API来进行报警。

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
8月前
|
数据处理
多线程与并发编程【线程对象锁、死锁及解决方案、线程并发协作、生产者与消费者模式】(四)-全面详解(学习总结---从入门到深化)
多线程与并发编程【线程对象锁、死锁及解决方案、线程并发协作、生产者与消费者模式】(四)-全面详解(学习总结---从入门到深化)
75 1
|
10天前
|
Java
Java—多线程实现生产消费者
本文介绍了多线程实现生产消费者模式的三个版本。Version1包含四个类:`Producer`(生产者)、`Consumer`(消费者)、`Resource`(公共资源)和`TestMain`(测试类)。通过`synchronized`和`wait/notify`机制控制线程同步,但存在多个生产者或消费者时可能出现多次生产和消费的问题。 Version2将`if`改为`while`,解决了多次生产和消费的问题,但仍可能因`notify()`随机唤醒线程而导致死锁。因此,引入了`notifyAll()`来唤醒所有等待线程,但这会带来性能问题。
Java—多线程实现生产消费者
|
3月前
|
安全 Java
Java多线程通信新解:本文通过生产者-消费者模型案例,深入解析wait()、notify()、notifyAll()方法的实用技巧
【10月更文挑战第20天】Java多线程通信新解:本文通过生产者-消费者模型案例,深入解析wait()、notify()、notifyAll()方法的实用技巧,包括避免在循环外调用wait()、优先使用notifyAll()、确保线程安全及处理InterruptedException等,帮助读者更好地掌握这些方法的应用。
26 1
|
8月前
|
存储 安全 Java
Qt线程池+生产者消费者模型
Qt线程池+生产者消费者模型
330 5
|
3月前
|
消息中间件 NoSQL 关系型数据库
【多线程-从零开始-捌】阻塞队列,消费者生产者模型
【多线程-从零开始-捌】阻塞队列,消费者生产者模型
36 0
|
5月前
|
算法 Java
JUC(1)线程和进程、并发和并行、线程的状态、lock锁、生产者和消费者问题
该博客文章综合介绍了Java并发编程的基础知识,包括线程与进程的区别、并发与并行的概念、线程的生命周期状态、`sleep`与`wait`方法的差异、`Lock`接口及其实现类与`synchronized`关键字的对比,以及生产者和消费者问题的解决方案和使用`Condition`对象替代`synchronized`关键字的方法。
JUC(1)线程和进程、并发和并行、线程的状态、lock锁、生产者和消费者问题
|
5月前
|
消息中间件 设计模式 安全
多线程魔法:揭秘一个JVM中如何同时运行多个消费者
【8月更文挑战第22天】在Java虚拟机(JVM)中探索多消费者模式,此模式解耦生产与消费过程,提升系统性能。通过`ExecutorService`和`BlockingQueue`构建含2个生产者及4个消费者的系统,实现实时消息处理。多消费者模式虽增强处理能力,但也引入线程安全与资源竞争等挑战,需谨慎设计以确保高效稳定运行。
105 2
|
5月前
|
安全 Java
Java模拟生产者-消费者问题。生产者不断的往仓库中存放产品,消费者从仓库中消费产品。其中生产者和消费者都可以有若干个。在这里,生产者是一个线程,消费者是一个线程。仓库容量有限,只有库满时生产者不能存
该博客文章通过Java代码示例演示了生产者-消费者问题,其中生产者在仓库未满时生产产品,消费者在仓库有产品时消费产品,通过同步机制确保多线程环境下的线程安全和有效通信。
|
6月前
|
设计模式 安全 NoSQL
Java面试题:设计一个线程安全的单例模式,并解释其内存占用和垃圾回收机制;使用生产者消费者模式实现一个并发安全的队列;设计一个支持高并发的分布式锁
Java面试题:设计一个线程安全的单例模式,并解释其内存占用和垃圾回收机制;使用生产者消费者模式实现一个并发安全的队列;设计一个支持高并发的分布式锁
77 0
|
6月前
|
存储 设计模式 监控
Java面试题:如何在不牺牲性能的前提下,实现一个线程安全的单例模式?如何在生产者-消费者模式中平衡生产和消费的速度?Java内存模型规定了变量在内存中的存储和线程间的交互规则
Java面试题:如何在不牺牲性能的前提下,实现一个线程安全的单例模式?如何在生产者-消费者模式中平衡生产和消费的速度?Java内存模型规定了变量在内存中的存储和线程间的交互规则
55 0