阻塞队列中的安全中断

简介: 阻塞队列中的安全中断

本文的小例子主要是说明如何安全中断阻塞队列中的任务,避免使用interrupt()中断线程,造成堵塞队列中没有被消费的任务都被忽略。

package com.thread;
import java.io.PrintWriter;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
 * 单个线程的安全取消
 * 
 * 这里模拟的场景是单个线程,处理阻塞队列中要处理的日志信息
 * 
 * 分析:由于使用了阻塞队列,而阻塞队列的take操作响应中断;如果日志线程被修改为捕获InterruptedException就退出,那么
 * 中断日志就能够停止了。
 * 但是,这样使用Interrupt来中断线程,突然的关闭会忽略等待中需要被记录的日志。
 * 
 * 解决方案:添加“已请求关闭”标志,和队列中要处理的日志的数量。
 * 当发起Interrupt来中断线程时,并不会直接中断线程,只是改变“已请求关闭”标志为ture,
 * 只有当标志为ture并且队列中已请求进来的要处理的日志都处理完了才会中断线程。
 * 
 * @author hadoop
 *
 */
public class SafeCancellThread {
       class  LogService{
        private final BlockingQueue<String>  queue = new ArrayBlockingQueue<String>(50);
        private final LoggerThread loggerThread = new LoggerThread();
        private  PrintWriter writer ; 
        private boolean isShutdown;
        private int reservations;
     public LogService(PrintWriter writer) {
       this.writer = writer;
     }
  public void start(){
         loggerThread.start();
        }
  public void stop(){
     synchronized (this) {
     isShutdown = true;
    }
     loggerThread.interrupt();
     }
        public void log(String msg) throws InterruptedException{
         synchronized (this) {
      if(isShutdown){
        throw new IllegalStateException();
      }
      ++reservations;
      }
         System.out.println("向队列中添加:"+msg);
         queue.put(msg);//向阻塞队列中添加任务
        }
        class LoggerThread extends Thread{
         public void run(){
        try {
          while(true){
           try {
            synchronized(LogService.this){
            if(isShutdown && reservations == 0){
              System.out.println("队列中的任务处理完成");
              break;
            }
            }
            //消费阻塞队列
            String msg = queue.take();
            synchronized(LogService.this){
             --reservations;
            }
            Thread.currentThread().sleep(100);
            System.out.println("处理"+msg);
            writer.println(msg);
          } catch (InterruptedException e) {
            System.out.println("接到中断请求,重试知道队列中的任务消费完");
          }
          }
        } finally{
          writer.close();
        }
         }
        } 
       }
  public static void main(String[] args) throws Exception {
   SafeCancellThread sct = new SafeCancellThread();
   PrintWriter writer = new PrintWriter("C:\\Users\\Administrator\\Desktop\\input\\temp.txt");
   writer.write("开始");
   final LogService logService = sct.new LogService(writer);
   new Thread(new Runnable() {
    @Override
    public void run() {
           int incre = 1;
      while(!logService.isShutdown){//如果服务停止,就不继续生产日志任务了。默认为false
        try {
        Thread.sleep(2*100);//每200ms产生一条日志写入任务
        logService.log("日志"+incre);//添加日志时会出现堵塞
        incre++;
        } catch (Exception e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        }
      }
    }
  }).start();
   Thread.sleep(6*1000);
   System.out.println("开始处理任务");
   logService.start();//开始处理
   //等待4s后执行取消任务的请求
   Thread.sleep(4*1000);
   System.out.println("发起取消请求");
   logService.stop();
  }
}


执行结果:


可以发现,调用了中断请求,停止了向阻塞队列中继续加入任务,并且没有立刻停止线程,而是继续将已添加到队列中的任务执行完,才停止线程。避免了加入阻塞队列中的任务由于线程中断而被忽略。

1.png


还可以使用基于服务的消息中断,管理消息的生命周期

package com.thread;
import java.io.FileNotFoundException;
import java.io.PrintWriter;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 使用Executor的日志服务
 * @author hadoop
 *
 */
