首页 > 代码库 > 18.并发类容器MQ

18.并发类容器MQ

2024-11-21 00:41:01   202人阅读


  1. package demo7.MQ;
  3. public class QueueData {
  5.     private  int id;
  6.     private  String name;
  7.     private  String taskCode;
  9.     public QueueData() {
  10.     }
  12.     public QueueData(int id, String name, String taskCode) {
  13.         this.id = id;
  14.         this.name = name;
  15.         this.taskCode = taskCode;
  16.     }
  18.     public int getId() {
  19.         return id;
  20.     }
  22.     public void setId(int id) {
  23.         this.id = id;
  24.     }
  26.     public String getName() {
  27.         return name;
  28.     }
  30.     public void setName(String name) {
  31.         this.name = name;
  32.     }
  34.     public String getTaskCode() {
  35.         return taskCode;
  36.     }
  38.     public void setTaskCode(String taskCode) {
  39.         this.taskCode = taskCode;
  40.     }
  41. }
  1. package demo7.MQ;
  3. import java.util.Random;
  4. import java.util.concurrent.BlockingQueue;
  5. import java.util.concurrent.TimeUnit;
  6. import java.util.concurrent.atomic.AtomicInteger;
  8. public class Provider implements  Runnable {
  10.     //共享缓存区
  11.     private BlockingQueue<QueueData> queue;
  13.     //多线程释放启动？
  14.     private volatile boolean isRunning = true;
  16.     //ID生成器
  17.     private static AtomicInteger count = new AtomicInteger();
  19.     //生产随机对象
  20.     private static Random random = new Random();
  23.     public Provider(BlockingQueue<QueueData> queue) {
  24.         this.queue = queue;
  25.     }
  27.     @Override
  28.     public void run() {
  29.         while (isRunning){
  30.             try {
  31.                 //随机休眠 - 1000 表示读取数据、生产数据的耗时
  32.                 Thread.sleep(random.nextInt(1000));
  33.                 //incrementAndGet 进行累加
  34.                 int id = count.incrementAndGet();
  35.                 QueueData queueData = new QueueData(id,"任务"+String.valueOf(id),String.valueOf(id).hashCode()+"");
  36.                 System.err.println("线程："+Thread.currentThread().getName()+"\t生产task："+queueData.getName()+"\t"+queueData.getId());
  37.                 if (!queue.offer(queueData,2, TimeUnit.SECONDS)){
  38.                     System.err.println("!!!!!!!!!生产数据失败 error");
  39.                 }
  40.             } catch (InterruptedException e) {
  41.                 e.printStackTrace();
  42.             }
  43.         }
  46.     }
  48.     public  void stop(){
  49.         this.isRunning=false;
  50.     }
  51. }
  1. package demo7.MQ;
  3. import java.util.Random;
  4. import java.util.WeakHashMap;
  5. import java.util.concurrent.BlockingQueue;
  7. public class Consumer implements  Runnable{
  9.     private BlockingQueue<QueueData> queue;
  11.     public Consumer(BlockingQueue<QueueData> queue) {
  12.         this.queue = queue;
  13.     }
  15.     private  static Random random = new Random();
  17.     @Override
  18.     public void run() {
  19.         while (true){
  20.             try {
  21.                 //take：无阻塞
  22.                 QueueData queueData = this.queue.take();
  23.                 Thread.sleep(random.nextInt(1000));
  24.                 System.err.println("线程："+Thread.currentThread().getName()+"\t消费task->："+queueData.getName()+"\t"+queueData.getId());
  25.             } catch (InterruptedException e) {
  26.                 e.printStackTrace();
  27.             }
  28.         }
  29.     }
  30. }
  1. package demo7.MQ;
  3. import java.util.concurrent.BlockingQueue;
  4. import java.util.concurrent.ExecutorService;
  5. import java.util.concurrent.Executors;
  6. import java.util.concurrent.LinkedBlockingQueue;
  8. public class MainMQ {
  9.     /**
  10.      * 生产者、消费者(多线程模式)
  11.      * 1.生产、消费：通常由2类线程，即若干了生产者的线程、若干个消费者的线程、
  12.      * 2.生产者线程负责提交用户请求、消费者线程负责处理生产者提交的任务请求
  13.      * 3.生产者、消费者之间通过共享内存缓存进行通信
  14.      */
  16.     public static void main(String[] args) {
  18.         //1.内存缓存区
  19.         BlockingQueue<QueueData> queueData = new LinkedBlockingQueue<QueueData>();
  21.         //2.生产者
  22.         Provider p1 = new Provider(queueData);
  23.         Provider p2 = new Provider(queueData);
  24.         Provider p3 = new Provider(queueData);
  26.         //3.消费者
  27.         Consumer c1 = new Consumer(queueData);
  28.         Consumer c2 = new Consumer(queueData);
  29.         Consumer c3 = new Consumer(queueData);
  32.         //创建【线程池】运行，可以创建n个线程，没有任务的时候不创建线程，空闲线程存活时间为60s(默认)
  33.         ExecutorService executorService = Executors.newCachedThreadPool();
  34.         executorService.execute(p1);
  35.         executorService.execute(p2);
  36.         executorService.execute(p3);
  37.         executorService.execute(c1);
  38.         executorService.execute(c2);
  39.         executorService.execute(c3);
  41.         try {
  42.             Thread.sleep(2000);
  43.         } catch (InterruptedException e) {
  44.             e.printStackTrace();
  45.         }
  46.         p1.stop();
  47.         p2.stop();
  48.         p3.stop();
  50.         try {
  51.             Thread.sleep(1000);
  52.         } catch (InterruptedException e) {
  53.             e.printStackTrace();
  54.         }
  55.         executorService.shutdown();
  56.     }
  57. }



null


18.并发类容器MQ


联系
我们