public class LogService {
    private final ExecutorService exec = Executors.newSingleThreadExecutor();
    private static final int TIMEOUT = 5;
    private static final TimeUnit UNIT = TimeUnit.SECONDS;
    private final PrintWriter writer ;  
    public void start(){
    }
    public LogService(PrintWriter writer) {
  this.writer = writer;
  }
  public void stop() throws InterruptedException{
      try {
      System.out.println("停止");
      exec.shutdown();//会等待已提交到线程池中的任务执行完毕后,在中断线程
          exec.awaitTermination(TIMEOUT, UNIT);//
  }finally{
    writer.close();
  }
    }
  class WriteTask implements Runnable{
     private String msg;
  public WriteTask(String msg) {
    this.msg = msg;
  }
  @Override
  public void run() {
    try {
    Thread.sleep(1000);
    } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    }
    System.out.println("处理"+msg);
    writer.write(msg);
  }
  }
    public void log(String msg){
      try {
       exec.execute(new WriteTask(msg));
  } catch (Exception e) {
    e.printStackTrace();
  }
    }
    public static void main(String[] args) throws FileNotFoundException {
      PrintWriter writer = new PrintWriter("C:\\Users\\Administrator\\Desktop\\input\\temp.txt");
      LogService logService = new LogService(writer);
       for (int i = 0; i < 100; i++) {
        System.out.println("日志" + i);
        try {
    Thread.sleep(300);
    } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    }
        logService.log("日志"+i);
   }
       try {
    logService.stop();
  } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    System.out.println("线程任务被中断");
  }
  }
}
相关实践学习
【涂鸦即艺术】基于云应用开发平台CAP部署AI实时生图绘板
【涂鸦即艺术】基于云应用开发平台CAP部署AI实时生图绘板
目录
相关文章
|
消息中间件 存储 负载均衡
拆解一下消息队列、任务队列、任务调度系统
拆解一下消息队列、任务队列、任务调度系统
1734 0
|
机器学习/深度学习 前端开发 数据可视化
Kimi K2 开源发布:擅长代码与 Agentic 任务!
今天,月之暗面正式发布 Kimi K2 模型,并同步开源。Kimi K2 是一款具备更强代码能力、更擅长通用 Agent 任务的 MoE 架构基础模型,总参数 1T,激活参数 32B。
2020 0
|
10月前
|
数据采集 API 开发工具
CNFANS模式淘宝1688代购系统搭建指南
CNFANS模式整合国内电商资源,对接淘宝、1688商品库,为海外用户提供代购、集运、物流清关等一站式服务。通过API打通电商平台、支付(PayPal/Stripe)、国际物流及仓储系统,实现商品采集、下单、支付、发货全流程自动化,解决海外用户“买不到、价格高”难题,提升跨境购物体验。(238字)
|
10月前
|
机器学习/深度学习 存储 传感器
面向课堂与自习场景的智能坐姿识别系统——从行为感知到可视化部署的完整工程【YOLOv8】
本系统基于YOLOv8实现课堂与自习场景下的智能坐姿识别,支持标准坐姿与驼背等不良姿态的实时检测。结合PyQt5开发图形界面,提供图片、视频、摄像头多模式输入,具备高精度、低延迟、易部署等特点,适用于智慧校园、健康办公等场景,助力行为规范与健康管理。
725 5
面向课堂与自习场景的智能坐姿识别系统——从行为感知到可视化部署的完整工程【YOLOv8】
|
10月前
|
自然语言处理 安全 Java
Spring Boot中集成Lucence
Lucene是Apache开源的全文检索库,通过分词、建立索引实现高效搜索。本文介绍其原理,并在Spring Boot中集成,实现中文分词、高亮显示等实战功能,适用于文件搜索、内容检索等场景。
|
运维 安全 数据可视化
如何开发一套设备管理系统?(附架构图+流程图+代码参考)
随着企业对生产效率和设备安全的重视,设备管理系统成为企业运营的重要工具。该系统可实现设备的全生命周期管理,涵盖点检、巡检、报修、维修、保养及备件管理等功能,提升设备使用率,降低运营成本。通过自动化管理与实时数据记录,系统为企业提供高效、可靠的管理支持。本文详解系统开发流程,并提供代码示例,帮助理解实现过程。
|
分布式计算 关系型数据库 MySQL
Flink CDC 3.3.0 发布公告
Flink CDC 3.3.0 发布公告
769 14
|
安全 PHP Apache
【ThinkPHP框架教程·Part-02】开发规范和目录结构
本章节介绍ThinkPHP6.0的编码开发规范及目录结构。开发规范遵循PSR-2和PSR-4标准,涵盖命名、文件结构等;目录结构支持单应用和多应用模式,默认仅public目录对外访问,确保安全性。单应用模式下,app目录包含控制器、模型等核心组件;多应用模式允许多个app_name共存,适应复杂项目需求。配置文件与运行时文件分离,确保高效与安全。
【ThinkPHP框架教程·Part-02】开发规范和目录结构
|
传感器 监控 安全
智慧工地云平台的技术架构解析:微服务+Spring Cloud如何支撑海量数据?
慧工地解决方案依托AI、物联网和BIM技术,实现对施工现场的全方位、立体化管理。通过规范施工、减少安全隐患、节省人力、降低运营成本,提升工地管理的安全性、效率和精益度。该方案适用于大型建筑、基础设施、房地产开发等场景,具备微服务架构、大数据与AI分析、物联网设备联网、多端协同等创新点,推动建筑行业向数字化、智能化转型。未来将融合5G、区块链等技术,助力智慧城市建设。
1036 1

热门文章

最新文章